123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""FC-5: gs_ingest unit tests."""
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models import Artist, Credential, Source
|
|
from backend.app.services.credential_crypto import CredentialCrypto
|
|
from backend.app.services.migrators import gs_ingest
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _gs_export(subscriptions=None, credentials=None) -> dict:
|
|
return {
|
|
"source_app": "gallerysubscriber",
|
|
"schema_version": 1,
|
|
"exported_at": datetime.now(UTC).isoformat(),
|
|
"subscriptions": subscriptions or [],
|
|
"credentials": credentials or [],
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subscription_creates_artist_with_slug(db):
|
|
data = _gs_export(subscriptions=[
|
|
{"name": "Alice Author", "enabled": True, "metadata": {}, "sources": []},
|
|
])
|
|
counts = await gs_ingest.migrate_async(db, data=data, dry_run=False)
|
|
assert counts["rows_inserted"] >= 1
|
|
|
|
artist = (await db.execute(
|
|
select(Artist).where(Artist.name == "Alice Author")
|
|
)).scalar_one()
|
|
assert artist.slug == "alice-author"
|
|
assert artist.is_subscription is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nested_source_attached_to_artist(db):
|
|
data = _gs_export(subscriptions=[{
|
|
"name": "bob-author",
|
|
"enabled": True,
|
|
"metadata": {},
|
|
"sources": [{
|
|
"platform": "patreon",
|
|
"url": "https://patreon.com/bob",
|
|
"enabled": True,
|
|
"check_interval": 28800,
|
|
"metadata": {"k": "v"},
|
|
}],
|
|
}])
|
|
await gs_ingest.migrate_async(db, data=data, dry_run=False)
|
|
|
|
src = (await db.execute(
|
|
select(Source).where(Source.url == "https://patreon.com/bob")
|
|
)).scalar_one()
|
|
artist = (await db.execute(
|
|
select(Artist).where(Artist.id == src.artist_id)
|
|
)).scalar_one()
|
|
assert artist.name == "bob-author"
|
|
assert src.platform == "patreon"
|
|
assert src.check_interval_override == 28800
|
|
assert src.config_overrides == {"k": "v"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_credential_re_encrypted_with_fc_key(db, tmp_path):
|
|
fc_key_path = tmp_path / "credential_key.b64"
|
|
fc_crypto = CredentialCrypto(fc_key_path)
|
|
|
|
data = _gs_export(credentials=[{
|
|
"platform": "patreon",
|
|
"credential_type": "cookies",
|
|
"plaintext": "session_cookie=abcdef",
|
|
"expires_at": None,
|
|
}])
|
|
await gs_ingest.migrate_async(db, data=data, fc_crypto=fc_crypto, dry_run=False)
|
|
|
|
cred = (await db.execute(
|
|
select(Credential).where(Credential.platform == "patreon")
|
|
)).scalar_one()
|
|
assert fc_crypto.decrypt(cred.encrypted_blob) == "session_cookie=abcdef"
|
|
assert cred.credential_type == "cookies"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dry_run_makes_no_changes(db):
|
|
data = _gs_export(subscriptions=[
|
|
{"name": "dave", "enabled": True, "metadata": {}, "sources": []},
|
|
])
|
|
counts = await gs_ingest.migrate_async(db, data=data, dry_run=True)
|
|
assert counts["rows_processed"] >= 1
|
|
|
|
after = (await db.execute(
|
|
select(Artist).where(Artist.name == "dave")
|
|
)).scalar_one_or_none()
|
|
assert after is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_idempotent_on_rerun(db):
|
|
data = _gs_export(subscriptions=[
|
|
{"name": "eve", "enabled": True, "metadata": {}, "sources": []},
|
|
])
|
|
await gs_ingest.migrate_async(db, data=data, dry_run=False)
|
|
counts2 = await gs_ingest.migrate_async(db, data=data, dry_run=False)
|
|
assert counts2["rows_skipped"] >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejects_wrong_source_app(db):
|
|
bad = {"source_app": "wrong", "schema_version": 1, "subscriptions": [], "credentials": []}
|
|
with pytest.raises(ValueError):
|
|
await gs_ingest.migrate_async(db, data=bad, dry_run=False)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejects_unsupported_schema_version(db):
|
|
bad = {"source_app": "gallerysubscriber", "schema_version": 99, "subscriptions": [], "credentials": []}
|
|
with pytest.raises(ValueError):
|
|
await gs_ingest.migrate_async(db, data=bad, dry_run=False)
|