Stride/tests/test_app.py

from datetime import UTC, datetime
from pathlib import Path

import pytest

from app import create_app
from app.display import persist_analysis
from app.extensions import db
from app.models import Activity, Friendship, User
from app.records import EFFORTS_VERSION, persist_efforts
from app.services import activity_data, import_cache


@pytest.fixture
def app(tmp_path):
    app = create_app(
        {
            "TESTING": True,
            "SQLALCHEMY_DATABASE_URI": "sqlite://",
            "SECRET_KEY": "test-only",
            "WTF_CSRF_ENABLED": False,
            "RATELIMIT_ENABLED": False,
            "LOCAL_SETUP": False,
            "DATA_ROOT": str(tmp_path),
        }
    )
    with app.app_context():
        db.create_all()
        for name in ("runner", "other"):
            user = User(username=name, email=f"{name}@example.test", display_name=name.title())
            user.set_password("a-long-test-password")
            db.session.add(user)
        db.session.commit()
        db.session.add(
            Activity(
                owner_id=1,
                source_key="test:1",
                title="A quiet morning run",
                has_route=True,
                sport="running",
                started_at=datetime(2026, 9, 30, tzinfo=UTC),
                distance_m=5000,
                duration_s=1800,
                elevation_m=40,
                avg_hr=140,
                samples=[
                    {
                        "point": [-33, 151],
                        "distance": 0,
                        "heart_rate": 120,
                        "speed": 3,
                        "elevation": 10,
                    }
                ],
                laps=[{"distance": 1000, "duration": 360, "heart_rate": 140, "pace": 360}],
            )
        )
        db.session.commit()
        persist_analysis(db.session.get(Activity, 1))
        db.session.commit()
    yield app


def login(client, name="runner"):
    return client.post(
        "/login", data={"email": f"{name}@example.test", "password": "a-long-test-password"}
    )


def test_private_activity_authorization_for_pages_json_and_mutations(app):
    client = app.test_client()
    assert client.get("/activities/1").status_code == 404
    assert client.get("/activities/1/data").status_code == 404
    login(client, "other")
    assert client.get("/activities/1").status_code == 404
    assert client.get("/activities/1/data").status_code == 404
    assert client.post("/activities/1/edit", data={"title": "stolen"}).status_code == 404
    assert client.post("/imports/cache").status_code == 404
    assert b"A quiet morning run" not in client.get("/dashboard").data


def test_owner_dashboard_activity_json_and_settings_render(app):
    client = app.test_client()
    login(client)
    for path in ("/dashboard", "/activities/1", "/settings"):
        response = client.get(path)
        assert response.status_code == 200
        assert b"Stride" in response.data
    data = client.get("/activities/1/data").json
    assert data["segments"] == [[[-33, 151]]]
    assert "point" not in data["chart"][0]


def test_route_preview_access_is_revoked_with_friendship(app):
    owner, friend, anonymous = app.test_client(), app.test_client(), app.test_client()
    login(owner)
    login(friend, "other")
    assert anonymous.get("/activities/1/route").status_code == 404
    assert friend.get("/activities/1/route").status_code == 404
    assert owner.get("/activities/1/route").json == {"segments": [[[-33, 151]]]}
    owner.post("/activities/1/edit", data={"title": "Shared", "visibility": "friends"})
    owner.post("/friends", data={"email": "other@example.test"})
    friend.post("/friends/1/accept")
    assert friend.get("/activities/1/route").status_code == 200
    friend.post("/friends/1/remove")
    assert friend.get("/activities/1/route").status_code == 404


def test_stored_analysis_requests_do_not_read_raw_samples_or_recalculate(app, monkeypatch):
    from sqlalchemy import event

    from app import display

    def fail(_row):
        raise AssertionError("Analysis must be precomputed")

    monkeypatch.setattr(display, "analyse", fail)
    with app.app_context():
        db.session.get(Activity, 1).share_token = "compact-test"
        db.session.commit()
        statements = []

        def capture(_conn, _cursor, statement, _params, _context, _many):
            statements.append(statement)

        engine = db.engine
        event.listen(engine, "before_cursor_execute", capture)
    client = app.test_client()
    login(client)
    try:
        for path in (
            "/dashboard",
            "/activities/1",
            "/activities/1/data",
            "/activities/1/route",
            "/shared/runs/compact-test",
            "/shared/runs/compact-test/data",
        ):
            assert client.get(path).status_code == 200
        assert all("activity.samples" not in s and "activity.laps" not in s for s in statements)
    finally:
        event.remove(engine, "before_cursor_execute", capture)


def test_analysis_backfill_resumes_after_failed_batch(app, monkeypatch):
    from app import display

    with app.app_context():
        db.session.get(Activity, 1).analysis_version = 0
        db.session.add(
            Activity(
                owner_id=2,
                source_key="analysis:2",
                title="Older activity",
                started_at=datetime(2026, 1, 1, tzinfo=UTC),
                sport="cycling",
                samples=[],
                laps=[],
            )
        )
        db.session.commit()
    original = display.persist_analysis

    def interrupt(row):
        if row.id == 2:
            raise ValueError("Interrupted")
        original(row)

    monkeypatch.setattr(display, "persist_analysis", interrupt)
    runner = app.test_cli_runner()
    result = runner.invoke(args=["backfill-analysis", "--batch-size", "1"])
    assert result.exit_code != 0
    with app.app_context():
        assert db.session.get(Activity, 1).analysis_version == display.ANALYSIS_VERSION
        assert db.session.get(Activity, 2).analysis_version == 0
    monkeypatch.setattr(display, "persist_analysis", original)
    result = runner.invoke(args=["backfill-analysis", "--owner-id", "1"])
    assert result.exit_code == 0
    assert "0 activities backfilled" in result.output
    result = runner.invoke(args=["backfill-analysis", "--batch-size", "1"])
    assert result.exit_code == 0
    assert "1 activities backfilled" in result.output
    result = runner.invoke(args=["backfill-analysis"])
    assert result.exit_code == 0
    assert "0 activities backfilled" in result.output


def test_personal_records_render_and_remain_owner_only(app):
    with app.app_context():
        row = db.session.get(Activity, 1)
        row.samples = [{"distance": 0, "time": 0}, {"distance": 5000, "time": 1500}]
        persist_efforts(row)
        persist_analysis(row)
        row.share_token = "record-test"
        db.session.commit()
    client = app.test_client()
    login(client)
    dashboard = client.get("/dashboard").data
    assert b"Personal records" in dashboard
    assert b"5 km PR" in dashboard
    detail = client.get("/activities/1").data
    assert b"Best efforts on this run" in detail
    assert b"5 km" in detail
    assert b"Best efforts on this run" not in client.get("/shared/runs/record-test").data
    client.post("/logout")
    login(client, "other")
    assert b"5 km PR" not in client.get("/dashboard").data


def test_import_and_refresh_persist_efforts_in_the_run_transaction(app):
    from app.services import activity_from_fit

    doc = {
        "session": [{"sport": "running", "start_time": "2026-09-30T10:00:00Z"}],
        "record": [
            {"timestamp": "2026-09-30T10:00:00Z", "distance": 0},
            {"timestamp": "2026-09-30T10:05:00Z", "distance": 1000},
        ],
    }
    with app.app_context():
        row = activity_from_fit(1, "test:new", [doc])
        db.session.commit()
        row_id = row.id
        db.session.remove()
        row = db.session.get(Activity, row_id)
        assert row.efforts_version == EFFORTS_VERSION
        assert row.best_efforts[0]["seconds"] == 300
        assert row.analysis_version == 1
        assert row.analysis_summary["summary"]["samples"] == 2
        assert activity_from_fit(1, "test:new", [doc]) is None
        doc["record"][1]["timestamp"] = "2026-09-30T10:04:00Z"
        activity_from_fit(1, "test:new", [doc], refresh=True)
        db.session.commit()
        assert row.best_efforts[0]["seconds"] == 240
        doc["record"] = []
        activity_from_fit(1, "test:new", [doc], refresh=True)
        db.session.commit()
        assert row.best_efforts == []
        assert row.display_data == {"route_segments": [], "chart": []}
        assert row.route_geometry == []
        assert row.efforts_version == EFFORTS_VERSION


def test_backfill_is_resumable_and_dashboard_uses_only_stored_efforts(app, monkeypatch):
    from sqlalchemy import event

    from app import records

    with app.app_context():
        row = db.session.get(Activity, 1)
        row.samples = [
            {"distance": 0, "time": 0, "point": [-33, 151]},
            {"distance": 5000, "time": 1500},
        ]
        db.session.add(
            Activity(
                owner_id=1,
                source_key="old:2",
                title="Old ride",
                sport="cycling",
                started_at=row.started_at,
                samples=[],
            )
        )
        db.session.commit()
    runner = app.test_cli_runner()
    result = runner.invoke(args=["backfill-efforts", "--batch-size", "1"])
    assert result.exit_code == 0, result.output
    assert "2 activities backfilled" in result.output
    result = runner.invoke(args=["backfill-efforts"])
    assert result.exit_code == 0
    assert "0 activities backfilled" in result.output
    with app.app_context():
        assert db.session.get(Activity, 2).best_efforts == []
        assert db.session.get(Activity, 2).efforts_version == EFFORTS_VERSION
        statements = []

        def capture(_conn, _cursor, statement, _params, _context, _many):
            statements.append(statement)

        event.listen(db.engine, "before_cursor_execute", capture)
        engine = db.engine

    def fail_calculation(_row):
        raise AssertionError("Page requests must never calculate from raw samples")

    monkeypatch.setattr(records, "best_efforts", fail_calculation)
    try:
        client = app.test_client()
        login(client)
        assert b"5 km PR" in client.get("/dashboard").data
        assert all("activity.samples" not in s and "activity.laps" not in s for s in statements)
        assert b"Best efforts on this run" in client.get("/activities/1").data
    finally:
        event.remove(engine, "before_cursor_execute", capture)


def test_backfill_keeps_completed_batches_after_failure(app, monkeypatch):
    from app import records

    with app.app_context():
        db.session.add(
            Activity(
                owner_id=2,
                source_key="old:2",
                title="Second athlete",
                sport="running",
                started_at=datetime(2026, 1, 1, tzinfo=UTC),
                samples=[],
            )
        )
        db.session.commit()
    original = records.persist_efforts

    def fail_second(row):
        if row.id == 2:
            raise ValueError("Interrupted batch")
        original(row)

    monkeypatch.setattr(records, "persist_efforts", fail_second)
    runner = app.test_cli_runner()
    result = runner.invoke(args=["backfill-efforts", "--batch-size", "1"])
    assert result.exit_code != 0
    with app.app_context():
        assert db.session.get(Activity, 1).efforts_version == EFFORTS_VERSION
        assert db.session.get(Activity, 2).efforts_version == 0
    monkeypatch.setattr(records, "persist_efforts", original)
    result = runner.invoke(args=["backfill-efforts", "--owner-id", "1"])
    assert result.exit_code == 0
    assert "0 activities backfilled" in result.output
    result = runner.invoke(args=["backfill-efforts", "--batch-size", "1"])
    assert result.exit_code == 0
    assert "1 activities backfilled" in result.output
    result = runner.invoke(args=["backfill-efforts", "--force", "--owner-id", "1"])
    assert result.exit_code == 0
    assert "1 activities backfilled" in result.output


def test_efforts_migration_upgrades_existing_schema(tmp_path):
    from sqlalchemy import text

    application = create_app(
        {
            "TESTING": True,
            "SECRET_KEY": "test",
            "SQLALCHEMY_DATABASE_URI": f"sqlite:///{tmp_path / 'migration.db'}",
            "RATELIMIT_ENABLED": False,
        }
    )
    runner = application.test_cli_runner()
    result = runner.invoke(args=["db", "upgrade", "d81b97ae5401"])
    assert result.exit_code == 0, result.output
    with application.app_context():
        db.session.execute(
            text(
                "INSERT INTO user (id, username, email, display_name, password_hash, timezone) "
                "VALUES (1, 'test', 'test@example.test', 'Test', 'test', 'UTC')"
            )
        )
        db.session.execute(
            text(
                "INSERT INTO activity (owner_id, source_key, title, sport, started_at, "
                "distance_m, duration_s, elevation_m, visibility, samples, laps) "
                "VALUES (1, 'old', 'Old run', 'running', '2026-01-01', "
                "1000, 300, 0, 'private', '[]', '[]')"
            )
        )
        db.session.commit()
    result = runner.invoke(args=["db", "upgrade"])
    assert result.exit_code == 0, result.output
    with application.app_context():
        row = db.session.get(Activity, 1)
        assert row.best_efforts == []
        assert row.efforts_version == 0
        assert row.analysis_version == 0
        assert row.analysis_summary == {}
        assert row.display_data == {}
        assert row.route_geometry == []
        assert row.has_route is False
        assert row.title == "Old run"
    result = runner.invoke(args=["backfill-efforts"])
    assert result.exit_code == 0, result.output
    with application.app_context():
        assert db.session.get(Activity, 1).efforts_version == EFFORTS_VERSION


def test_dashboard_replays_latest_accessible_gps_run(app):
    client = app.test_client()
    login(client)
    page = client.get("/dashboard").data
    assert b'id="dashboard-route-map" data-url="/activities/1/route"' in page
    assert b'id="toggle-route-replay"' not in page
    assert b"vendor/leaflet.js" in page
    assert b'class="dashboard-map"' not in page
    client.post("/logout")
    login(client, "other")
    page = client.get("/dashboard").data
    assert b"No GPS running route available yet." in page
    assert b"/activities/1/data" not in page
    with app.app_context():
        row = db.session.get(Activity, 1)
        row.samples = []
        persist_efforts(row)
        persist_analysis(row)
        db.session.commit()
    client.post("/logout")
    login(client)
    assert b"No GPS running route available yet." in client.get("/dashboard").data


def test_friend_access_requires_acceptance_and_revokes_immediately(app):
    owner = app.test_client()
    friend = app.test_client()
    anonymous = app.test_client()
    login(owner)
    login(friend, "other")
    owner.post("/activities/1/edit", data={"title": "Shared morning run", "visibility": "friends"})
    assert anonymous.get("/activities/1").status_code == 404
    assert anonymous.get("/activities/1/data").status_code == 404
    assert anonymous.get("/explore").status_code == 404
    assert friend.get("/activities/1").status_code == 404
    assert friend.get("/u/runner").status_code == 404
    owner.post("/friends", data={"email": "other@example.test"})
    assert friend.get("/activities/1/data").status_code == 404
    assert owner.post("/friends/1/accept").status_code == 400
    assert friend.post("/friends/1/accept").status_code == 302
    assert b"Shared morning run" in friend.get("/friends").data
    assert b"Shared morning run" in friend.get("/u/runner").data
    data = friend.get("/activities/1/data")
    assert data.status_code == 200
    assert data.json["segments"] == [[[-33, 151]]]
    assert "no-store" in data.headers["Cache-Control"]
    assert friend.post("/activities/1/edit", data={"title": "changed"}).status_code == 404
    friend.post("/friends/1/remove")
    assert friend.get("/activities/1/data").status_code == 404
    assert friend.get("/u/runner").status_code == 404
    assert b"Shared morning run" not in friend.get("/friends").data


def test_private_activity_is_hidden_even_from_accepted_friend(app):
    with app.app_context():
        db.session.add(Friendship(low_id=1, high_id=2, requested_by=1, status="accepted"))
        db.session.commit()
    friend = app.test_client()
    login(friend, "other")
    assert friend.get("/activities/1").status_code == 404
    assert friend.get("/activities/1/data").status_code == 404
    assert b"A quiet morning run" not in friend.get("/u/runner").data


def test_full_route_includes_start_and_finish():
    points = [[0, 0], [0, 0.001], [0, 0.01], [0, 0.02]]
    row = Activity(
        sport="running", distance_m=0, laps=[], samples=[{"point": point} for point in points]
    )
    persist_analysis(row)
    assert activity_data(row)["segments"] == [points]


def test_public_run_link_copy_view_revoke_and_recreate(app):
    owner = app.test_client()
    anonymous = app.test_client()
    login(owner)
    assert b'id="copy-run-link"' in owner.get("/activities/1").data
    response = owner.post("/activities/1/share")
    assert response.status_code == 200
    url = response.json["url"]
    assert url.startswith("http://localhost/shared/runs/")
    assert len(url.rsplit("/", 1)[1]) >= 40
    assert owner.post("/activities/1/share").json["url"] == url
    page = anonymous.get(url)
    assert page.status_code == 200
    assert b"A quiet morning run" in page.data
    assert b"Shared via link" in page.data
    assert b'id="copy-run-link"' not in page.data
    assert b"/activities/1/edit" not in page.data
    assert b"/u/runner" not in page.data
    assert (url.removeprefix("http://localhost") + "/data").encode() in page.data
    assert page.headers["X-Robots-Tag"] == "noindex, nofollow"
    assert page.headers["Referrer-Policy"] == "strict-origin"
    data = anonymous.get(url + "/data")
    assert data.status_code == 200
    assert data.json["segments"] == [[[-33, 151]]]
    assert "no-store" in data.headers["Cache-Control"]
    assert anonymous.get("/activities/1").status_code == 404
    assert anonymous.get("/activities/1/data").status_code == 404
    with app.app_context():
        assert db.session.get(Activity, 1).visibility == "private"
    assert owner.post("/activities/1/share/revoke").status_code == 302
    assert anonymous.get(url).status_code == 404
    assert anonymous.get(url + "/data").status_code == 404
    assert owner.post("/activities/1/share").json["url"] != url


def test_only_owner_can_create_or_revoke_public_run_links(app):
    client = app.test_client()
    for suffix in ("share", "share/revoke"):
        assert client.post(f"/activities/1/{suffix}").status_code == 302
    login(client, "other")
    with app.app_context():
        db.session.add(Friendship(low_id=1, high_id=2, requested_by=1, status="accepted"))
        db.session.get(Activity, 1).visibility = "friends"
        db.session.commit()
    assert client.get("/activities/1").status_code == 200
    assert b'id="copy-run-link"' not in client.get("/activities/1").data
    for suffix in ("share", "share/revoke"):
        assert client.post(f"/activities/1/{suffix}").status_code == 404
    for suffix in ("", "/data"):
        assert client.get(f"/shared/runs/invalid-token{suffix}").status_code == 404


def test_public_run_link_mutations_require_csrf(app):
    client = app.test_client()
    login(client)
    app.config["WTF_CSRF_ENABLED"] = True
    for suffix in ("share", "share/revoke"):
        assert client.post(f"/activities/1/{suffix}").status_code == 400


def test_register_creates_separate_user_with_no_local_account(app):
    client = app.test_client()
    response = client.post(
        "/register",
        data={
            "username": "newrunner",
            "email": "new@example.test",
            "display_name": "New runner",
            "password": "a-long-safe-password",
            "local_account_key": "forged",
        },
    )
    assert response.status_code == 302
    with app.app_context():
        user = db.session.scalar(db.select(User).filter_by(email="new@example.test"))
        assert user.local_account_key is None
        assert user.password_hash != "a-long-safe-password"
    assert b"A quiet morning run" not in client.get("/dashboard").data


def test_csrf_rejects_state_changes(app):
    app.config["WTF_CSRF_ENABLED"] = True
    client = app.test_client()
    assert client.post("/login", data={"email": "runner@example.test"}).status_code == 400
    assert client.post("/imports/fit").status_code == 400


def test_setup_only_on_loopback_and_before_accounts(tmp_path):
    app = create_app(
        {
            "TESTING": True,
            "SQLALCHEMY_DATABASE_URI": "sqlite://",
            "SECRET_KEY": "test",
            "WTF_CSRF_ENABLED": False,
            "LOCAL_SETUP": True,
            "RATELIMIT_ENABLED": False,
            "DATA_ROOT": str(tmp_path),
        }
    )
    with app.app_context():
        db.create_all()
    client = app.test_client()
    assert client.get("/setup", environ_base={"REMOTE_ADDR": "192.168.1.5"}).status_code == 404
    assert client.get("/setup").status_code == 200
    assert (
        client.post(
            "/setup",
            data={
                "username": "owner",
                "display_name": "Owner",
                "email": "owner@example.test",
                "password": "a-long-safe-password",
            },
        ).status_code
        == 302
    )
    assert client.get("/setup").status_code == 404


def test_real_cache_import_compatibility_without_persisting_user_data(tmp_path):
    # Optional local check; never embeds the real data in the repository or test output.
    source = Path(__file__).resolve().parents[1] / "data" / "garmin"
    dirs = [p for p in source.glob("*") if p.is_dir()]
    if not dirs:
        pytest.skip("No local Garmin cache")
    app = create_app(
        {
            "TESTING": True,
            "SQLALCHEMY_DATABASE_URI": "sqlite://",
            "SECRET_KEY": "test",
            "RATELIMIT_ENABLED": False,
        }
    )
    with app.app_context():
        db.create_all()
        user = User(
            username="test",
            display_name="Test",
            email="test@example.test",
            password_hash="not-a-login",
            local_account_key=dirs[0].name,
        )
        db.session.add(user)
        db.session.flush()
        count = import_cache(user, source)
        db.session.commit()
        assert count > 0
        assert import_cache(user, source) == 0
        for row in db.session.scalars(db.select(Activity)):
            assert row.distance_m >= 0 and row.duration_s >= 0
            assert row.visibility == "private"


def test_refresh_preserves_sharing_and_converts_running_units(app):
    from app.services import activity_from_fit

    doc = {
        "session": [{"sport": "running", "start_time": "2026-09-30T10:00:00Z"}],
        "record": [
            {
                "timestamp": "2026-09-30T10:00:00Z",
                "cadence": 80,
                "fractional_cadence": 0.5,
                "step_length": 1100,
                "vertical_oscillation": 85,
                "stance_time": 250,
                "power": 300,
            }
        ],
    }
    with app.app_context():
        activity = db.session.get(Activity, 1)
        activity.title = "My chosen title"
        activity.visibility = "friends"
        result = activity_from_fit(1, "test:1", [doc], refresh=True)
        db.session.commit()
        assert result.id == 1
        assert result.title == "My chosen title"
        assert result.visibility == "friends"
        assert result.samples[0]["cadence"] == 161
        assert result.samples[0]["stride_length"] == 1.1
        assert result.samples[0]["vertical_oscillation"] == 8.5
        assert result.samples[0]["ground_contact"] == 250
        assert result.samples[0]["power"] == 300