Implement WP-0009-T04: Control Plane interactive UI on whynot-design
Builds the Control Plane's browser UI (login, dashboard, Phase registration, Development Credit entry/proposal/review, credential admin, audit log) as a FastAPI + Jinja2 app over the already-finished T03 backend, rather than from scratch — whynot-design's Lit web components are vendored as static assets (source commit 4b62cffc, v0.4.1), with lit itself resolved via an esm.sh CDN import map. Session auth re-checks the credential token against the database on every request rather than trusting the session cookie's cached rights, so a mid-session revocation takes effect immediately. 9 new Docker-gated HTTP-level tests via FastAPI's TestClient (no browser-automation tool available, so real rendering of the <wn-*> components was never visually verified). All four WP-0009 tasks are now done; workplan marked finished. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
5fbae0df02
commit
c89b4aa4a5
24 changed files with 3486 additions and 3 deletions
289
tests/test_control_plane_app.py
Normal file
289
tests/test_control_plane_app.py
Normal file
|
|
@ -0,0 +1,289 @@
|
|||
"""HTTP-level integration tests for the Control Plane interactive UI
|
||||
(WP-0009-T04), via FastAPI's TestClient.
|
||||
|
||||
Same ephemeral, disposable Postgres-via-Docker pattern as
|
||||
`test_control_plane.py` (never the shared state-hub instance).
|
||||
|
||||
Scope disclosure: these tests exercise the FastAPI app's routing, form
|
||||
handling, session auth, and rights gating at the HTTP layer — they do not
|
||||
render or interact with the pages in a real browser, since no
|
||||
browser-automation tool is available in this environment. The
|
||||
whynot-design web components (`<wn-*>` custom elements) are not
|
||||
themselves exercised; only the server-rendered HTML/session/redirect
|
||||
behavior around them is.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
psycopg = pytest.importorskip("psycopg")
|
||||
pytest.importorskip("jinja2")
|
||||
pytest.importorskip("itsdangerous")
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
MIGRATIONS = [
|
||||
REPO_ROOT / "migrations" / "0001_registries.sql",
|
||||
REPO_ROOT / "migrations" / "0002_ledger.sql",
|
||||
REPO_ROOT / "migrations" / "0003_attestations.sql",
|
||||
REPO_ROOT / "migrations" / "0004_breach_records.sql",
|
||||
REPO_ROOT / "migrations" / "0005_licensor_credentials.sql",
|
||||
REPO_ROOT / "migrations" / "0006_control_plane.sql",
|
||||
]
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
shutil.which("docker") is None, reason="docker not available"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def pg_container():
|
||||
name = f"trf-test-pg-cpapp-{uuid.uuid4().hex[:8]}"
|
||||
subprocess.run(
|
||||
[
|
||||
"docker", "run", "--rm", "-d",
|
||||
"--name", name,
|
||||
"-e", "POSTGRES_PASSWORD=postgres",
|
||||
"-e", "POSTGRES_DB=target_revenue_test",
|
||||
"-p", "127.0.0.1::5432",
|
||||
"postgres:16-alpine",
|
||||
],
|
||||
check=True, capture_output=True,
|
||||
)
|
||||
try:
|
||||
port_out = subprocess.run(
|
||||
["docker", "port", name, "5432/tcp"], check=True, capture_output=True, text=True
|
||||
).stdout.strip()
|
||||
host_port = port_out.split(":")[-1]
|
||||
dsn = f"host=127.0.0.1 port={host_port} dbname=target_revenue_test user=postgres password=postgres"
|
||||
|
||||
for _ in range(60):
|
||||
try:
|
||||
with psycopg.connect(dsn, connect_timeout=1):
|
||||
break
|
||||
except psycopg.OperationalError:
|
||||
time.sleep(0.5)
|
||||
else:
|
||||
raise RuntimeError("postgres container did not become ready in time")
|
||||
|
||||
with psycopg.connect(dsn) as admin_conn:
|
||||
for migration in MIGRATIONS:
|
||||
admin_conn.execute(migration.read_text(encoding="utf-8"))
|
||||
admin_conn.commit()
|
||||
admin_conn.execute(
|
||||
"INSERT INTO licensors (token, licensor_id, credential_label, rights, issued_by) "
|
||||
"VALUES (%s, %s, %s, %s, %s)",
|
||||
("founding-admin-token", "binky", "founder", "admin", "bootstrap"),
|
||||
)
|
||||
admin_conn.commit()
|
||||
|
||||
app_dsn = (
|
||||
f"host=127.0.0.1 port={host_port} dbname=target_revenue_test "
|
||||
f"user=trf_app password=changeme-in-deployment"
|
||||
)
|
||||
yield {"admin_dsn": dsn, "app_dsn": app_dsn, "admin_token": "founding-admin-token"}
|
||||
finally:
|
||||
subprocess.run(["docker", "stop", name], capture_output=True)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client(pg_container, monkeypatch):
|
||||
monkeypatch.setenv("TRF_DATABASE_URL", pg_container["app_dsn"])
|
||||
monkeypatch.setenv("TRF_CONTROL_PLANE_SECRET_KEY", "test-secret-key")
|
||||
monkeypatch.setenv("TRF_SIGNING_KEY_HEX", "11" * 32)
|
||||
|
||||
import importlib
|
||||
|
||||
from target_revenue.service import control_plane_app as mod
|
||||
|
||||
importlib.reload(mod)
|
||||
if hasattr(mod.app.state, "pool"):
|
||||
mod.app.state.pool.close()
|
||||
del mod.app.state.pool
|
||||
if hasattr(mod.app.state, "signing_key"):
|
||||
del mod.app.state.signing_key
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
with TestClient(mod.app) as test_client:
|
||||
yield test_client
|
||||
|
||||
if hasattr(mod.app.state, "pool"):
|
||||
mod.app.state.pool.close()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def conn(pg_container):
|
||||
with psycopg.connect(pg_container["app_dsn"]) as connection:
|
||||
yield connection
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def credentials(conn, pg_container):
|
||||
from target_revenue import registry
|
||||
|
||||
admin = registry.authenticate(conn, pg_container["admin_token"])
|
||||
suffix = uuid.uuid4().hex[:8]
|
||||
tiers = {}
|
||||
for label, rights in [
|
||||
("viewer-user", "viewer"),
|
||||
("contributor-user", "contributor"),
|
||||
("operator-user", "operator"),
|
||||
]:
|
||||
cred = registry.issue_sub_credential(
|
||||
conn, licensor_id="binky", credential_label=f"{label}-{suffix}", rights=rights,
|
||||
issued_by="founder",
|
||||
)
|
||||
tiers[rights] = cred
|
||||
conn.commit()
|
||||
tiers["admin"] = admin
|
||||
return tiers
|
||||
|
||||
|
||||
def _login(client, token):
|
||||
resp = client.post("/login", data={"token": token}, follow_redirects=False)
|
||||
assert resp.status_code == 303
|
||||
assert resp.headers["location"] == "/"
|
||||
|
||||
|
||||
def test_root_requires_login_redirects(client):
|
||||
resp = client.get("/", follow_redirects=False)
|
||||
assert resp.status_code == 303
|
||||
assert resp.headers["location"] == "/login"
|
||||
|
||||
|
||||
def test_invalid_token_rejected(client):
|
||||
resp = client.post("/login", data={"token": "not-a-real-token"}, follow_redirects=False)
|
||||
assert resp.status_code == 303
|
||||
assert resp.headers["location"] == "/login"
|
||||
|
||||
|
||||
def test_valid_login_then_dashboard(client, credentials):
|
||||
_login(client, credentials["viewer"].token)
|
||||
resp = client.get("/")
|
||||
assert resp.status_code == 200
|
||||
assert "Phases for binky" in resp.text
|
||||
|
||||
|
||||
def test_viewer_cannot_reach_phase_new(client, credentials):
|
||||
_login(client, credentials["viewer"].token)
|
||||
resp = client.get("/phases/new", follow_redirects=False)
|
||||
assert resp.status_code == 303
|
||||
assert resp.headers["location"] == "/"
|
||||
|
||||
|
||||
def test_operator_registers_phase_and_appends_entry(client, credentials):
|
||||
_login(client, credentials["operator"].token)
|
||||
phase_id = "trsl:phase:cpapp-test-" + uuid.uuid4().hex[:8]
|
||||
|
||||
resp = client.post(
|
||||
"/phases/new",
|
||||
data={
|
||||
"phase_id": phase_id,
|
||||
"milestone_release_name": "CP UI smoke test release",
|
||||
"source_revision": "abc123",
|
||||
"initial_target_amount": "1000",
|
||||
"currency": "USD",
|
||||
"future_license": "MIT",
|
||||
"degeneration_policy": "trsl:policy:linear-longstop-v0@1.0",
|
||||
"longstop_at": "2027-01-01T00:00:00Z",
|
||||
"ledger": f"examples/{phase_id}/ledger.json",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert resp.status_code == 303
|
||||
assert resp.headers["location"] == f"/phases/{phase_id}"
|
||||
|
||||
detail = client.get(f"/phases/{phase_id}")
|
||||
assert detail.status_code == 200
|
||||
assert phase_id in detail.text
|
||||
|
||||
ledger_resp = client.post(
|
||||
f"/phases/{phase_id}/ledger",
|
||||
data={
|
||||
"entry_id": "trsl:entry:cpappledger0001",
|
||||
"amount": "500",
|
||||
"currency": "USD",
|
||||
"recognized_at": "2026-08-01T00:00:00Z",
|
||||
"evidence_reference": "confidential:evidence:cpappledger0001",
|
||||
"extension_id": "trsl:extension:development-license",
|
||||
"extension_version": "1.0",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert ledger_resp.status_code == 303
|
||||
|
||||
detail_after = client.get(f"/phases/{phase_id}")
|
||||
assert "trsl:entry:cpappledger0001" in detail_after.text
|
||||
|
||||
|
||||
def test_contributor_proposes_operator_approves(client, credentials):
|
||||
_login(client, credentials["operator"].token)
|
||||
phase_id = "trsl:phase:cpapp-propose-" + uuid.uuid4().hex[:8]
|
||||
client.post(
|
||||
"/phases/new",
|
||||
data={
|
||||
"phase_id": phase_id,
|
||||
"milestone_release_name": "CP UI propose-flow release",
|
||||
"source_revision": "def456",
|
||||
"initial_target_amount": "1000",
|
||||
"currency": "USD",
|
||||
"future_license": "MIT",
|
||||
"degeneration_policy": "trsl:policy:linear-longstop-v0@1.0",
|
||||
"longstop_at": "2027-01-01T00:00:00Z",
|
||||
"ledger": f"examples/{phase_id}/ledger.json",
|
||||
},
|
||||
)
|
||||
client.post("/logout")
|
||||
|
||||
_login(client, credentials["contributor"].token)
|
||||
propose_resp = client.post(
|
||||
f"/phases/{phase_id}/ledger",
|
||||
data={
|
||||
"entry_id": "trsl:entry:cpappprop0001",
|
||||
"amount": "250",
|
||||
"currency": "USD",
|
||||
"recognized_at": "2026-08-01T00:00:00Z",
|
||||
"evidence_reference": "confidential:evidence:cpappprop0001",
|
||||
"extension_id": "trsl:extension:development-license",
|
||||
"extension_version": "1.0",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert propose_resp.status_code == 303
|
||||
client.post("/logout")
|
||||
|
||||
_login(client, credentials["operator"].token)
|
||||
proposals_page = client.get("/proposals")
|
||||
assert proposals_page.status_code == 200
|
||||
assert phase_id in proposals_page.text
|
||||
|
||||
|
||||
def test_admin_issues_and_revokes_credential(client, credentials):
|
||||
_login(client, credentials["admin"].token)
|
||||
issue_resp = client.post(
|
||||
"/admin/credentials",
|
||||
data={"credential_label": f"dana-{uuid.uuid4().hex[:6]}", "rights": "viewer"},
|
||||
)
|
||||
assert issue_resp.status_code == 200
|
||||
assert "New credential issued" in issue_resp.text
|
||||
|
||||
|
||||
def test_non_admin_cannot_reach_admin_credentials(client, credentials):
|
||||
_login(client, credentials["operator"].token)
|
||||
resp = client.get("/admin/credentials", follow_redirects=False)
|
||||
assert resp.status_code == 303
|
||||
assert resp.headers["location"] == "/"
|
||||
|
||||
|
||||
def test_audit_log_visible_to_signed_in_user(client, credentials):
|
||||
_login(client, credentials["viewer"].token)
|
||||
resp = client.get("/audit")
|
||||
assert resp.status_code == 200
|
||||
Loading…
Add table
Add a link
Reference in a new issue