FabledCurator/tests/test_source_service.py

import pytest
from sqlalchemy import select

from backend.app.models import Artist, Source
from backend.app.services.source_service import (
    KNOWN_PLATFORMS,
    ArtistNotFoundError,
    DuplicateSourceError,
    EmptyUrlError,
    InvalidConfigError,
    SourceService,
    UnknownPlatformError,
)

pytestmark = pytest.mark.integration


async def _artist(db, name="Alice"):
    a = Artist(name=name, slug=name.lower())
    db.add(a)
    await db.flush()
    return a


@pytest.mark.asyncio
async def test_known_platforms_is_gs_six(db):
    assert KNOWN_PLATFORMS == frozenset({
        "patreon", "subscribestar", "hentaifoundry",
        "discord", "pixiv", "deviantart",
    })
    assert "fanbox" not in KNOWN_PLATFORMS


@pytest.mark.asyncio
async def test_create_flips_is_subscription_on_first_source(db):
    artist = await _artist(db)
    svc = SourceService(db)
    rec = await svc.create(
        artist_id=artist.id, platform="patreon", url="https://patreon.com/alice",
    )
    assert rec.id is not None
    is_sub = (await db.execute(
        select(Artist.is_subscription).where(Artist.id == artist.id)
    )).scalar_one()
    assert is_sub is True


@pytest.mark.asyncio
async def test_delete_last_source_flips_is_subscription_off(db):
    artist = await _artist(db)
    svc = SourceService(db)
    rec = await svc.create(
        artist_id=artist.id, platform="patreon", url="https://patreon.com/alice",
    )
    await svc.delete(rec.id)
    is_sub = (await db.execute(
        select(Artist.is_subscription).where(Artist.id == artist.id)
    )).scalar_one()
    assert is_sub is False


@pytest.mark.asyncio
async def test_create_rejects_unknown_platform(db):
    artist = await _artist(db)
    svc = SourceService(db)
    with pytest.raises(UnknownPlatformError):
        await svc.create(
            artist_id=artist.id, platform="myspace", url="https://m/x",
        )


@pytest.mark.asyncio
async def test_create_rejects_non_dict_config(db):
    artist = await _artist(db)
    svc = SourceService(db)
    with pytest.raises(InvalidConfigError):
        await svc.create(
            artist_id=artist.id, platform="patreon",
            url="https://patreon.com/alice", config_overrides=[1, 2, 3],
        )


@pytest.mark.asyncio
async def test_create_rejects_empty_url(db):
    artist = await _artist(db)
    svc = SourceService(db)
    with pytest.raises(EmptyUrlError):
        await svc.create(artist_id=artist.id, platform="patreon", url="   ")


@pytest.mark.asyncio
async def test_create_rejects_unknown_artist(db):
    svc = SourceService(db)
    with pytest.raises(ArtistNotFoundError):
        await svc.create(artist_id=99999, platform="patreon", url="https://x/y")


@pytest.mark.asyncio
async def test_create_duplicate_raises_with_existing_id(db):
    artist = await _artist(db)
    svc = SourceService(db)
    first = await svc.create(
        artist_id=artist.id, platform="patreon", url="https://patreon.com/alice",
    )
    with pytest.raises(DuplicateSourceError) as exc:
        await svc.create(
            artist_id=artist.id, platform="patreon",
            url="https://patreon.com/alice",
        )
    assert exc.value.existing_id == first.id


@pytest.mark.asyncio
async def test_list_filters_by_artist(db):
    a = await _artist(db, "Alice")
    b = await _artist(db, "Bob")
    svc = SourceService(db)
    await svc.create(artist_id=a.id, platform="patreon", url="https://patreon.com/a")
    await svc.create(artist_id=b.id, platform="patreon", url="https://patreon.com/b")
    only_a = await svc.list(artist_id=a.id)
    assert [s.artist_id for s in only_a] == [a.id]
    all_rows = await svc.list()
    assert len(all_rows) == 2


@pytest.mark.asyncio
async def test_update_changes_fields(db):
    artist = await _artist(db)
    svc = SourceService(db)
    rec = await svc.create(
        artist_id=artist.id, platform="patreon", url="https://patreon.com/a",
    )
    updated = await svc.update(rec.id, enabled=False, config_overrides={"videos": False})
    assert updated.enabled is False
    assert updated.config_overrides == {"videos": False}


@pytest.mark.asyncio
async def test_list_hides_sidecar_synthetic_anchors(db):
    """Filesystem-import synthetic Sources (url='sidecar:<platform>:<slug>',
    enabled=False — see importer._source_for_sidecar) used to leak into the
    Subscriptions UI as phantom subscriptions because list() didn't filter
    them. They aren't pollable feeds; hide by default."""
    artist = await _artist(db, "Alice")
    real = Source(
        artist_id=artist.id, platform="patreon",
        url="https://patreon.com/alice", enabled=True, config_overrides={},
    )
    synthetic = Source(
        artist_id=artist.id, platform="patreon",
        url="sidecar:patreon:alice", enabled=False, config_overrides={},
    )
    db.add_all([real, synthetic])
    await db.commit()

    svc = SourceService(db)
    visible = await svc.list()
    visible_urls = {s.url for s in visible}
    assert "https://patreon.com/alice" in visible_urls
    assert "sidecar:patreon:alice" not in visible_urls

    # Same filter applies to the artist-scoped list path (the artist detail
    # page hits /api/sources?artist_id=N).
    artist_scoped = await svc.list(artist_id=artist.id)
    assert {s.url for s in artist_scoped} == {"https://patreon.com/alice"}

    # include_synthetic=True opts back in for admin tooling.
    everything = await svc.list(include_synthetic=True)
    assert {s.url for s in everything} >= {
        "https://patreon.com/alice", "sidecar:patreon:alice",
    }