129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""Sprint 3.6 signed export contract tests."""
|
|
from __future__ import annotations
|
|
|
|
|
|
def _headers(token: str) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def _setup_admin(user_store) -> None:
|
|
user_store.complete_setup(
|
|
"admin",
|
|
"admin@example.com",
|
|
"admin-ready-password",
|
|
accepted_terms=True,
|
|
accepted_terms_at="2026-08-15T00:00:00Z",
|
|
)
|
|
|
|
|
|
def _issue_export_link(client, token: str) -> dict:
|
|
response = client.get("/api/analytics/export/token", headers=_headers(token))
|
|
assert response.status_code == 200, response.get_json()
|
|
return response.get_json()
|
|
|
|
|
|
def test_signed_export_works_without_bearer_and_replay_is_denied(
|
|
client, user_store, login
|
|
):
|
|
_setup_admin(user_store)
|
|
bearer = login("admin", "admin-ready-password")["token"]
|
|
link = _issue_export_link(client, bearer)
|
|
|
|
first = client.get(link["url"])
|
|
replay = client.get(link["url"])
|
|
|
|
assert first.status_code == 200, first.get_json()
|
|
assert first.headers["Content-Type"].startswith("text/csv")
|
|
assert replay.status_code == 403
|
|
|
|
from app.api import analytics_routes
|
|
|
|
jti = link["token"].split(".", 1)[1].rsplit(":", 1)[-1]
|
|
record = analytics_routes._export_token_store().get(jti)
|
|
assert record["used"] is True
|
|
|
|
|
|
def test_signed_export_rejects_expired_and_tampered_links(
|
|
client, user_store, login, monkeypatch
|
|
):
|
|
_setup_admin(user_store)
|
|
bearer = login("admin", "admin-ready-password")["token"]
|
|
|
|
from app.api import analytics_routes
|
|
|
|
monkeypatch.setattr(analytics_routes, "_EXPORT_TTL", -1)
|
|
expired = _issue_export_link(client, bearer)
|
|
assert client.get(expired["url"]).status_code == 403
|
|
|
|
monkeypatch.setattr(analytics_routes, "_EXPORT_TTL", 300)
|
|
fresh = _issue_export_link(client, bearer)
|
|
tampered_token = fresh["token"][:-1] + ("a" if fresh["token"][-1] != "a" else "b")
|
|
tampered = client.get(f"/api/analytics/export?token={tampered_token}")
|
|
|
|
assert tampered.status_code == 403
|
|
|
|
|
|
def test_signed_export_rejects_wrong_current_bearer_tenant(
|
|
client, user_store, login
|
|
):
|
|
_setup_admin(user_store)
|
|
admin_bearer = login("admin", "admin-ready-password")["token"]
|
|
link = _issue_export_link(client, admin_bearer)
|
|
|
|
other_org = user_store.create_org("Other tenant", org_id="org-other")
|
|
other = user_store.create_user(
|
|
org_id=other_org["id"],
|
|
username="other-admin",
|
|
password="other-admin-password",
|
|
name="Other Admin",
|
|
role="admin",
|
|
email="other-admin@example.com",
|
|
must_setup=False,
|
|
)
|
|
other_bearer = login(other["username"], "other-admin-password")["token"]
|
|
|
|
response = client.get(link["url"], headers=_headers(other_bearer))
|
|
|
|
assert response.status_code == 403
|
|
assert client.get(link["url"]).status_code == 200
|
|
|
|
|
|
def test_signed_export_failure_does_not_burn_link(client, user_store, login, monkeypatch):
|
|
_setup_admin(user_store)
|
|
bearer = login("admin", "admin-ready-password")["token"]
|
|
link = _issue_export_link(client, bearer)
|
|
|
|
from app.api import analytics_routes
|
|
original_export = analytics_routes._export_csv_for_actor
|
|
|
|
def fail_export(*_args, **_kwargs):
|
|
raise RuntimeError("simulated export failure")
|
|
|
|
monkeypatch.setattr(analytics_routes, "_export_csv_for_actor", fail_export)
|
|
failed = client.get(link["url"])
|
|
assert failed.status_code == 500
|
|
|
|
monkeypatch.setattr(analytics_routes, "_export_csv_for_actor", original_export)
|
|
assert client.get(link["url"]).status_code == 200
|
|
|
|
|
|
def test_normal_bearer_export_still_works(client, user_store, login):
|
|
_setup_admin(user_store)
|
|
bearer = login("admin", "admin-ready-password")["token"]
|
|
|
|
response = client.get("/api/analytics/export", headers=_headers(bearer))
|
|
|
|
assert response.status_code == 200, response.get_json()
|
|
assert response.headers["Content-Type"].startswith("text/csv")
|
|
|
|
|
|
def test_csv_cells_escape_spreadsheet_formula_prefixes():
|
|
from app.api.analytics_routes import _csv_cell
|
|
|
|
assert _csv_cell("=SUM(A1:A2)") == "'=SUM(A1:A2)"
|
|
assert _csv_cell("+1") == "'+1"
|
|
assert _csv_cell("-1") == "'-1"
|
|
assert _csv_cell("@user") == "'@user"
|
|
assert _csv_cell("safe") == "safe"
|
|
assert _csv_cell(None) == ""
|