117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
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
|