siftlode/backend/tests/test_owner_scoping.py

"""The core's owner filters, each held by a second account whose rows must not count, show or change.

Every one of these reads or writes a user-owned table through `userscope.owned_by`; the fence holds
the form (no inline `Model.user_id == …`), these hold that the filter is there at all. Each test gives
a second account the same kind of rows, so a filter that went missing answers with theirs. DB lane.
"""

from datetime import datetime, timedelta, timezone

from sqlalchemy import func, select

from app import debugsession, notifications, quota
from app.catalog.models import Channel, Video
from app.downloads.models import DownloadJob, DownloadProfile
from app.models import DebugScreenshot, DebugSession, Notification, QuotaEvent
from app.reader.models import Playlist, PlaylistItem, VideoState
from app.routes import tokens as tokens_routes
from app.routes.auth import RESET_TTL, _issue_token


def _notify(db, user, n=1, *, read=False):
    for i in range(n):
        db.add(Notification(user_id=user.id, type="t", title=f"n{i}", read=read))
    db.commit()


def _visible(db, user) -> int:
    return db.scalar(
        select(func.count())
        .select_from(Notification)
        .where(Notification.user_id == user.id, Notification.dismissed.is_(False))
    )


def test_the_notification_list_shows_only_mine(db, make_user, as_user):
    me, other = make_user(), make_user()
    _notify(db, me)
    _notify(db, other, 2)
    body = as_user(me).get("/api/me/notifications").json()
    assert (body["total"], len(body["items"])) == (1, 1)


def test_the_unread_count_counts_only_mine(db, make_user, as_user):
    me, other = make_user(), make_user()
    _notify(db, me)
    _notify(db, other, 2)
    assert as_user(me).get("/api/me/notifications/unread_count").json() == {"count": 1}


def test_reading_all_reads_only_mine(db, make_user, as_user):
    me, other = make_user(), make_user()
    _notify(db, me)
    _notify(db, other, 2)
    assert as_user(me).post("/api/me/notifications/read_all").status_code == 200
    assert as_user(other).get("/api/me/notifications/unread_count").json() == {"count": 2}


def test_clearing_all_dismisses_only_mine(db, make_user, as_user):
    me, other = make_user(), make_user()
    _notify(db, me)
    _notify(db, other, 2)
    assert as_user(me).post("/api/me/notifications/clear").status_code == 200
    db.expire_all()
    assert (_visible(db, me), _visible(db, other)) == (0, 2)


def test_trimming_read_notifications_trims_only_mine(db, make_user):
    me, other = make_user(), make_user()
    _notify(db, me, 3, read=True)
    _notify(db, other, 3, read=True)
    assert notifications.trim_read(db, me.id, cap=1) == 2
    db.commit()
    assert _visible(db, other) == 3


def test_todays_actions_count_only_mine(db, make_user):
    me, other = make_user(), make_user()
    quota.claim_action(me.id, quota.QuotaAction.VIDEOS_SEARCH, 5)
    quota.claim_action(other.id, quota.QuotaAction.VIDEOS_SEARCH, 5)
    quota.claim_action(other.id, quota.QuotaAction.VIDEOS_SEARCH, 5)
    assert quota.actions_today(db, me.id, quota.QuotaAction.VIDEOS_SEARCH) == 1


def test_my_usage_sums_only_mine(db, make_user, as_user):
    me, other = make_user(), make_user()
    db.add_all(
        [
            QuotaEvent(user_id=me.id, action="videos.list", units=5),
            QuotaEvent(user_id=other.id, action="videos.list", units=7),
        ]
    )
    db.commit()
    body = as_user(me).get("/api/quota/my-usage").json()
    assert (body["today"], body["all_time"], body["by_action"]) == (5, 5, {"videos.list": 5})


def test_the_token_ceiling_counts_only_mine(db, make_user, as_user, monkeypatch):
    monkeypatch.setattr(tokens_routes, "MAX_TOKENS", 1)
    me, other = make_user(), make_user()
    first = as_user(other).post("/api/tokens", json={"name": "theirs", "preset": "hub"})
    assert first.status_code == 200, first.text
    r = as_user(me).post("/api/tokens", json={"name": "mine", "preset": "hub"})
    assert r.status_code == 200, r.text


def test_a_reset_burns_only_my_other_links(client, db, make_user):
    me, other = make_user(), make_user()
    mine = _issue_token(db, me, "reset", RESET_TTL)
    theirs = _issue_token(db, other, "reset", RESET_TTL)
    db.commit()
    r = client.post(
        "/auth/password-reset/confirm", json={"token": mine, "password": "a brand new passphrase"}
    )
    assert r.json() == {"ok": True}
    assert client.post("/auth/password-reset/check", json={"token": theirs}).json() == {"ok": True}


def test_screenshot_storage_counts_only_mine(db, make_user):
    me, other = make_user(), make_user()
    now = datetime.now(timezone.utc)
    for i, (user, size) in enumerate([(me, 10), (other, 1000)]):
        db.add(
            DebugSession(
                id=f"s{i}",
                user_id=user.id,
                level="DEBUG",
                started_at=now,
                until=now + timedelta(hours=1),
            )
        )
        db.flush()
        db.add(
            DebugScreenshot(
                session_id=f"s{i}", created_at=now, mime="image/png", size=size, data=b"x" * size
            )
        )
    db.commit()
    assert debugsession.stored_screenshot_bytes(db, me.id) == 10


# --- the units' owner filters, through their routes --------------------------------------------
# The same rule one layer up: a second account acting on the SAME video, list or profile id must
# reach only its own row. Each test holds the first account's row unchanged afterwards.


def _videos(db, *ids):
    db.add(Channel(id="UCo", title="O"))
    db.flush()
    for vid in ids:
        db.add(
            Video(
                id=vid,
                channel_id="UCo",
                published_at=datetime.now(timezone.utc),
                duration_seconds=3600,
            )
        )
    db.commit()


def _video_state(db, user, vid="v1"):
    db.expire_all()
    row = db.execute(
        select(VideoState).where(VideoState.user_id == user.id, VideoState.video_id == vid)
    ).scalar_one_or_none()
    return None if row is None else (row.status, row.position_seconds)


def test_unmarking_a_video_leaves_another_users_state(db, make_user, as_user):
    me, other = make_user(), make_user()
    _videos(db, "v1")
    db.add(VideoState(user_id=other.id, video_id="v1", status="watched"))
    db.commit()
    r = as_user(me).post("/api/videos/v1/state", json={"status": "new"})
    assert r.status_code == 200, r.text
    assert _video_state(db, other) == ("watched", 0)


def test_clearing_a_video_leaves_another_users_state(db, make_user, as_user):
    me, other = make_user(), make_user()
    _videos(db, "v1")
    db.add(VideoState(user_id=other.id, video_id="v1", status="hidden", position_seconds=90))
    db.commit()
    assert as_user(me).delete("/api/videos/v1/progress").status_code == 200
    assert _video_state(db, other) == ("hidden", 90)


def test_a_resume_point_is_set_and_cleared_on_my_own_row_only(db, make_user, as_user):
    me, other = make_user(), make_user()
    _videos(db, "v1")
    db.add(VideoState(user_id=other.id, video_id="v1", status="hidden", position_seconds=90))
    db.commit()
    client = as_user(me)
    body = {"position_seconds": 600, "duration_seconds": 3600}
    assert client.post("/api/videos/v1/progress", json=body).status_code == 200
    assert (_video_state(db, me), _video_state(db, other)) == (("new", 600), ("hidden", 90))
    body = {"position_seconds": 0, "duration_seconds": 3600}
    assert client.post("/api/videos/v1/progress", json=body).status_code == 200
    assert (_video_state(db, me), _video_state(db, other)) == (None, ("hidden", 90))


def _watch_later_ids(db, user):
    db.expire_all()
    return sorted(
        db.execute(
            select(PlaylistItem.video_id)
            .join(Playlist, Playlist.id == PlaylistItem.playlist_id)
            .where(Playlist.user_id == user.id, Playlist.kind == "watch_later")
        ).scalars()
    )


def test_watch_later_adds_and_removes_use_my_own_list(db, make_user, as_user):
    me, other = make_user(), make_user()
    _videos(db, "v1", "v2")
    theirs = as_user(other).post("/api/playlists/watch-later", json={"video_id": "v1"}).json()
    client = as_user(me)
    mine = client.post("/api/playlists/watch-later", json={"video_id": "v2"}).json()
    assert mine["playlist_id"] != theirs["playlist_id"]
    # Their saved video is not on my list, so removing it from mine leaves theirs alone.
    assert client.delete("/api/playlists/watch-later/v1").status_code == 200
    assert (_watch_later_ids(db, me), _watch_later_ids(db, other)) == (["v2"], ["v1"])


def test_another_users_private_profile_is_refused_like_a_missing_one(db, make_user, as_user):
    me, other = make_user(), make_user()
    theirs = DownloadProfile(user_id=other.id, name="Theirs", spec={"format": "best"})
    builtin = DownloadProfile(user_id=None, name="Best", spec={"format": "best"}, is_builtin=True)
    db.add_all([theirs, builtin])
    db.commit()
    client = as_user(me)

    def enqueue(profile_id):
        return client.post(
            "/api/downloads", json={"source": "dQw4w9WgXcQ", "profile_id": profile_id}
        )

    missing = enqueue(theirs.id + builtin.id + 1)
    refused = enqueue(theirs.id)
    assert (refused.status_code, refused.json()) == (404, missing.json())
    assert refused.json()["detail"] == {"code": "not_found", "reason": "profile"}
    db.expire_all()
    assert db.execute(select(DownloadJob)).first() is None
    # A built-in is everyone's to use.
    r = enqueue(builtin.id)
    assert r.status_code == 200, r.text
    db.expire_all()
    assert db.execute(select(DownloadJob.profile_id)).scalars().all() == [builtin.id]