tegwick
3064c0fe0c
Accept WP-0008-T05 for trsl:phase:info-tech-canon-service-surface (history/260805-T05-GoLive-info-tech-canon.md). Finish WP-0013 remission automation and WP-0014 extension/breach/attestation Control Plane UI. Update SCOPE, README, and pilot-candidate notes for pilot Stage 1.
532 lines
22 KiB
Python
532 lines
22 KiB
Python
"""Integration tests for WP-0006-T04 (hosted Target Ledger append API).
|
|
|
|
Same ephemeral, disposable Postgres-via-Docker pattern as
|
|
test_registry_hosting.py (never the shared state-hub instance). The single
|
|
property this file tests most aggressively, per the task's own framing:
|
|
appending entries through the hosted API and then folding the exported
|
|
result offline (`fold.fold_outstanding_target`) must produce exactly the
|
|
same Outstanding Target a hosted metrics/attestation component would later
|
|
report — a hosted service that silently diverges from the offline fold is
|
|
the highest-risk defect this component could introduce.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
psycopg = pytest.importorskip("psycopg")
|
|
pytest.importorskip("fastapi")
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
from conftest import golden_manifest # noqa: E402
|
|
|
|
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",
|
|
]
|
|
|
|
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-ledger-{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 conn:
|
|
for migration in MIGRATIONS:
|
|
conn.execute(migration.read_text(encoding="utf-8"))
|
|
conn.commit()
|
|
token = "test-token-acme"
|
|
conn.execute(
|
|
"INSERT INTO licensors (token, licensor_id) VALUES (%s, %s)",
|
|
(token, "acme-corp"),
|
|
)
|
|
other_token = "test-token-other"
|
|
conn.execute(
|
|
"INSERT INTO licensors (token, licensor_id) VALUES (%s, %s)",
|
|
(other_token, "other-corp"),
|
|
)
|
|
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,
|
|
"token": token,
|
|
"other_token": other_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_SIGNING_KEY_HEX", "11" * 32)
|
|
from target_revenue.service import app as app_module
|
|
|
|
if hasattr(app_module.app.state, "pool"):
|
|
app_module.app.state.pool.close()
|
|
del app_module.app.state.pool
|
|
if hasattr(app_module.app.state, "signing_key"):
|
|
del app_module.app.state.signing_key
|
|
with TestClient(app_module.app) as c:
|
|
yield c
|
|
|
|
|
|
def auth_headers(token: str):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.fixture()
|
|
def registered_phase(client, pg_container):
|
|
manifest = golden_manifest()
|
|
manifest["phase"]["id"] = manifest["phase"]["id"] + "-ledger-test-" + uuid.uuid4().hex[:6]
|
|
resp = client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 201, resp.text
|
|
return manifest
|
|
|
|
|
|
def _entry(phase_id: str, entry_id: str, entry_type: str, amount: float, recognized_at: str):
|
|
return {
|
|
"id": entry_id,
|
|
"phase": phase_id,
|
|
"type": entry_type,
|
|
"amount": amount,
|
|
"currency": "USD",
|
|
"recognized_at": recognized_at,
|
|
"evidence_reference": f"confidential:evidence:{entry_id}",
|
|
"extension": {"id": "trsl:extension:development-license", "version": "1.0"},
|
|
}
|
|
|
|
|
|
def test_append_and_hash_chain_links(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e1 = _entry(phase_id, "trsl:entry:ledgertest0001", "development-credit", 1000, "2026-08-01T00:00:00Z")
|
|
r1 = client.post(f"/phases/{phase_id}/ledger", json=e1, headers=auth_headers(pg_container["token"]))
|
|
assert r1.status_code == 201, r1.text
|
|
stored1 = r1.json()
|
|
assert stored1["previous_entry_hash"] == "GENESIS"
|
|
assert "signature" in stored1
|
|
|
|
e2 = _entry(phase_id, "trsl:entry:ledgertest0002", "development-credit", 500, "2026-08-02T00:00:00Z")
|
|
r2 = client.post(f"/phases/{phase_id}/ledger", json=e2, headers=auth_headers(pg_container["token"]))
|
|
assert r2.status_code == 201, r2.text
|
|
stored2 = r2.json()
|
|
|
|
from target_revenue import hashing
|
|
|
|
assert stored2["previous_entry_hash"] == hashing.entry_hash(stored1)
|
|
# Full chain, as exported, must independently verify.
|
|
hashing.verify_chain([stored1, stored2])
|
|
|
|
|
|
def test_client_cannot_supply_previous_entry_hash_or_signature(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
forged = _entry(phase_id, "trsl:entry:ledgertest0003", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
forged["previous_entry_hash"] = "GENESIS"
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=forged, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 422
|
|
assert "must not be supplied" in resp.json()["detail"]
|
|
|
|
|
|
def test_other_licensor_cannot_append_to_this_phase(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertest0004", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["other_token"]))
|
|
assert resp.status_code == 422
|
|
assert "not authorized" in resp.json()["detail"]
|
|
|
|
|
|
def test_currency_mismatch_rejected(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertest0005", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
e["currency"] = "EUR"
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 422
|
|
assert "currency" in resp.json()["detail"]
|
|
|
|
|
|
def test_hosted_ledger_export_folds_identically_to_offline_computation(
|
|
client, pg_container, registered_phase
|
|
):
|
|
"""The property T04's task description calls out as highest-risk:
|
|
hosted append + export, folded offline, must match what was appended."""
|
|
from target_revenue import fold, hashing
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
plan = [
|
|
("development-credit", 25000),
|
|
("development-credit", 10000),
|
|
("remission-credit", 5000),
|
|
("credit-reversal", 2000),
|
|
]
|
|
for i, (entry_type, amount) in enumerate(plan, start=1):
|
|
e = _entry(
|
|
phase_id,
|
|
f"trsl:entry:ledgertestfold{i:04d}",
|
|
entry_type,
|
|
amount,
|
|
f"2026-08-0{i}T00:00:00Z",
|
|
)
|
|
if entry_type == "credit-reversal":
|
|
e["reverses"] = "trsl:entry:ledgertestfold0001"
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 201, resp.text
|
|
|
|
export = client.get(f"/phases/{phase_id}/ledger")
|
|
assert export.status_code == 200
|
|
entries = export.json()
|
|
assert len(entries) == len(plan)
|
|
|
|
# Chain integrity survives the round trip through Postgres/JSONB/API.
|
|
hashing.verify_chain(entries)
|
|
|
|
result = fold.fold_outstanding_target(
|
|
registered_phase["phase"]["initial_target"]["amount"], entries
|
|
)
|
|
expected_development_credit = 25000 + 10000 - 2000
|
|
expected_remission_credit = 5000
|
|
assert result.development_credit == expected_development_credit
|
|
assert result.remission_credit == expected_remission_credit
|
|
assert result.outstanding_target == max(
|
|
0.0,
|
|
registered_phase["phase"]["initial_target"]["amount"]
|
|
- expected_development_credit
|
|
- expected_remission_credit,
|
|
)
|
|
|
|
|
|
def test_public_key_endpoint_unauthenticated_and_verifies_signature(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertestpk0001", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
stored = client.post(
|
|
f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])
|
|
).json()
|
|
|
|
pk_resp = client.get("/public-key")
|
|
assert pk_resp.status_code == 200
|
|
assert pk_resp.json()["algorithm"] == "ed25519"
|
|
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
|
|
|
|
from target_revenue import hashing
|
|
|
|
public_key = Ed25519PublicKey.from_public_bytes(
|
|
bytes.fromhex(pk_resp.json()["public_key_hex"])
|
|
)
|
|
assert hashing.verify_record_signature(stored, stored["signature"], public_key)
|
|
|
|
|
|
def test_application_role_cannot_update_or_delete_ledger_entries(pg_container):
|
|
with psycopg.connect(pg_container["app_dsn"]) as conn:
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("UPDATE ledger_entries SET signature = 'x' WHERE entry_id = 'nonexistent'")
|
|
conn.rollback()
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("DELETE FROM ledger_entries WHERE entry_id = 'nonexistent'")
|
|
conn.rollback()
|
|
|
|
|
|
def test_duplicate_entry_id_rejected(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertestdup0001", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
first = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert first.status_code == 201
|
|
second = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert second.status_code == 422
|
|
assert "already exists" in second.json()["detail"]
|
|
|
|
|
|
def test_metrics_endpoint_unauthenticated_and_matches_offline_computation(
|
|
client, pg_container, registered_phase
|
|
):
|
|
"""WP-0006-T05: the hosted /metrics response must match what
|
|
target_revenue.metrics.compute_metrics computes offline from the same
|
|
exported Manifest + Ledger — the metrics-layer analogue of T04's
|
|
hosted/offline fold-agreement test."""
|
|
from target_revenue import metrics as metrics_module
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertestmetrics0001", "development-credit", 4000, "2026-08-01T00:00:00Z")
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 201, resp.text
|
|
|
|
# No Authorization header at all: metrics are public per FR-9/FR-10.
|
|
metrics_resp = client.get(f"/phases/{phase_id}/metrics")
|
|
assert metrics_resp.status_code == 200
|
|
hosted = metrics_resp.json()
|
|
|
|
entries = client.get(f"/phases/{phase_id}/ledger").json()
|
|
as_of = metrics_module.datetime.fromisoformat(hosted["as_of"])
|
|
# Hosted metrics pass phase_manifests.registered_at as activated_at
|
|
# (WP-0013); offline recompute must use the same t0 for parity.
|
|
activated_at = None
|
|
if hosted["facts"].get("activated_at"):
|
|
activated_at = metrics_module.datetime.fromisoformat(hosted["facts"]["activated_at"])
|
|
offline = metrics_module.compute_metrics(
|
|
registered_phase, entries, as_of, activated_at=activated_at
|
|
)
|
|
|
|
assert hosted == offline
|
|
assert hosted["facts"]["cumulative_development_credit"] == 4000
|
|
assert hosted["facts"]["activated_at"] is not None
|
|
|
|
|
|
def test_attestation_not_available_before_conversion(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertestattest0001", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 201
|
|
|
|
metrics_resp = client.get(f"/phases/{phase_id}/metrics")
|
|
assert metrics_resp.json()["facts"]["is_converted"] is False
|
|
|
|
attestation_resp = client.get(f"/phases/{phase_id}/attestation")
|
|
assert attestation_resp.status_code == 404
|
|
|
|
|
|
def test_conversion_is_true_independent_of_attestation_call(client, pg_container):
|
|
"""TSD §3.5 legal-technical rule: conversion is already true from the
|
|
ledger the instant the fold reaches zero; nothing about calling (or not
|
|
calling) /attestation may change what /metrics already reports."""
|
|
manifest = golden_manifest()
|
|
manifest["phase"]["id"] = manifest["phase"]["id"] + "-attest-conv-" + uuid.uuid4().hex[:6]
|
|
manifest["phase"]["initial_target"]["amount"] = 1000
|
|
client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"])).raise_for_status()
|
|
phase_id = manifest["phase"]["id"]
|
|
|
|
e = _entry(phase_id, "trsl:entry:ledgertestattest0002", "development-credit", 1000, "2026-08-01T00:00:00Z")
|
|
client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])).raise_for_status()
|
|
|
|
before = client.get(f"/phases/{phase_id}/metrics").json()
|
|
assert before["facts"]["is_converted"] is True
|
|
assert before["facts"]["outstanding_target"] == 0
|
|
|
|
# No attestation has been requested yet; conversion is already true.
|
|
attestation_resp = client.get(f"/phases/{phase_id}/attestation")
|
|
assert attestation_resp.status_code == 200
|
|
|
|
after = client.get(f"/phases/{phase_id}/metrics").json()
|
|
assert after["facts"]["is_converted"] is True
|
|
# Publishing the attestation must change nothing about the conversion
|
|
# facts/calculations/forecasts themselves (only `as_of` legitimately
|
|
# differs, since /metrics stamps wall-clock time on every call).
|
|
assert after["facts"] == before["facts"]
|
|
assert after["calculations"] == before["calculations"]
|
|
assert after["forecasts"] == before["forecasts"]
|
|
|
|
|
|
def test_attestation_published_once_and_idempotent_on_reread(client, pg_container):
|
|
manifest = golden_manifest()
|
|
manifest["phase"]["id"] = manifest["phase"]["id"] + "-attest-idem-" + uuid.uuid4().hex[:6]
|
|
manifest["phase"]["initial_target"]["amount"] = 500
|
|
client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"])).raise_for_status()
|
|
phase_id = manifest["phase"]["id"]
|
|
|
|
e = _entry(phase_id, "trsl:entry:ledgertestattest0003", "development-credit", 500, "2026-08-01T00:00:00Z")
|
|
client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])).raise_for_status()
|
|
|
|
first = client.get(f"/phases/{phase_id}/attestation")
|
|
assert first.status_code == 200
|
|
second = client.get(f"/phases/{phase_id}/attestation")
|
|
assert second.status_code == 200
|
|
assert first.json() == second.json() # same signed record, not regenerated
|
|
|
|
from target_revenue import validation
|
|
|
|
validation.validate_conversion_attestation(
|
|
{k: v for k, v in first.json().items() if k != "signature"}
|
|
)
|
|
assert first.json()["conversion_timestamp"] == "2026-08-01T00:00:00Z"
|
|
assert first.json()["ledger_checkpoint"] == "trsl:entry:ledgertestattest0003"
|
|
|
|
|
|
def test_attestation_verifiable_with_public_key(client, pg_container):
|
|
manifest = golden_manifest()
|
|
manifest["phase"]["id"] = manifest["phase"]["id"] + "-attest-verify-" + uuid.uuid4().hex[:6]
|
|
manifest["phase"]["initial_target"]["amount"] = 500
|
|
client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"])).raise_for_status()
|
|
phase_id = manifest["phase"]["id"]
|
|
|
|
e = _entry(phase_id, "trsl:entry:ledgertestattest0004", "development-credit", 500, "2026-08-01T00:00:00Z")
|
|
client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])).raise_for_status()
|
|
|
|
published = client.get(f"/phases/{phase_id}/attestation").json()
|
|
pk_hex = client.get("/public-key").json()["public_key_hex"]
|
|
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
|
|
|
|
from target_revenue import hashing
|
|
|
|
public_key = Ed25519PublicKey.from_public_bytes(bytes.fromhex(pk_hex))
|
|
assert hashing.verify_record_signature(published, published["signature"], public_key)
|
|
|
|
|
|
def test_application_role_cannot_update_or_delete_attestations(pg_container):
|
|
with psycopg.connect(pg_container["app_dsn"]) as conn:
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("UPDATE attestations SET signature = 'x' WHERE phase_id = 'nonexistent'")
|
|
conn.rollback()
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("DELETE FROM attestations WHERE phase_id = 'nonexistent'")
|
|
conn.rollback()
|
|
|
|
|
|
def _breach_event(event_id: str, case_id: str, event_type: str, **overrides) -> dict:
|
|
event = {
|
|
"id": event_id,
|
|
"case_id": case_id,
|
|
"event_type": event_type,
|
|
"category": "unauthorized-commercial-use",
|
|
"event_at": "2026-08-01T00:00:00Z",
|
|
}
|
|
event.update(overrides)
|
|
return event
|
|
|
|
|
|
def test_breach_record_defaults_anonymized_and_lists_lifecycle_events(
|
|
client, pg_container, registered_phase
|
|
):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
case_id = "trsl:case:conformcase0001"
|
|
|
|
alleged = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event("trsl:breach:conformcase00010001", case_id, "alleged"),
|
|
headers=auth_headers(pg_container["token"]),
|
|
)
|
|
assert alleged.status_code == 201, alleged.text
|
|
assert alleged.json()["anonymized"] is True
|
|
assert alleged.json()["named_entitlement_holder"] is None
|
|
assert "signature" in alleged.json()
|
|
|
|
determined = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event(
|
|
"trsl:breach:conformcase00010002", case_id, "determined",
|
|
event_at="2026-08-15T00:00:00Z",
|
|
),
|
|
headers=auth_headers(pg_container["token"]),
|
|
)
|
|
assert determined.status_code == 201, determined.text
|
|
|
|
records = client.get(f"/phases/{phase_id}/breach-records")
|
|
assert records.status_code == 200
|
|
events = records.json()
|
|
assert [e["event_type"] for e in events] == ["alleged", "determined"]
|
|
# The "alleged" event must never have been edited by the "determined"
|
|
# publication — both remain, in order, as separate records.
|
|
assert events[0]["id"] == "trsl:breach:conformcase00010001"
|
|
assert events[1]["id"] == "trsl:breach:conformcase00010002"
|
|
|
|
|
|
def test_breach_record_named_disclosure_requires_explicit_cua_authorization(
|
|
client, pg_container, registered_phase
|
|
):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
unauthorized = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event(
|
|
"trsl:breach:conformnamed0001", "trsl:case:conformnamed", "alleged",
|
|
anonymized=False, named_entitlement_holder="Acme Corp",
|
|
),
|
|
headers=auth_headers(pg_container["token"]),
|
|
)
|
|
assert unauthorized.status_code == 422
|
|
assert "named_disclosure_authorized_under_cua" in unauthorized.json()["detail"]
|
|
|
|
authorized = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event(
|
|
"trsl:breach:conformnamed0002", "trsl:case:conformnamed", "alleged",
|
|
anonymized=False,
|
|
named_entitlement_holder="Acme Corp",
|
|
named_disclosure_authorized_under_cua=True,
|
|
),
|
|
headers=auth_headers(pg_container["token"]),
|
|
)
|
|
assert authorized.status_code == 201, authorized.text
|
|
assert authorized.json()["named_entitlement_holder"] == "Acme Corp"
|
|
|
|
|
|
def test_other_licensor_cannot_publish_breach_record_for_this_phase(
|
|
client, pg_container, registered_phase
|
|
):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
resp = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event("trsl:breach:conformcross0001", "trsl:case:conformcross", "alleged"),
|
|
headers=auth_headers(pg_container["other_token"]),
|
|
)
|
|
assert resp.status_code == 422
|
|
assert "not authorized" in resp.json()["detail"]
|
|
|
|
|
|
def test_breach_record_signature_verifiable_with_public_key(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
published = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event("trsl:breach:conformverify0001", "trsl:case:conformverify", "alleged"),
|
|
headers=auth_headers(pg_container["token"]),
|
|
).json()
|
|
|
|
pk_hex = client.get("/public-key").json()["public_key_hex"]
|
|
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
|
|
|
|
from target_revenue import hashing
|
|
|
|
public_key = Ed25519PublicKey.from_public_bytes(bytes.fromhex(pk_hex))
|
|
assert hashing.verify_record_signature(published, published["signature"], public_key)
|
|
|
|
|
|
def test_application_role_cannot_update_or_delete_breach_records(pg_container):
|
|
with psycopg.connect(pg_container["app_dsn"]) as conn:
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("UPDATE breach_records SET category = 'x' WHERE record_id = 'nonexistent'")
|
|
conn.rollback()
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("DELETE FROM breach_records WHERE record_id = 'nonexistent'")
|
|
conn.rollback()
|