869 lines
32 KiB
Python
869 lines
32 KiB
Python
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"):
|
|
response = client.post(
|
|
"/login", data={"email": f"{name}@example.test", "password": "a-long-test-password"}
|
|
)
|
|
assert response.status_code == 302, response.data.decode()
|
|
return response
|
|
|
|
|
|
def add_history(count, visibility="private"):
|
|
from datetime import timedelta
|
|
|
|
for index in range(count):
|
|
db.session.add(
|
|
Activity(
|
|
owner_id=1,
|
|
source_key=f"history:{index}",
|
|
title=f"History run {index:03d}",
|
|
sport="running",
|
|
started_at=datetime(2026, 9, 1, tzinfo=UTC) + timedelta(days=index),
|
|
distance_m=1000,
|
|
duration_s=300,
|
|
elevation_m=1,
|
|
visibility=visibility,
|
|
efforts_version=EFFORTS_VERSION,
|
|
best_efforts=[
|
|
{
|
|
"label": "1 km",
|
|
"distance": 1000,
|
|
"seconds": 250 if index == 0 else 300,
|
|
"start_m": 0,
|
|
"end_m": 1000,
|
|
}
|
|
],
|
|
samples=[],
|
|
laps=[],
|
|
)
|
|
)
|
|
db.session.commit()
|
|
|
|
|
|
def test_history_pages_keep_full_totals_records_and_latest_route(app):
|
|
import re
|
|
|
|
from flask import template_rendered
|
|
|
|
with app.app_context():
|
|
add_history(60)
|
|
contexts = []
|
|
|
|
def capture(_app, template, context, **_extra):
|
|
contexts.append(context)
|
|
|
|
template_rendered.connect(capture, app)
|
|
client = app.test_client()
|
|
login(client)
|
|
try:
|
|
seen = []
|
|
for page, size in ((1, 25), (2, 25), (3, 11)):
|
|
response = client.get(f"/dashboard?page={page}")
|
|
assert response.status_code == 200
|
|
context = contexts[-1]
|
|
assert len(context["rows"]) == size
|
|
assert context["stats"]["count"] == 61
|
|
assert context["stats"]["distance"] == 65
|
|
assert context["stats"]["duration"] == 19800
|
|
assert context["stats"]["elevation"] == 100
|
|
assert context["mapped_run"].id == 1
|
|
assert context["record_book"]["records"][0]["seconds"] == 250
|
|
assert set(context["record_book"]["runs"]).issubset({row.id for row in context["rows"]})
|
|
seen.extend(
|
|
int(i)
|
|
for i in re.findall(
|
|
rb'class="activity-name" href="/activities/(\d+)"', response.data
|
|
)
|
|
)
|
|
assert len(seen) == len(set(seen)) == 61
|
|
assert b"Showing 51" in response.data
|
|
assert b"1 km PR" in response.data
|
|
assert client.get("/dashboard?page=4").status_code == 404
|
|
for value in ("0", "-1", "nope", "999999999999999999999", "1.5"):
|
|
assert client.get(f"/dashboard?page={value}").status_code == 404
|
|
finally:
|
|
template_rendered.disconnect(capture, app)
|
|
|
|
|
|
def test_pagination_uses_id_to_break_equal_timestamp_ties(app):
|
|
with app.app_context():
|
|
add_history(50)
|
|
db.session.execute(db.update(Activity).values(started_at=datetime(2026, 1, 1, tzinfo=UTC)))
|
|
db.session.commit()
|
|
import re
|
|
|
|
client = app.test_client()
|
|
login(client)
|
|
ids = []
|
|
for page in (1, 2, 3):
|
|
ids.extend(
|
|
int(i)
|
|
for i in re.findall(
|
|
rb'class="activity-name" href="/activities/(\d+)"',
|
|
client.get(f"/dashboard?page={page}").data,
|
|
)
|
|
)
|
|
assert ids == list(reversed(range(1, 52)))
|
|
|
|
|
|
def test_profile_and_friends_pages_include_only_visible_history(app):
|
|
from flask import template_rendered
|
|
|
|
with app.app_context():
|
|
add_history(53, visibility="friends")
|
|
db.session.add(Friendship(low_id=1, high_id=2, requested_by=1, status="accepted"))
|
|
db.session.commit()
|
|
contexts = []
|
|
|
|
def capture(_app, template, context, **_extra):
|
|
contexts.append(context)
|
|
|
|
template_rendered.connect(capture, app)
|
|
client = app.test_client()
|
|
login(client, "other")
|
|
try:
|
|
for path in ("/u/runner", "/friends"):
|
|
for page, expected in ((1, 25), (2, 25), (3, 3)):
|
|
response = client.get(f"{path}?page={page}")
|
|
assert response.status_code == 200
|
|
assert len(contexts[-1]["rows"]) == expected
|
|
assert contexts[-1]["pagination"].total == 53
|
|
assert b"A quiet morning run" not in response.data
|
|
if path == "/u/runner":
|
|
assert contexts[-1]["stats"]["distance"] == 53
|
|
client.post("/friends/1/remove")
|
|
assert client.get("/u/runner?page=2").status_code == 404
|
|
assert client.get("/friends?page=2").status_code == 404
|
|
finally:
|
|
template_rendered.disconnect(capture, app)
|
|
|
|
|
|
def test_weekly_sql_totals_respect_sydney_dst_boundaries(app):
|
|
from app.history import overview
|
|
|
|
with app.app_context():
|
|
for index, (started, distance) in enumerate(
|
|
(
|
|
(datetime(2026, 10, 4, 12, 59, 59, tzinfo=UTC), 1000),
|
|
(datetime(2026, 10, 4, 13, 0, 0, tzinfo=UTC), 2000),
|
|
)
|
|
):
|
|
db.session.add(
|
|
Activity(
|
|
owner_id=2,
|
|
source_key=f"dst:{index}",
|
|
title="DST run",
|
|
started_at=started,
|
|
distance_m=distance,
|
|
duration_s=600,
|
|
)
|
|
)
|
|
db.session.commit()
|
|
stats = overview([Activity.owner_id == 2], "Australia/Sydney")
|
|
assert stats["weeks"][-2:] == ["28 Sep", "05 Oct"]
|
|
assert stats["weekly_km"][-2:] == [1, 2]
|
|
assert stats["distance"] == 3
|
|
assert stats["count"] == 2
|
|
empty = overview([Activity.owner_id == 999], "Australia/Sydney")
|
|
assert empty["distance"] == 0
|
|
assert empty["pace"] == "—"
|
|
assert empty["weekly_km"] == [0] * 8
|
|
|
|
|
|
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
|
|
from app.display import ANALYSIS_VERSION
|
|
|
|
assert row.analysis_version == ANALYSIS_VERSION
|
|
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
|
|
|
|
|
|
def test_friend_dashboard_records_and_totals_exclude_private_runs(app):
|
|
from flask import template_rendered
|
|
|
|
with app.app_context():
|
|
private = db.session.get(Activity, 1)
|
|
private.samples = [{"distance": 0, "time": 0}, {"distance": 5000, "time": 1000}]
|
|
persist_efforts(private)
|
|
shared = Activity(
|
|
owner_id=1,
|
|
source_key="shared-record",
|
|
title="Shared slower run",
|
|
sport="running",
|
|
visibility="friends",
|
|
has_route=True,
|
|
started_at=datetime(2026, 10, 1, tzinfo=UTC),
|
|
distance_m=5000,
|
|
duration_s=1500,
|
|
elevation_m=20,
|
|
samples=[{"distance": 0, "time": 0}, {"distance": 5000, "time": 1500}],
|
|
)
|
|
persist_efforts(shared)
|
|
shared.has_route = True
|
|
db.session.add(shared)
|
|
db.session.add(Friendship(low_id=1, high_id=2, requested_by=1, status="accepted"))
|
|
db.session.commit()
|
|
contexts = []
|
|
|
|
def capture(_app, template, context, **_extra):
|
|
contexts.append(context)
|
|
|
|
template_rendered.connect(capture, app)
|
|
client = app.test_client()
|
|
login(client, "other")
|
|
try:
|
|
response = client.get("/u/runner")
|
|
assert response.status_code == 200
|
|
context = contexts[-1]
|
|
assert context["stats"]["count"] == 1
|
|
assert context["stats"]["distance"] == 5
|
|
assert context["stats"]["duration"] == 1500
|
|
assert context["mapped_run"].title == "Shared slower run"
|
|
assert all(
|
|
record["activity"].title == "Shared slower run"
|
|
for record in context["record_book"]["records"]
|
|
)
|
|
assert context["record_book"]["records"][-1]["seconds"] == 1500
|
|
assert b"A quiet morning run" not in response.data
|
|
assert b"header-import" not in response.data
|
|
assert b"Shared profile" not in response.data
|
|
assert b'<aside class="sidebar"' not in response.data
|
|
assert b"Friends \xc2\xb7 Shared runs" in response.data
|
|
assert client.post("/activities/2/edit", data={"title": "changed"}).status_code == 404
|
|
finally:
|
|
template_rendered.disconnect(capture, app)
|
|
owner = app.test_client()
|
|
login(owner)
|
|
assert owner.get("/u/runner").location.endswith("/dashboard")
|