340 lines
13 KiB
Python
340 lines
13 KiB
Python
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app import create_app
|
|
from app.extensions import db
|
|
from app.models import Activity, Friendship, User
|
|
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",
|
|
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()
|
|
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_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/data"' in page
|
|
assert b'id="toggle-route-replay"' 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():
|
|
db.session.get(Activity, 1).samples = []
|
|
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(samples=[{"point": point} for point in points])
|
|
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"] == "no-referrer"
|
|
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
|