diff --git a/tests/test_acceptance.py b/tests/test_acceptance.py new file mode 100644 index 0000000..2db5b65 --- /dev/null +++ b/tests/test_acceptance.py @@ -0,0 +1,250 @@ +"""Application-level, multi-user acceptance tests for the full MyWhoosh -> Garmin +sync pipeline (Task 7 of the sync-scheduler-web plan). + +These tests build a real `SyncManager` wired to a real SQLite database and a +real `CredentialCipher`, but with fake MyWhoosh/Garmin factories, so they +exercise the whole state machine end-to-end for two independent users without +touching any real external service. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from cryptography.fernet import Fernet +from sqlalchemy import func, select +from sqlalchemy.orm import sessionmaker + +from app.db.models import Activity, ActivityStatus, HealthState, SyncUser +from app.db.repositories import UserRepository +from app.db.session import create_db_engine, create_session_factory, initialize_schema +from app.garmin.uploader import GarminUploadBlocked, UploadResult +from app.mywhoosh.models import MyWhooshActivity +from app.security.credentials import CredentialCipher +from app.sync.manager import SyncManager + + +class StubSettings: + """Minimal stand-in for app.config.Settings exposing only the two + properties SyncManager needs.""" + + def __init__(self, tmp_path: Path) -> None: + self.tokens_dir = tmp_path / "tokens" + self.activities_dir = tmp_path / "activities" + + +class FakeMyWhooshClient: + def __init__(self, activities, fit_bytes: bytes) -> None: + self.activities = activities + self.fit_bytes = fit_bytes + self.list_calls = 0 + self.download_calls = 0 + + async def list_activities(self, email, password): + self.list_calls += 1 + return list(self.activities) + + async def download_fit(self, activity_file_id, email, password): + self.download_calls += 1 + return self.fit_bytes + + +class FakeGarminUploader: + def __init__(self, result: UploadResult | None = None, error: Exception | None = None) -> None: + self.result = result or UploadResult("imported", False, "g-1", {"activityId": "g-1"}) + self.error = error + self.calls = 0 + + def import_fit(self, fit_path, mfa_code=None): + self.calls += 1 + if self.error is not None: + raise self.error + return self.result + + +def fake_fit_converter(source_path: Path, output_path: Path) -> None: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(b"fake-converted-fit-bytes") + + +def count_terminal_activities(session_factory: sessionmaker, user_id: int) -> int: + with session_factory() as session: + return session.scalar( + select(func.count()) + .select_from(Activity) + .where( + Activity.user_id == user_id, + Activity.status.in_([ActivityStatus.IMPORTED, ActivityStatus.DUPLICATE]), + ) + ) + + +def _create_user(session, cipher: CredentialCipher, *, name: str) -> SyncUser: + return UserRepository(session).create( + name=name, + enabled=True, + mywhoosh_email_enc=cipher.encrypt(f"{name.lower()}-mywhoosh@example.com"), + mywhoosh_password_enc=cipher.encrypt("mw-secret"), + garmin_email_enc=cipher.encrypt(f"{name.lower()}-garmin@example.com"), + garmin_password_enc=cipher.encrypt("garmin-secret"), + ) + + +@pytest.fixture +def session_factory(tmp_path: Path): + db_path = tmp_path / "acceptance.db" + engine = create_db_engine(f"sqlite:///{db_path}") + initialize_schema(engine) + factory = create_session_factory(engine) + try: + yield factory + finally: + engine.dispose() + + +@pytest.fixture +def cipher() -> CredentialCipher: + return CredentialCipher(Fernet.generate_key().decode("ascii")) + + +@pytest.fixture +def settings(tmp_path: Path) -> StubSettings: + return StubSettings(tmp_path) + + +@pytest.fixture +def two_users(session_factory, cipher: CredentialCipher): + with session_factory() as session: + user_a = _create_user(session, cipher, name="Alice") + user_b = _create_user(session, cipher, name="Bob") + return user_a, user_b + + +def _build_manager( + *, + session_factory, + cipher: CredentialCipher, + settings: StubSettings, + user_a: SyncUser, + user_b: SyncUser, + fake_mw_a: FakeMyWhooshClient, + fake_mw_b: FakeMyWhooshClient, + fake_garmin_a: FakeGarminUploader, + fake_garmin_b: FakeGarminUploader, +): + mywhoosh_fakes = {str(user_a.id): fake_mw_a, str(user_b.id): fake_mw_b} + garmin_fakes = {str(user_a.id): fake_garmin_a, str(user_b.id): fake_garmin_b} + garmin_factory_calls: list[tuple[str, str, Path]] = [] + + def mywhoosh_factory(token_store): + user_key = token_store.path.parent.name + return mywhoosh_fakes[user_key] + + def garmin_factory(email, password, tokenstore): + garmin_factory_calls.append((email, password, tokenstore)) + user_key = tokenstore.parent.name + return garmin_fakes[user_key] + + manager = SyncManager( + session_factory=session_factory, + credential_cipher=cipher, + settings=settings, + mywhoosh_factory=mywhoosh_factory, + garmin_factory=garmin_factory, + fit_converter=fake_fit_converter, + ) + return manager, garmin_factory_calls + + +@pytest.mark.asyncio +async def test_two_user_happy_path(session_factory, cipher, settings, two_users): + user_a, user_b = two_users + + remote_a = MyWhooshActivity( + id="remote-a-1", title="Alice Ride", activity_file_id="file-a-1", started_at=None + ) + remote_b = MyWhooshActivity( + id="remote-b-1", title="Bob Ride", activity_file_id="file-b-1", started_at=None + ) + fake_mw_a = FakeMyWhooshClient(activities=[remote_a], fit_bytes=b"alice-source-bytes") + fake_mw_b = FakeMyWhooshClient(activities=[remote_b], fit_bytes=b"bob-source-bytes") + fake_garmin_a = FakeGarminUploader() + fake_garmin_b = FakeGarminUploader() + + manager, garmin_factory_calls = _build_manager( + session_factory=session_factory, + cipher=cipher, + settings=settings, + user_a=user_a, + user_b=user_b, + fake_mw_a=fake_mw_a, + fake_mw_b=fake_mw_b, + fake_garmin_a=fake_garmin_a, + fake_garmin_b=fake_garmin_b, + ) + + results = await manager.sync_all_enabled() + + assert all(result.status == "success" for result in results) + assert count_terminal_activities(session_factory, user_a.id) == 1 + assert count_terminal_activities(session_factory, user_b.id) == 1 + + with session_factory() as session: + activity_a = session.scalar(select(Activity).where(Activity.user_id == user_a.id)) + activity_b = session.scalar(select(Activity).where(Activity.user_id == user_b.id)) + assert Path(activity_a.source_fit_path).exists() + assert Path(activity_b.source_fit_path).exists() + assert Path(activity_a.source_fit_path).parent != Path(activity_b.source_fit_path).parent + + tokenstores = {str(call[2]) for call in garmin_factory_calls} + assert len(tokenstores) == 2 + + +@pytest.mark.asyncio +async def test_isolation_when_one_user_requires_mfa(session_factory, cipher, settings, two_users): + user_a, user_b = two_users + + remote_a = MyWhooshActivity( + id="remote-a-1", title="Alice Ride", activity_file_id="file-a-1", started_at=None + ) + remote_b = MyWhooshActivity( + id="remote-b-1", title="Bob Ride", activity_file_id="file-b-1", started_at=None + ) + fake_mw_a = FakeMyWhooshClient(activities=[remote_a], fit_bytes=b"alice-source-bytes") + fake_mw_b = FakeMyWhooshClient(activities=[remote_b], fit_bytes=b"bob-source-bytes") + fake_garmin_a = FakeGarminUploader() + fake_garmin_b = FakeGarminUploader(error=GarminUploadBlocked("Garmin requested MFA")) + + manager, garmin_factory_calls = _build_manager( + session_factory=session_factory, + cipher=cipher, + settings=settings, + user_a=user_a, + user_b=user_b, + fake_mw_a=fake_mw_a, + fake_mw_b=fake_mw_b, + fake_garmin_a=fake_garmin_a, + fake_garmin_b=fake_garmin_b, + ) + + results = await manager.sync_all_enabled() + + results_by_user = {result.user_id: result for result in results} + assert results_by_user[user_a.id].status == "success" + assert results_by_user[user_b.id].status != "success" + assert results_by_user[user_b.id].status == "failed" + + assert count_terminal_activities(session_factory, user_a.id) == 1 + + with session_factory() as session: + reloaded_b = UserRepository(session).get(user_b.id) + assert reloaded_b.health_state == HealthState.ACTION_REQUIRED + assert reloaded_b.action_reason == "garmin_mfa_required" + + reloaded_a = UserRepository(session).get(user_a.id) + assert reloaded_a.health_state == HealthState.HEALTHY + assert reloaded_a.action_reason is None + + activity_a = session.scalar(select(Activity).where(Activity.user_id == user_a.id)) + assert activity_a.status in (ActivityStatus.IMPORTED, ActivityStatus.DUPLICATE)