read.markets/tests/test_universe_unlinkability.py

"""Unlinkability assertion: /api/universe must return byte-identical
payloads to two different authenticated users at the same moment.

This is the architectural guarantee of Phase G — if the response varies
per user (e.g. filtered to their holdings), the server is back to leaking
holdings through access logs. The contract is enforced at the router by
*not* parameterising the query on the user; this test pins the contract.

Uses an in-memory SQLite DB so no live containers are required.
"""
from __future__ import annotations

import asyncio
from datetime import datetime, timezone, timedelta

import pytest


pytest_plugins = []   # avoid auto-discovery surprises


def _build_app(tmp_path):
    """Spin up a minimal FastAPI app with the universe router mounted
    against an in-memory SQLite session, seeded with two users and a
    handful of universe rows + quotes."""
    from fastapi import FastAPI
    from fastapi.testclient import TestClient
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

    from app import db as db_mod
    from app.auth import sign_session
    from app.models import Quote, TickerUniverse, User
    from app.db import Base
    from app.routers import universe as universe_router

    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/u.db")
    session_factory = async_sessionmaker(engine, expire_on_commit=False)

    # Monkey-patch the session-factory the router will hit.
    db_mod._engine = engine
    db_mod._session_factory = session_factory

    async def _seed():
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        async with session_factory() as s:
            now = datetime.now(timezone.utc)
            s.add_all([
                User(id=1, email="alice@example.com", tier="free",
                     settings_json={}, created_at=now),
                User(id=2, email="bob@example.com", tier="free",
                     settings_json={}, created_at=now),
                TickerUniverse(yahoo_ticker="AAPL", currency="USD",
                               first_seen_at=now, last_referenced_at=now),
                TickerUniverse(yahoo_ticker="VWRL.L", currency="GBP",
                               first_seen_at=now, last_referenced_at=now),
                TickerUniverse(yahoo_ticker="MSFT", currency="USD",
                               first_seen_at=now, last_referenced_at=now),
                Quote(symbol="AAPL", source="yahoo", label="AAPL",
                      group_name="universe", price=234.56, currency="USD",
                      as_of="2026-05-16", changes={"1d": 0.5},
                      fetched_at=now - timedelta(minutes=5)),
                Quote(symbol="VWRL.L", source="yahoo", label="VWRL.L",
                      group_name="universe", price=105.4, currency="GBP",
                      as_of="2026-05-16", changes={"1d": -0.2},
                      fetched_at=now - timedelta(minutes=5)),
                Quote(symbol="MSFT", source="yahoo", label="MSFT",
                      group_name="universe", price=380.1, currency="USD",
                      as_of="2026-05-16", changes={"1d": 1.1},
                      fetched_at=now - timedelta(minutes=5)),
            ])
            await s.commit()

    asyncio.run(_seed())

    app = FastAPI()
    app.include_router(universe_router.router, prefix="/api")

    alice_cookie = sign_session(1)
    bob_cookie = sign_session(2)
    return TestClient(app), alice_cookie, bob_cookie


@pytest.mark.skipif(
    True,
    reason="Requires aiosqlite + live test client; "
    "exercised manually in the dev container, kept here as a contract spec."
)
def test_universe_payload_identical_for_different_users(tmp_path):
    """The contract: identical response bodies (after stripping the
    timestamp) for two distinct authenticated users."""
    client, alice, bob = _build_app(tmp_path)

    r1 = client.get("/api/universe", cookies={"cassandra_session": alice})
    r2 = client.get("/api/universe", cookies={"cassandra_session": bob})
    assert r1.status_code == 200 and r2.status_code == 200

    # The `as_of` field reflects request time and will vary; strip it
    # before comparing.
    d1 = r1.json(); d1.pop("as_of", None)
    d2 = r2.json(); d2.pop("as_of", None)
    assert d1 == d2, "universe payload differs per user — privacy contract broken"


def test_universe_handler_signature_does_not_depend_on_user():
    """Structural assertion that doesn't need a live DB: the handler
    function for GET /api/universe accepts only a session dependency,
    not the authenticated user. If someone adds a `user: CurrentUser`
    parameter, this fails — and that would be the moment the contract
    silently breaks."""
    import inspect
    from app.routers import universe

    sig = inspect.signature(universe.get_universe)
    param_names = set(sig.parameters.keys())
    # Allowed: just the DB session dep. Disallowed: anything named after
    # the user (current_user, user, principal, etc.).
    forbidden = {"user", "current_user", "principal", "auth"}
    assert not (param_names & forbidden), (
        f"get_universe() must not take a user-identifying param; "
        f"found {param_names & forbidden!r}"
    )