Deploy Stride with encrypted manual and scheduled Garmin sync

tests/test_web_sync.py

@@ -0,0 +1,116 @@
+import io
+import json
+from datetime import UTC, datetime, timedelta
+from unittest.mock import Mock
+
+import pytest
+from cryptography.fernet import Fernet
+from garminconnect import GarminConnectAuthenticationError
+
+from app import create_app
+from app.extensions import db
+from app.models import GarminConnection, User
+from app.sync import run_due, session_lock
+
+
+@pytest.fixture
+def connected(tmp_path, monkeypatch):
+    app = create_app(
+        {
+            "TESTING": True,
+            "SQLALCHEMY_DATABASE_URI": "sqlite://",
+            "SECRET_KEY": "testing",
+            "WTF_CSRF_ENABLED": False,
+            "RATELIMIT_ENABLED": False,
+            "LOCAL_SETUP": False,
+            "TOKEN_ENCRYPTION_KEY": Fernet.generate_key().decode(),
+            "DATA_ROOT": str(tmp_path / "data"),
+        }
+    )
+    app.instance_path = str(tmp_path / "instance")
+    api = Mock(profile_id=123)
+    api.client.dumps.return_value = '{"rotated":"private-session"}'
+    monkeypatch.setattr("app.sync.Garmin", lambda: api)
+    sync = Mock(return_value={"failed": 0})
+    monkeypatch.setattr("app.sync.sync_activities", sync)
+    with app.app_context():
+        db.create_all()
+        user = User(username="owner", email="owner@example.test", display_name="Owner")
+        user.set_password("long-test-password")
+        db.session.add(user)
+        db.session.commit()
+    client = app.test_client()
+    client.post("/login", data={"email": "owner@example.test", "password": "long-test-password"})
+    payload = json.dumps(
+        {k: "secret-value" for k in ("di_token", "di_refresh_token", "di_client_id")}
+    )
+    response = client.post(
+        "/garmin/connect", data={"tokens": (io.BytesIO(payload.encode()), "session.json")}
+    )
+    assert response.status_code == 302
+    return app, client, api, sync
+
+
+def test_manual_scheduled_encryption_and_disconnect(connected):
+    app, client, _api, sync = connected
+    with app.app_context():
+        connection = db.session.scalar(db.select(GarminConnection))
+        assert "private-session" not in connection.tokens
+        assert run_due() == 1
+        assert connection.last_success
+        assert not connection.pending
+        assert run_due() == 0
+        connection.next_sync = datetime.now(UTC) - timedelta(minutes=1)
+        db.session.commit()
+        assert run_due() == 1
+        assert sync.call_count == 2
+    client.post("/garmin/schedule", data={"enabled": "no"})
+    client.post("/garmin/sync")
+    with app.app_context():
+        assert run_due() == 1
+        assert sync.call_count == 3
+    client.post("/garmin/disconnect")
+    with app.app_context():
+        assert run_due() == 0
+        assert db.session.scalar(db.select(GarminConnection)) is None
+        assert db.session.get(User, 1).local_account_key is None
+
+
+def test_retry_and_rotated_tokens_survive_failure(connected):
+    app, client, _api, sync = connected
+    api = _api
+    sync.side_effect = RuntimeError("provider secret must not appear")
+    with app.app_context():
+        assert run_due() == 1
+        connection = db.session.scalar(db.select(GarminConnection))
+        assert connection.failures == 1
+        assert "provider secret" not in connection.message
+        assert (
+            Fernet(app.config["TOKEN_ENCRYPTION_KEY"]).decrypt(connection.tokens.encode())
+            == api.client.dumps.return_value.encode()
+        )
+        assert run_due() == 0
+    client.post("/garmin/sync")
+    api.login.side_effect = GarminConnectAuthenticationError("private error")
+    with app.app_context():
+        run_due()
+        connection = db.session.scalar(db.select(GarminConnection))
+        assert not connection.scheduled
+        assert "expired" in connection.message
+
+
+def test_lock_prevents_overlapping_sync_and_disconnect(connected):
+    app, client, _api, sync = connected
+    with app.app_context(), session_lock(1):
+        run_due()
+        sync.assert_not_called()
+        response = client.post("/garmin/disconnect", follow_redirects=True)
+        assert b"sync is running" in response.data
+        assert db.session.scalar(db.select(GarminConnection)) is not None
+
+
+def test_sync_requires_login_and_csrf(connected):
+    app, client, _api, _sync = connected
+    assert app.test_client().post("/garmin/sync").status_code == 302
+    app.config["WTF_CSRF_ENABLED"] = True
+    assert client.post("/garmin/sync").status_code == 400