feat: build Stride local app with Garmin imports and friends-only sharing
This commit is contained in:
@@ -0,0 +1,225 @@
|
||||
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_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_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"
|
||||
Reference in New Issue
Block a user