95 lines
3.9 KiB
Python
95 lines
3.9 KiB
Python
"""VEP P1 boundary tests (canonical Part 6 discipline): test CONTRACTS, not internals.
|
|
These block merge (CI). Covers the four P1 acceptance criteria.
|
|
|
|
Run: cd /root/vep && uv run pytest -q
|
|
Requires DATABASE_URL to point at a fresh P1 postgres (the compose 'postgres' service).
|
|
"""
|
|
import os
|
|
import json
|
|
import pytest
|
|
import pytest_asyncio
|
|
import asyncpg
|
|
from httpx import AsyncClient, ASGITransport
|
|
from vep.main import app
|
|
from vep.initdb import init as db_init
|
|
from vep.db import connection
|
|
|
|
URL = os.environ.get("DATABASE_URL", "postgresql://vep:vep_dev_pass@postgres:5432/vep")
|
|
A_URL = URL.replace("vep", "vep_test_a")
|
|
B_URL = URL.replace("vep", "vep_test_b")
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="module")
|
|
async def client():
|
|
await db_init()
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
async def _seed_tenant(conn, tid, actor="a_owner"):
|
|
await conn.execute("INSERT INTO tenant (id,name) VALUES ($1,$2) ON CONFLICT DO NOTHING", tid, tid)
|
|
await conn.execute(
|
|
"INSERT INTO actor (id,tenant_id,kind,authority_tier) VALUES ($1,$2,'human',0) "
|
|
"ON CONFLICT DO NOTHING", actor, tid)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tenant_isolation(client):
|
|
"""JL-003: tenant A cannot read tenant B's events."""
|
|
# Insert an event for tenant A under A's session scope.
|
|
async with connection("t_a") as conn:
|
|
await _seed_tenant(conn, "t_a", "a_a")
|
|
await conn.execute(
|
|
"INSERT INTO event_log (tenant_id, actor_id, event_type, payload, create_capture, confidence, raw_ref) "
|
|
"VALUES ($1,$2,$3,$4,$5,$6,$7)",
|
|
"t_a", "a_a", "test", json.dumps({"x": 1}), "create", 0.9, "src://a",
|
|
)
|
|
# Now B tries to read: RLS should hide A's row.
|
|
async with connection("t_b") as conn:
|
|
await _seed_tenant(conn, "t_b", "a_b")
|
|
n = await conn.fetchval("SELECT count(*) FROM event_log")
|
|
assert n == 0, "tenant B saw tenant A's events — RLS isolation failed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_unvalidated_row(client):
|
|
"""JL-006: a missing actor_id is rejected AND recorded in dead_letter (no silent loss)."""
|
|
before = await _dead_count()
|
|
resp = await client.post("/events", json={"event_type": "no_actor", "payload": {}})
|
|
assert resp.status_code == 422, "invalid event was accepted"
|
|
after = await _dead_count()
|
|
assert after == before + 1, "rejected event was silently dropped"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provenance_required(client):
|
|
"""#2: a valid event records actor_id + raw_ref + confidence in the log."""
|
|
async with connection("t_p") as conn:
|
|
await _seed_tenant(conn, "t_p", "a_p")
|
|
resp = await client.post("/events", json={
|
|
"tenant_id": "t_p", "actor_id": "a_p", "event_type": "probe",
|
|
"payload": {"v": 1}, "create_capture": "create", "confidence": 0.7, "raw_ref": "sheet://x",
|
|
})
|
|
assert resp.status_code == 200
|
|
async with connection("t_p") as conn:
|
|
row = await conn.fetchrow("SELECT actor_id, raw_ref, confidence FROM event_log WHERE id=$1", resp.json()["id"])
|
|
assert row["actor_id"] == "a_p" and row["raw_ref"] == "sheet://x" and row["confidence"] == 0.7
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_append_only(client):
|
|
"""#2 / JL-006: the log cannot be mutated via the app role (privilege layer)."""
|
|
async with connection("t_ao") as conn:
|
|
await _seed_tenant(conn, "t_ao", "a_ao")
|
|
await conn.execute(
|
|
"INSERT INTO event_log (tenant_id, actor_id, event_type, payload) VALUES ($1,$2,$3,$4)",
|
|
"t_ao", "a_ao", "immutable", json.dumps({}),
|
|
)
|
|
with pytest.raises(Exception):
|
|
await conn.execute("UPDATE event_log SET event_type='hacked' WHERE tenant_id='t_ao'")
|
|
|
|
|
|
async def _dead_count():
|
|
async with connection() as conn:
|
|
return await conn.fetchval("SELECT count(*) FROM dead_letter")
|