Lars
72d101415f
CI container only installs pytest; use asyncio.run() for async route handlers. Co-authored-by: Cursor <cursoragent@cursor.com>
111 lines
3.3 KiB
Python
111 lines
3.3 KiB
Python
"""Feature-Enforcement für Legacy Activity CSV-Import (#37)."""
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from routers.activity import import_activity_csv
|
|
|
|
|
|
def test_import_activity_csv_blocks_when_limit_reached():
|
|
upload = AsyncMock()
|
|
upload.read = AsyncMock(return_value=b"Workout Type,Start\n")
|
|
|
|
with patch("routers.activity.get_pid", return_value="profile-1"), patch(
|
|
"routers.activity.check_feature_access",
|
|
return_value={
|
|
"allowed": False,
|
|
"reason": "limit_exceeded",
|
|
"used": 30,
|
|
"limit": 30,
|
|
},
|
|
), patch("routers.activity.log_feature_usage"):
|
|
with pytest.raises(HTTPException) as exc:
|
|
asyncio.run(
|
|
import_activity_csv(
|
|
file=upload,
|
|
x_profile_id=None,
|
|
session={"profile_id": "profile-1"},
|
|
)
|
|
)
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
def test_import_activity_csv_increments_usage_per_insert(monkeypatch):
|
|
upload = AsyncMock()
|
|
upload.read = AsyncMock(
|
|
return_value=(
|
|
"Workout Type,Start,Duration,Aktive Energie (kJ),Ruheeinträge (kJ),"
|
|
"Durchschn. Herzfrequenz (count/min),Max. Herzfrequenz (count/min),Distanz (km),End\n"
|
|
"Running,2026-07-22 10:00:00 +0200,0:45:00,1000,500,140,160,5.0,\n"
|
|
).encode("utf-8")
|
|
)
|
|
|
|
increments = []
|
|
|
|
class FakeCursor:
|
|
def execute(self, *args, **kwargs):
|
|
return None
|
|
|
|
def fetchone(self):
|
|
return None
|
|
|
|
class FakeConn:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
fake_cur = FakeCursor()
|
|
|
|
monkeypatch.setattr("routers.activity.get_pid", lambda _h: "profile-1")
|
|
monkeypatch.setattr(
|
|
"routers.activity.check_feature_access",
|
|
lambda *_a, **_k: {"allowed": True, "reason": "ok", "used": 0, "limit": 100},
|
|
)
|
|
monkeypatch.setattr("routers.activity.log_feature_usage", lambda *_a, **_k: None)
|
|
monkeypatch.setattr("routers.activity.get_db", lambda: FakeConn())
|
|
monkeypatch.setattr("routers.activity.get_cursor", lambda _c: fake_cur)
|
|
monkeypatch.setattr(
|
|
"routers.activity.normalize_activity_start",
|
|
lambda _s: ("2026-07-22", "10:00:00"),
|
|
)
|
|
monkeypatch.setattr(
|
|
"routers.activity.get_training_type_for_activity",
|
|
lambda *_a, **_k: (1, "cardio", None),
|
|
)
|
|
monkeypatch.setattr(
|
|
"routers.activity.find_activity_duplicate_id",
|
|
lambda *_a, **_k: None,
|
|
)
|
|
monkeypatch.setattr(
|
|
"routers.activity.new_activity_id",
|
|
lambda: "new-id",
|
|
)
|
|
monkeypatch.setattr(
|
|
"routers.activity.insert_activity_csv_minimal",
|
|
lambda *_a, **_k: None,
|
|
)
|
|
monkeypatch.setattr(
|
|
"routers.activity.run_activity_post_write_hooks_import",
|
|
lambda *_a, **_k: None,
|
|
)
|
|
monkeypatch.setattr(
|
|
"routers.activity.increment_feature_usage",
|
|
lambda _pid, feature: increments.append(feature),
|
|
)
|
|
|
|
result = asyncio.run(
|
|
import_activity_csv(
|
|
file=upload,
|
|
x_profile_id=None,
|
|
session={"profile_id": "profile-1"},
|
|
)
|
|
)
|
|
|
|
assert result["inserted"] == 1
|
|
assert increments == ["activity_entries"]
|