"""Layer 6: Contract / parity tests. These tests assert the exact JSON response shapes produced by the Python API, ensuring they remain stable and compatible with any consumers that were previously served by the .NET AdminAPI. Two modes of operation ---------------------- 1. Local (CI) mode — runs against an in-process FastAPI app backed by SQLite. Tests that the Python API produces the correct *shape* and *field names*. This runs with no external services. 2. Side-by-side mode (pre-cutover) — run with: DOTNET_API_URL=http://dotnet-host/AdminAPI \ PYTHON_API_URL=http://python-host \ pytest tests/contract/test_contract.py -m parity In this mode the tests hit both real deployments and fail if the JSON responses diverge. Enable by setting DOTNET_API_URL in the environment. Field naming conventions ------------------------ The Python API uses snake_case internally (SQLModel) but the JSON responses must be compatible with any existing consumers. These tests document and enforce the exact field names for each response model. """ import os from datetime import datetime, timedelta, timezone import pytest import pytest_asyncio from httpx import ASGITransport, AsyncClient from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from sqlmodel import SQLModel from playlist_sync.api import app from playlist_sync.models.application import Application from playlist_sync.models.service_account import ServiceAccount # noqa: F401 from playlist_sync.models.sync_log import PlaylistSynchronizationLog # noqa: F401 from playlist_sync.models.sync_task import PlaylistSynchronization # noqa: F401 from playlist_sync.services.application_service import ApplicationService from playlist_sync.services.database import get_session TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:" DOTNET_API_URL = os.environ.get("DOTNET_API_URL", "") ADMIN_API_KEY = os.environ.get( "ADMIN_API_KEY", "" ) # Authorization header value for .NET AdminAPI # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest_asyncio.fixture(scope="module") async def contract_engine(): engine = create_async_engine( TEST_DATABASE_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) async with engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) yield engine async with engine.begin() as conn: await conn.run_sync(SQLModel.metadata.drop_all) await engine.dispose() @pytest_asyncio.fixture(scope="module") async def contract_client(contract_engine): Session = sessionmaker(contract_engine, class_=AsyncSession, expire_on_commit=False) async with Session() as session: session.add( Application( id=1, name="US", spotify_region_code="US", active=True, fallback_application=False, global_push_application=False, workout_market=False, ) ) await session.commit() ApplicationService.invalidate_cache() async def override_get_session(): async with Session() as s: yield s app.dependency_overrides[get_session] = override_get_session async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test" ) as ac: yield ac, Session app.dependency_overrides.clear() ApplicationService.invalidate_cache() # --------------------------------------------------------------------------- # PlaylistSynchronization response shape # --------------------------------------------------------------------------- class TestPlaylistSynchronizationShape: """Assert the exact JSON field set for PlaylistSynchronization responses.""" # Fields the .NET PlaylistSynchronization model exposes via the API EXPECTED_FIELDS = { "id", "applicationId", "fromPlaylistId", "fromServiceType", "fromMusicServiceId", "toPlaylistId", "toServiceAccountId", "title", "description", "active", "synchronizedTrackCount", "titleCopyMode", "descriptionCopyMode", "appendTrackList", "error", "createdAt", "lastUpdated", } async def test_create_response_has_all_required_fields(self, contract_client): """POST response must include all required PlaylistSynchronization fields.""" client, _ = contract_client resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:contract001", "to_playlist_id": "deezer:playlist:contract001", "to_service_account_id": 1, "title": "Contract Test", }, ) assert resp.status_code == 201 data = resp.json() for field in self.EXPECTED_FIELDS: assert field in data, f"Missing field: '{field}' in POST response" async def test_get_response_has_all_required_fields(self, contract_client): """GET response must include all required PlaylistSynchronization fields.""" client, _ = contract_client post_resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:contract002", "to_playlist_id": "deezer:playlist:contract002", "to_service_account_id": 1, }, ) sync_id = post_resp.json()["id"] resp = await client.get(f"/PlaylistSync/US/playlists/{sync_id}") assert resp.status_code == 200 data = resp.json() for field in self.EXPECTED_FIELDS: assert field in data, f"Missing field: '{field}' in GET response" async def test_list_response_elements_have_all_required_fields( self, contract_client ): """GET /playlists list elements must include all required fields.""" client, _ = contract_client resp = await client.get("/PlaylistSync/US/playlists") assert resp.status_code == 200 items = resp.json() assert len(items) > 0, "List should have at least one seeded item" for item in items: for field in self.EXPECTED_FIELDS: assert field in item, f"Missing field: '{field}' in list item" async def test_id_is_integer(self, contract_client): """Id field must be an integer, not a string.""" client, _ = contract_client resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:id-type", "to_playlist_id": "deezer:playlist:id-type", "to_service_account_id": 1, }, ) assert resp.status_code == 201 data = resp.json() assert isinstance(data["id"], int), f"Expected int id, got {type(data['id'])}" async def test_active_is_true_on_create(self, contract_client): """Newly created syncs must have active=True.""" client, _ = contract_client resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:active-check", "to_playlist_id": "deezer:playlist:active-check", "to_service_account_id": 1, }, ) assert resp.status_code == 201 assert resp.json()["active"] is True async def test_application_id_matches_market(self, contract_client): """application_id in response must match resolved Application.id for 'US'.""" client, _ = contract_client resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:app-id-check", "to_playlist_id": "deezer:playlist:app-id-check", "to_service_account_id": 1, }, ) assert resp.status_code == 201 assert resp.json()["applicationId"] == 1 async def test_update_response_reflects_changes(self, contract_client): """PUT response must reflect the updated fields immediately.""" client, _ = contract_client post_resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:update-check", "to_playlist_id": "deezer:playlist:update-check", "to_service_account_id": 1, "title": "Original Title", }, ) sync_id = post_resp.json()["id"] put_resp = await client.put( f"/PlaylistSync/US/playlists/{sync_id}", json={ "title": "Updated Title", }, ) assert put_resp.status_code == 200 assert put_resp.json()["title"] == "Updated Title" async def test_deactivate_sets_active_false(self, contract_client): """PUT with active=False must set active to False in response.""" client, _ = contract_client post_resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:deactivate", "to_playlist_id": "deezer:playlist:deactivate", "to_service_account_id": 1, }, ) sync_id = post_resp.json()["id"] put_resp = await client.put( f"/PlaylistSync/US/playlists/{sync_id}", json={ "active": False, }, ) assert put_resp.status_code == 200 assert put_resp.json()["active"] is False # --------------------------------------------------------------------------- # PlaylistSynchronizationLog response shape # --------------------------------------------------------------------------- class TestSyncLogShape: """Assert the exact JSON field set for PlaylistSynchronizationLog responses.""" EXPECTED_LOG_FIELDS = { "playlistSynchronizationId", "sourceTracks", "addedTracks", "deletedTracks", "error", "time", } async def test_log_list_returns_array(self, contract_client): """GET /log must return a JSON array.""" client, Session = contract_client post_resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:log-shape", "to_playlist_id": "deezer:playlist:log-shape", "to_service_account_id": 1, }, ) sync_id = post_resp.json()["id"] # Seed a log entry directly async with Session() as session: session.add( PlaylistSynchronizationLog( sync_id=sync_id, source_tracks=10, added_tracks=5, deleted_tracks=2, ) ) await session.commit() resp = await client.get(f"/PlaylistSync/US/playlists/{sync_id}/log") assert resp.status_code == 200 items = resp.json() assert isinstance(items, list) assert len(items) >= 1 async def test_log_entry_has_all_required_fields(self, contract_client): """Log entries must include all required PlaylistSynchronizationLog fields.""" client, Session = contract_client post_resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:log-fields", "to_playlist_id": "deezer:playlist:log-fields", "to_service_account_id": 1, }, ) sync_id = post_resp.json()["id"] async with Session() as session: session.add( PlaylistSynchronizationLog( sync_id=sync_id, source_tracks=3, added_tracks=1, deleted_tracks=0, ) ) await session.commit() resp = await client.get(f"/PlaylistSync/US/playlists/{sync_id}/log") assert resp.status_code == 200 items = resp.json() for field in self.EXPECTED_LOG_FIELDS: assert field in items[0], f"Missing field: '{field}' in log entry" async def test_log_pagination_with_limit(self, contract_client): """Limit parameter restricts the number of log entries returned.""" client, Session = contract_client post_resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "spotify:playlist:log-paginate", "to_playlist_id": "deezer:playlist:log-paginate", "to_service_account_id": 1, }, ) sync_id = post_resp.json()["id"] async with Session() as session: now = datetime.now(timezone.utc) for i in range(5): session.add( PlaylistSynchronizationLog( sync_id=sync_id, source_tracks=i, added_tracks=0, deleted_tracks=0, time=now - timedelta(seconds=i), ) ) await session.commit() resp = await client.get(f"/PlaylistSync/US/playlists/{sync_id}/log?limit=2") assert resp.status_code == 200 assert len(resp.json()) <= 2 # --------------------------------------------------------------------------- # HTTP status code contract # --------------------------------------------------------------------------- class TestStatusCodeContract: """Document and enforce the expected HTTP status codes for every operation.""" async def test_create_returns_201(self, contract_client): client, _ = contract_client resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "src-sc-001", "to_playlist_id": "tgt-sc-001", "to_service_account_id": 1, }, ) assert resp.status_code == 201 async def test_get_returns_200(self, contract_client): client, _ = contract_client post_resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "src-sc-002", "to_playlist_id": "tgt-sc-002", "to_service_account_id": 1, }, ) sync_id = post_resp.json()["id"] resp = await client.get(f"/PlaylistSync/US/playlists/{sync_id}") assert resp.status_code == 200 async def test_update_returns_200(self, contract_client): client, _ = contract_client post_resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "src-sc-003", "to_playlist_id": "tgt-sc-003", "to_service_account_id": 1, }, ) sync_id = post_resp.json()["id"] resp = await client.put( f"/PlaylistSync/US/playlists/{sync_id}", json={"title": "Updated"} ) assert resp.status_code == 200 async def test_delete_returns_204(self, contract_client): client, _ = contract_client post_resp = await client.post( "/PlaylistSync/US/playlists", json={ "from_playlist_id": "src-sc-004", "to_playlist_id": "tgt-sc-004", "to_service_account_id": 1, }, ) sync_id = post_resp.json()["id"] resp = await client.delete(f"/PlaylistSync/US/playlists/{sync_id}") assert resp.status_code == 204 async def test_duplicate_create_returns_409(self, contract_client): """Creating a sync for an already-used to_playlist_id returns 409.""" client, _ = contract_client payload = { "from_playlist_id": "src-sc-005", "to_playlist_id": "tgt-sc-005-unique", "to_service_account_id": 1, } await client.post("/PlaylistSync/US/playlists", json=payload) resp = await client.post("/PlaylistSync/US/playlists", json=payload) assert resp.status_code == 409 async def test_unknown_market_returns_400(self, contract_client): client, _ = contract_client resp = await client.get("/PlaylistSync/INVALID_MARKET/playlists") assert resp.status_code == 400 async def test_missing_sync_returns_404(self, contract_client): client, _ = contract_client resp = await client.get("/PlaylistSync/US/playlists/9999999") assert resp.status_code == 404 async def test_delete_missing_returns_404(self, contract_client): client, _ = contract_client resp = await client.delete("/PlaylistSync/US/playlists/9999999") assert resp.status_code == 404 async def test_update_missing_returns_404(self, contract_client): client, _ = contract_client resp = await client.put( "/PlaylistSync/US/playlists/9999999", json={"title": "X"} ) assert resp.status_code == 404 async def test_health_returns_200(self, contract_client): """Health endpoint returns a valid status response.""" client, _ = contract_client resp = await client.get("/health") # 200 when all deps healthy; 503 when Redis not available (e.g. CI) assert resp.status_code in (200, 503) data = resp.json() assert "status" in data assert data["status"] in ("healthy", "degraded") # --------------------------------------------------------------------------- # Side-by-side parity tests (skipped unless DOTNET_API_URL is set) # --------------------------------------------------------------------------- @pytest.mark.parity @pytest.mark.skipif( not DOTNET_API_URL, reason="DOTNET_API_URL not configured; skipping parity tests" ) class TestDotnetParity: r"""Compare Python API responses against the live .NET AdminAPI. To run these tests: DOTNET_API_URL=https://admin-api-ext-internal.apollo.stream \\ ADMIN_API_KEY= \\ PYTHON_API_URL=http://localhost:8001 \\ pytest tests/contract/test_contract.py::TestDotnetParity -v Or load from .env.shadow: set -a && source .env.shadow && set +a DOTNET_API_URL=$ADMIN_API_HOST PYTHON_API_URL=http://localhost:8001 \\ pytest tests/contract/test_contract.py -m parity -v """ PYTHON_API_URL = os.environ.get("PYTHON_API_URL", "http://localhost:8001") @property def dotnet_headers(self) -> dict: """Authentication header for the .NET AdminAPI. The .NET AdminAPI uses a custom 'FiltrAuthentication' header, not 'Authorization'. """ if ADMIN_API_KEY: return {"FiltrAuthentication": ADMIN_API_KEY} return {} async def test_list_syncs_response_shapes_match(self): """Both APIs return the same set of top-level fields for list responses. Known differences (tracked as TODO): - Python API only returns active syncs; .NET returns all (including inactive). - Python API does not yet join the last log-entry fields (addedTracks, deletedTracks, madeChange, time, errorText, errorType, triggeredManually, syncedTrackCount, toServiceType, insertMedia). """ import httpx async with httpx.AsyncClient() as client: dotnet_resp = await client.get( f"{DOTNET_API_URL}/PlaylistSync/US/playlists", headers=self.dotnet_headers, ) python_resp = await client.get( f"{self.PYTHON_API_URL}/PlaylistSync/US/playlists" ) assert dotnet_resp.status_code == 200, ( f".NET API returned {dotnet_resp.status_code}: {dotnet_resp.text[:200]}" ) assert python_resp.status_code == 200, ( f"Python API returned {python_resp.status_code}: {python_resp.text[:200]}" ) dotnet_items = dotnet_resp.json() python_items = python_resp.json() if dotnet_items and python_items: # Normalise to lowercase before comparing so camelCase vs PascalCase # (a JSON serialisation convention difference) doesn't mask real gaps. dotnet_keys_norm = {k.lower() for k in dotnet_items[0].keys()} python_keys_norm = {k.lower() for k in python_items[0].keys()} # Known missing: log-entry joined fields not yet implemented in Python API. KNOWN_MISSING = { "addedtracks", "deletedtracks", "deletedduplicates", "madechange", "time", "errortext", "errortype", "triggedmanually", "triggeredmanually", "syncedtrackcount", "toservicetype", "insertmedia", } actually_missing = dotnet_keys_norm - python_keys_norm - KNOWN_MISSING assert not actually_missing, ( "Python response missing unexpected fields" f" (not in known-TODO list): {actually_missing}" ) async def test_sync_counts_match(self): """Both APIs return the same number of active syncs for US market. NOTE: Python currently filters to active=True only; .NET may return inactive syncs too. This test is informational — it reports the difference but does not hard-fail. """ import httpx async with httpx.AsyncClient() as client: dotnet_resp = await client.get( f"{DOTNET_API_URL}/PlaylistSync/US/playlists", headers=self.dotnet_headers, ) python_resp = await client.get( f"{self.PYTHON_API_URL}/PlaylistSync/US/playlists" ) assert dotnet_resp.status_code == 200 assert python_resp.status_code == 200 dotnet_count = len(dotnet_resp.json()) python_count = len(python_resp.json()) # Python only serves active syncs — a known intentional difference. # If the difference is larger than the expected inactive-record gap, fail. assert python_count <= dotnet_count, ( f"Python returned MORE syncs ({python_count}) than .NET ({dotnet_count});" " unexpected — Python should return a subset of active records" ) # Soft informational message — not a hard failure for the count gap itself. if python_count != dotnet_count: print( f"\nINFO: Count difference — .NET: {dotnet_count}," f" Python: {python_count}." " Python filters active=True; inactive syncs are excluded by design." ) async def test_get_single_sync_fields_match(self): """For the first sync, both APIs return the same fields on the detail endpoint. Known difference: the .NET detail endpoint joins the last log entry and returns its fields (addedTracks, deletedTracks, madeChange, time, errorText, errorType, triggeredManually, syncedTrackCount, toServiceType, insertMedia). The Python API does not yet implement this join — tracked as a TODO. """ import httpx async with httpx.AsyncClient() as client: list_resp = await client.get( f"{DOTNET_API_URL}/PlaylistSync/US/playlists", headers=self.dotnet_headers, ) items = list_resp.json() if not items: pytest.skip("No syncs in US market to compare") sync_id = items[0]["id"] async with httpx.AsyncClient() as client: dotnet_resp = await client.get( f"{DOTNET_API_URL}/PlaylistSync/US/playlists/{sync_id}", headers=self.dotnet_headers, ) python_resp = await client.get( f"{self.PYTHON_API_URL}/PlaylistSync/US/playlists/{sync_id}" ) assert dotnet_resp.status_code == python_resp.status_code if dotnet_resp.status_code == 200: dotnet_keys_norm = {k.lower() for k in dotnet_resp.json().keys()} python_keys_norm = {k.lower() for k in python_resp.json().keys()} KNOWN_MISSING = { "addedtracks", "deletedtracks", "deletedduplicates", "madechange", "time", "errortext", "errortype", "triggedmanually", "triggeredmanually", "syncedtrackcount", "toservicetype", "insertmedia", } actually_missing = dotnet_keys_norm - python_keys_norm - KNOWN_MISSING assert not actually_missing, ( f"Python detail response missing unexpected fields: {actually_missing}" ) # --------------------------------------------------------------------------- # AppMarketResponse shape # --------------------------------------------------------------------------- class TestAppMarketShape: """Assert the exact JSON field set for GET /apollo-api/app-markets/ responses. These field names must match the apollo-api ApplicationMarketSchema serializer output so that any existing consumer continues to work after migration. """ REQUIRED_KEYS = { "id", "name", "cultureInfo", "spotifyRegionCode", "active", "gaCountryName", "defaultService", "services", "workoutMarket", "includeOtherPlaylists", } async def test_returns_200(self, contract_client): ac, _ = contract_client response = await ac.get("/apollo-api/app-markets/") assert response.status_code == 200 async def test_response_is_list(self, contract_client): ac, _ = contract_client response = await ac.get("/apollo-api/app-markets/") assert isinstance(response.json(), list) async def test_item_has_all_required_keys(self, contract_client): ac, _ = contract_client data = (await ac.get("/apollo-api/app-markets/")).json() assert len(data) >= 1, "Need at least one seeded market to assert shape" item = data[0] missing = self.REQUIRED_KEYS - item.keys() assert not missing, f"Response item missing keys: {missing}" async def test_no_unexpected_snake_case_keys(self, contract_client): """Ensures no raw Python attribute names (snake_case) leak into the response.""" ac, _ = contract_client data = (await ac.get("/apollo-api/app-markets/")).json() snake_case_keys = { "language_id", "spotify_region_code", "ga_country_name", "default_service", "service_list", "workout_market", "include_other_playlists", } for item in data: leaked = snake_case_keys & item.keys() assert not leaked, f"Snake-case keys leaked into response: {leaked}" async def test_services_is_list(self, contract_client): """Services must always be a JSON array, never a raw string.""" ac, _ = contract_client data = (await ac.get("/apollo-api/app-markets/")).json() for item in data: assert isinstance(item["services"], list), ( f"services is not a list for market id={item.get('id')}" ) async def test_boolean_fields_are_booleans(self, contract_client): """workoutMarket, includeOtherPlaylists, and active must be JSON booleans.""" ac, _ = contract_client data = (await ac.get("/apollo-api/app-markets/")).json() for item in data: assert isinstance(item["workoutMarket"], bool) assert isinstance(item["includeOtherPlaylists"], bool) assert isinstance(item["active"], bool) async def test_id_is_integer(self, contract_client): """Id field must be an integer.""" ac, _ = contract_client data = (await ac.get("/apollo-api/app-markets/")).json() for item in data: if item["id"] is not None: assert isinstance(item["id"], int)