add astra draft
This commit is contained in:
@@ -0,0 +1,222 @@
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime, timezone
|
||||
import json
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
import pytest
|
||||
|
||||
from apc.api import Settings, create_app
|
||||
|
||||
TOKEN = "test-only-parent-token-with-more-than-32-characters"
|
||||
HEADERS = {"Authorization": f"Bearer {TOKEN}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env(tmp_path):
|
||||
clock = [datetime(2026, 9, 17, 10, tzinfo=timezone.utc).timestamp()]
|
||||
app = create_app(Settings(TOKEN, str(tmp_path / "test.db")), clock=lambda: clock[0])
|
||||
with TestClient(app) as client:
|
||||
yield client, app.state.store, clock
|
||||
|
||||
|
||||
def enroll(client, name="Example child"):
|
||||
child = client.post("/v1/parent/children", json={"name": name}, headers=HEADERS).json()
|
||||
code = client.post(f"/v1/parent/children/{child['id']}/pairing-code", headers=HEADERS).json()["code"]
|
||||
device = client.post("/v1/pair", json={"code": code, "name": "Test phone"}).json()
|
||||
return child, device, {"Authorization": f"Bearer {device['token']}"}
|
||||
|
||||
|
||||
def solve(challenge):
|
||||
result = []
|
||||
for question in challenge["questions"]:
|
||||
a, op, b = question["prompt"].split()
|
||||
result.append({"question_id": question["id"],
|
||||
"answer": int(a) + int(b) if op == "+" else int(a) - int(b)})
|
||||
return {"answers": result}
|
||||
|
||||
|
||||
def policy_update(client, **updates):
|
||||
f = client.get("/v1/parent/family", headers=HEADERS).json()
|
||||
return client.put("/v1/parent/policy", headers=HEADERS,
|
||||
json={"expected_revision": f["revision"], "policy": f["policy"] | updates})
|
||||
|
||||
|
||||
def test_parent_auth_and_device_scope(env):
|
||||
client, _, _ = env
|
||||
assert client.get("/v1/parent/family").status_code == 401
|
||||
child, _, headers = enroll(client)
|
||||
assert client.get("/v1/parent/family", headers=headers).status_code == 401
|
||||
assert client.get("/v1/device/snapshot", headers=HEADERS).status_code == 401
|
||||
assert client.get("/v1/device/snapshot", headers=headers).json()["child_id"] == child["id"]
|
||||
|
||||
|
||||
def test_inheritance_overrides_and_optimistic_concurrency(env):
|
||||
client, _, _ = env
|
||||
a, _, ha = enroll(client, "A")
|
||||
_, _, hb = enroll(client, "B")
|
||||
f = client.get("/v1/parent/family", headers=HEADERS).json()
|
||||
body = {"expected_revision": f["revision"], "overrides": {"daily_minutes": 45}}
|
||||
assert client.put(f"/v1/parent/children/{a['id']}/overrides", json=body, headers=HEADERS).status_code == 200
|
||||
assert client.put(f"/v1/parent/children/{a['id']}/overrides", json=body, headers=HEADERS).status_code == 409
|
||||
assert policy_update(client, daily_minutes=60, correct_answers=3).status_code == 200
|
||||
pa = client.get("/v1/device/snapshot", headers=ha).json()["policy"]
|
||||
pb = client.get("/v1/device/snapshot", headers=hb).json()["policy"]
|
||||
assert (pa["daily_minutes"], pb["daily_minutes"]) == (45, 60)
|
||||
assert pa["correct_answers"] == pb["correct_answers"] == 3
|
||||
revision = client.get("/v1/parent/family", headers=HEADERS).json()["revision"]
|
||||
assert client.put(f"/v1/parent/children/{a['id']}/overrides", headers=HEADERS,
|
||||
json={"expected_revision": revision, "overrides": {}}).status_code == 200
|
||||
assert client.get("/v1/device/snapshot", headers=ha).json()["policy"]["daily_minutes"] == 60
|
||||
|
||||
|
||||
def test_reward_is_server_graded_capped_and_idempotent(env):
|
||||
client, store, _ = env
|
||||
_, _, headers = enroll(client)
|
||||
assert policy_update(client, max_bonus_minutes=15).status_code == 200
|
||||
for expected in (10, 5):
|
||||
challenge = client.post("/v1/device/challenges", headers=headers).json()
|
||||
assert all("answer" not in q for q in challenge["questions"])
|
||||
same = client.post("/v1/device/challenges", headers=headers).json()
|
||||
assert same == challenge
|
||||
path = f"/v1/device/challenges/{challenge['id']}/submit"
|
||||
result = client.post(path, json=solve(challenge), headers=headers)
|
||||
assert result.json()["awarded_minutes"] == expected
|
||||
assert client.post(path, json=solve(challenge), headers=headers).json() == result.json()
|
||||
assert client.post("/v1/device/challenges", headers=headers).status_code == 409
|
||||
with store.transaction() as db:
|
||||
assert db.execute("SELECT COUNT(*), SUM(minutes) FROM rewards").fetchone()[:] == (2, 15)
|
||||
|
||||
|
||||
def test_concurrent_submission_issues_one_credit(env):
|
||||
client, store, _ = env
|
||||
_, _, headers = enroll(client)
|
||||
challenge = client.post("/v1/device/challenges", headers=headers).json()
|
||||
def submit(_):
|
||||
return client.post(f"/v1/device/challenges/{challenge['id']}/submit",
|
||||
headers=headers, json=solve(challenge)).json()
|
||||
with ThreadPoolExecutor(max_workers=6) as pool:
|
||||
results = list(pool.map(submit, range(12)))
|
||||
assert all(result == results[0] for result in results)
|
||||
with store.transaction() as db:
|
||||
assert db.execute("SELECT COUNT(*) FROM rewards").fetchone()[0] == 1
|
||||
|
||||
|
||||
def test_bad_answers_fail_and_cannot_be_retried_into_credit(env):
|
||||
client, _, _ = env
|
||||
_, _, headers = enroll(client)
|
||||
ch = client.post("/v1/device/challenges", headers=headers).json()
|
||||
bad = solve(ch)
|
||||
bad["answers"][0]["answer"] += 1
|
||||
path = f"/v1/device/challenges/{ch['id']}/submit"
|
||||
result = client.post(path, headers=headers, json=bad).json()
|
||||
assert not result["passed"] and result["awarded_minutes"] == 0
|
||||
assert client.post(path, headers=headers, json=solve(ch)).json() == result
|
||||
|
||||
|
||||
def test_duplicate_question_and_cross_child_submission_rejected(env):
|
||||
client, _, _ = env
|
||||
_, _, ha = enroll(client, "A")
|
||||
_, _, hb = enroll(client, "B")
|
||||
ch = client.post("/v1/device/challenges", headers=ha).json()
|
||||
path = f"/v1/device/challenges/{ch['id']}/submit"
|
||||
answers = solve(ch)
|
||||
assert client.post(path, headers=hb, json=answers).status_code == 404
|
||||
answers["answers"][1] = answers["answers"][0]
|
||||
assert client.post(path, headers=ha, json=answers).status_code == 422
|
||||
|
||||
|
||||
def test_expiry_policy_change_and_local_midnight(env):
|
||||
client, _, clock = env
|
||||
_, _, h = enroll(client)
|
||||
def start(): return client.post("/v1/device/challenges", headers=h).json()
|
||||
def submit(ch): return client.post(f"/v1/device/challenges/{ch['id']}/submit", headers=h, json=solve(ch))
|
||||
ch = start(); clock[0] += 601
|
||||
assert submit(ch).status_code == 409
|
||||
ch = start(); policy_update(client, daily_minutes=40)
|
||||
assert submit(ch).status_code == 409
|
||||
ch = start(); assert submit(ch).json()["awarded_minutes"] == 10
|
||||
clock[0] = datetime(2026, 9, 17, 21, 59, tzinfo=timezone.utc).timestamp()
|
||||
ch = start(); clock[0] += 120 # Cross Berlin midnight, not UTC midnight.
|
||||
assert submit(ch).status_code == 409
|
||||
snapshot = client.get("/v1/device/snapshot", headers=h).json()
|
||||
assert snapshot["day"] == "2026-09-18" and snapshot["earned_minutes"] == 0
|
||||
|
||||
|
||||
def test_pairing_single_use_expiry_revocation_and_configuration_scope(env):
|
||||
client, _, clock = env
|
||||
a, device, h = enroll(client)
|
||||
assert client.post(f"/v1/parent/children/{a['id']}/pairing-code", headers=HEADERS).status_code == 409
|
||||
code = client.post(f"/v1/parent/devices/{device['device_id']}/configuration-code", headers=HEADERS).json()["code"]
|
||||
_, _, other = enroll(client, "Other")
|
||||
assert client.post("/v1/device/configuration-unlock", json={"code": code}, headers=other).status_code == 403
|
||||
assert client.post("/v1/device/configuration-unlock", json={"code": code}, headers=h).status_code == 200
|
||||
assert client.post("/v1/device/configuration-unlock", json={"code": code}, headers=h).status_code == 403
|
||||
client.delete(f"/v1/parent/devices/{device['device_id']}", headers=HEADERS)
|
||||
assert client.get("/v1/device/snapshot", headers=h).status_code == 401
|
||||
code = client.post(f"/v1/parent/children/{a['id']}/pairing-code", headers=HEADERS).json()["code"]
|
||||
clock[0] += 601
|
||||
assert client.post("/v1/pair", json={"code": code, "name": "New phone"}).status_code == 403
|
||||
|
||||
|
||||
def test_cookie_auth_csrf_headers_and_logout(env):
|
||||
client, _, _ = env
|
||||
result = client.post("/v1/session", json={"token": TOKEN}, headers={"Origin": "http://localhost:8000"})
|
||||
assert "HttpOnly" in result.headers["set-cookie"] and "SameSite=strict" in result.headers["set-cookie"]
|
||||
assert client.get("/v1/parent/family").status_code == 200
|
||||
assert client.post("/v1/parent/children", json={"name": "Test"}).status_code == 403
|
||||
assert client.post("/v1/parent/children", json={"name": "Test"}, headers={"Origin": "https://evil.test"}).status_code == 403
|
||||
assert client.delete("/v1/session", headers={"Origin": "http://localhost:8000"}).status_code == 200
|
||||
assert client.get("/v1/parent/family").status_code == 401
|
||||
assert "frame-ancestors 'none'" in client.get("/").headers["content-security-policy"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("update", [{"daily_minutes": -1}, {"daily_minutes": True},
|
||||
{"timezone": "Not/AZone"}, {"allowed_start": 1300},
|
||||
{"daily_minutes": "30"}, {"unexpected": 123}])
|
||||
def test_invalid_policy_rejected(env, update):
|
||||
assert policy_update(env[0], **update).status_code == 422
|
||||
|
||||
|
||||
def test_status_is_acknowledgement_not_server_usage_claim(env):
|
||||
client, store, _ = env
|
||||
_, device, headers = enroll(client)
|
||||
status = {"policy_revision": 1, "state": "not_configured", "shielded": True,
|
||||
"detail": "Select apps with guardian present", "consumed_lower_bound": 0,
|
||||
"usage_day": "2026-09-17"}
|
||||
assert client.post("/v1/device/status", headers=headers, json=status).status_code == 200
|
||||
family = client.get("/v1/parent/family", headers=HEADERS).json()
|
||||
assert family["children"][0]["device"]["status"] == status
|
||||
with store.transaction() as db:
|
||||
stored = db.execute("SELECT token_hash FROM devices WHERE id=?", (device["device_id"],)).fetchone()[0]
|
||||
assert stored != device["token"] and len(stored) == 64
|
||||
|
||||
|
||||
def test_configuration_validation_and_sqlite_persistence(env):
|
||||
client, store, _ = env
|
||||
_, _, headers = enroll(client)
|
||||
ch = client.post("/v1/device/challenges", headers=headers).json()
|
||||
client.post(f"/v1/device/challenges/{ch['id']}/submit", headers=headers, json=solve(ch))
|
||||
assert policy_update(client, timezone="UTC").status_code == 409
|
||||
with store.transaction() as db:
|
||||
assert json.loads(db.execute("SELECT result FROM challenges").fetchone()[0])["passed"]
|
||||
with pytest.raises(ValueError): Settings("too short")
|
||||
with pytest.raises(ValueError): Settings(TOKEN, origin="http://public.example")
|
||||
|
||||
|
||||
def test_session_rotation_and_restart_preserve_ledger(env):
|
||||
client, store, clock = env
|
||||
_, _, headers = enroll(client)
|
||||
ch = client.post("/v1/device/challenges", headers=headers).json()
|
||||
client.post(f"/v1/device/challenges/{ch['id']}/submit", headers=headers, json=solve(ch))
|
||||
client.post("/v1/session", json={"token": TOKEN})
|
||||
cookies = dict(client.cookies)
|
||||
restarted = create_app(Settings(TOKEN + "-rotated", store.path), clock=lambda: clock[0])
|
||||
with TestClient(restarted) as next_client:
|
||||
next_client.cookies.update(cookies)
|
||||
assert next_client.get("/v1/parent/family").status_code == 401
|
||||
assert next_client.get("/v1/device/snapshot", headers=headers).json()["earned_minutes"] == 10
|
||||
|
||||
|
||||
@pytest.mark.parametrize("length,expected", [("invalid", 400), ("-1", 400), ("40000", 413)])
|
||||
def test_request_length_validation(env, length, expected):
|
||||
assert env[0].post("/v1/pair", headers={"Content-Length": length}, content="{}").status_code == expected
|
||||
Reference in New Issue
Block a user