siftlode/backend/tests/test_playlist_merge.py

"""S7 — two-way (3-way) YouTube playlist merge.

The load-bearing property: an item added (or deleted) on EITHER side survives the sync — the old
push was one-directional (local wins, deletes everything not local from YouTube), so YouTube-side
adds were lost. Pure-merge classification runs with no DB; the apply path uses the `db` lane with a
fake YouTube client so both the local rows and the simulated YouTube state can be asserted.
"""

from datetime import datetime, timezone

import pytest

from app.models import Channel, Playlist, PlaylistItem, Video
from app.sync import playlists as pl_mod
from app.sync.playlists import merge_plan, merge_playlist, parse_baseline

# --- pure merge classification (no DB) --------------------------------------------------------


class TestMergePlan:
    def test_no_baseline_unions_losslessly(self):
        # First-ever merge: can't tell an add from a delete, so keep everything, delete nothing.
        mp = merge_plan(None, local=["a", "b"], youtube=["b", "c"])
        assert mp.add_to_yt == ["a"]  # local-only pushed out
        assert mp.add_to_local == ["c"]  # youtube-only pulled in
        assert mp.remove_from_yt == [] and mp.remove_from_local == []
        assert mp.final == ["a", "b", "c"]  # local order, youtube-only appended
        assert mp.baseline_known is False

    def test_adds_on_both_sides_both_survive(self):
        # baseline b,c ; local added 'a' ; youtube added 'd' — nothing deleted.
        mp = merge_plan(["b", "c"], local=["a", "b", "c"], youtube=["b", "c", "d"])
        assert mp.add_to_yt == ["a"]
        assert mp.add_to_local == ["d"]
        assert mp.remove_from_yt == [] and mp.remove_from_local == []
        assert mp.final == ["a", "b", "c", "d"]

    def test_genuine_local_delete_propagates_to_youtube(self):
        # baseline a,b,c ; local removed 'b' ; youtube unchanged -> remove b from youtube.
        mp = merge_plan(["a", "b", "c"], local=["a", "c"], youtube=["a", "b", "c"])
        assert mp.remove_from_yt == ["b"]
        assert mp.add_to_yt == [] and mp.add_to_local == []
        assert mp.final == ["a", "c"]

    def test_genuine_youtube_delete_propagates_to_local(self):
        # baseline a,b,c ; youtube removed 'b' ; local unchanged -> remove b from local.
        mp = merge_plan(["a", "b", "c"], local=["a", "b", "c"], youtube=["a", "c"])
        assert mp.remove_from_local == ["b"]
        assert mp.final == ["a", "c"]

    def test_delete_on_both_sides_is_gone_not_a_conflict(self):
        mp = merge_plan(["a", "b", "c"], local=["a", "c"], youtube=["a", "c"])
        assert mp.remove_from_yt == [] and mp.remove_from_local == []
        assert mp.final == ["a", "c"]

    def test_a_local_delete_beats_a_stale_youtube_copy(self):
        # 'b' was synced, deleted locally, still present on youtube -> it must be removed there,
        # NOT resurrected into local as a youtube-add.
        mp = merge_plan(["a", "b"], local=["a"], youtube=["a", "b"])
        assert mp.add_to_local == []
        assert mp.remove_from_yt == ["b"]
        assert "b" not in mp.final


# --- apply path (db lane + fake YouTube) ------------------------------------------------------


class FakeYouTube:
    """Enough of YouTubeClient for merge_playlist: an in-memory playlist of (item_id, video_id)
    and the write ops that mutate it. Every catalog video is pre-seeded so _ensure_videos never
    needs get_videos."""

    def __init__(self, *, title: str, items: list[str]):
        self.title = title
        self._seq = 0
        self.items = [(self._new_id(), v) for v in items]

    def _new_id(self) -> str:
        self._seq += 1
        return f"it{self._seq}"

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def create_playlist(self, title, description="", privacy="private"):
        self.title = title
        return "PLnew"

    def get_playlist_snippet(self, pid):
        return {"title": self.title}

    def iter_playlist_items_with_ids(self, pid):
        return [{"item_id": iid, "video_id": v} for iid, v in self.items]

    def update_playlist_title(self, pid, title):
        self.title = title

    def add_playlist_item(self, pid, vid, position=None):
        iid = self._new_id()
        self.items.append((iid, vid))
        return iid

    def delete_playlist_item(self, item_id):
        self.items = [(iid, v) for iid, v in self.items if iid != item_id]

    def move_playlist_item(self, item_id, pid, vid, position):
        row = next((r for r in self.items if r[0] == item_id), None)
        if row is not None:
            self.items.remove(row)
            self.items.insert(position, row)

    @property
    def video_ids(self) -> list[str]:
        return [v for _, v in self.items]


def _seed_videos(db, ids):
    db.add(Channel(id="UCtest", title="Test"))
    now = datetime.now(timezone.utc)
    for v in ids:
        db.add(Video(id=v, channel_id="UCtest", published_at=now))
    db.commit()


def _linked_playlist(db, user, *, local_ids, baseline):
    pl = Playlist(
        user_id=user.id,
        name="Mix",
        source="local",
        yt_playlist_id="PLyt",
        synced_video_ids=None if baseline is None else pl_mod._dump_ids(baseline),
    )
    db.add(pl)
    db.flush()
    for pos, v in enumerate(local_ids):
        db.add(PlaylistItem(playlist_id=pl.id, video_id=v, position=pos))
    db.commit()
    return pl


def _local_ids(db, pl):
    return pl_mod._local_video_ids(db, pl)


@pytest.fixture
def _fake_yt(monkeypatch):
    holder = {}

    def install(fake):
        holder["fake"] = fake
        monkeypatch.setattr(pl_mod, "YouTubeClient", lambda db, user: fake)
        return fake

    return install


def test_adds_on_both_sides_survive_on_both_sides(db, make_user, _fake_yt):
    user = make_user()
    _seed_videos(db, ["a", "b", "c", "d"])
    # baseline b,c ; local added 'a' ; youtube added 'd'
    pl = _linked_playlist(db, user, local_ids=["a", "b", "c"], baseline=["b", "c"])
    fake = _fake_yt(FakeYouTube(title="Mix", items=["b", "c", "d"]))

    r = merge_playlist(db, user, pl)

    assert r["failures"] == []
    assert set(_local_ids(db, pl)) == {"a", "b", "c", "d"}, "youtube-add 'd' pulled into Siftlode"
    assert set(fake.video_ids) == {"a", "b", "c", "d"}, "local-add 'a' pushed to YouTube"
    assert r["inserted"] == 1 and r["pulled"] == 1
    # Baseline advanced to the merged set so a re-sync is a no-op.
    assert set(parse_baseline(pl)) == {"a", "b", "c", "d"}


def test_a_genuine_delete_propagates_without_collateral_loss(db, make_user, _fake_yt):
    user = make_user()
    _seed_videos(db, ["a", "b", "c", "d"])
    # baseline a,b,c ; local removed 'b' ; youtube added 'd' (a real add, must NOT be collateral)
    pl = _linked_playlist(db, user, local_ids=["a", "c"], baseline=["a", "b", "c"])
    fake = _fake_yt(FakeYouTube(title="Mix", items=["a", "b", "c", "d"]))

    r = merge_playlist(db, user, pl)

    assert r["failures"] == []
    assert set(fake.video_ids) == {"a", "c", "d"}, "'b' removed on YouTube, 'd' kept"
    assert set(_local_ids(db, pl)) == {"a", "c", "d"}, "'d' pulled in, 'b' stays deleted"
    assert r["deleted"] == 1


def test_first_merge_with_no_baseline_never_deletes(db, make_user, _fake_yt):
    user = make_user()
    _seed_videos(db, ["a", "b", "c"])
    pl = _linked_playlist(db, user, local_ids=["a", "b"], baseline=None)
    fake = _fake_yt(FakeYouTube(title="Mix", items=["b", "c"]))

    r = merge_playlist(db, user, pl)

    assert r["deleted"] == 0 and r["removed_local"] == 0
    assert set(_local_ids(db, pl)) == {"a", "b", "c"}
    assert set(fake.video_ids) == {"a", "b", "c"}
    assert parse_baseline(pl) is not None, "a baseline is recorded after the first union"


def test_create_path_appends_and_records_baseline(db, make_user, _fake_yt):
    user = make_user()
    _seed_videos(db, ["a", "b"])
    pl = Playlist(user_id=user.id, name="Fresh", source="local")
    db.add(pl)
    db.flush()
    for pos, v in enumerate(["a", "b"]):
        db.add(PlaylistItem(playlist_id=pl.id, video_id=v, position=pos))
    db.commit()
    _fake_yt(FakeYouTube(title="Fresh", items=[]))

    r = merge_playlist(db, user, pl)

    assert r["created"] == 1 and r["inserted"] == 2
    assert pl.yt_playlist_id  # created and linked
    assert parse_baseline(pl) == ["a", "b"]