"""Opt-in checks against provisioned sample tables, never business-table writes.""" from concurrent.futures import ThreadPoolExecutor import os import subprocess import sys from uuid import uuid4 import pytest from fastapi.testclient import TestClient from samples.peripheral import schemas as peripheral_schemas from samples.peripheral.api import create_app from samples.peripheral.auth import Identity, identity from samples.peripheral.config import SampleSettings from samples.peripheral.errors import SampleFailure from samples.peripheral.schemas import ( AddFeedback, CreateTask, CreateTestDataRecord, CreateWorkOrder, DeleteResource, Observations, ScenarioCommand, Task, Transition, UpdateTestDataRecord, UpdateWorkOrder, WorkOrder, ) from samples.peripheral.store import PostgresStore pytestmark = pytest.mark.skipif(os.getenv("PERIPHERAL_SAMPLE_LIVE_TEST") != "1", reason="explicit live sample opt-in required") @pytest.fixture def sample(): store = PostgresStore(SampleSettings(enable_faults=True)) who = Identity(str(uuid4()), str(uuid4()), True) yield store, who # Only this test's unguessable project key; tables created by the authorized setup. with store.connection() as conn: for table in ( "work_order_revisions", "test_data_record_revisions", "work_orders", "test_data_records", "feedback", "requests", "revisions", "tasks", "scenarios", ): from psycopg import sql exists = conn.execute( "SELECT to_regclass(%s) AS name", (f"peripheral_sample.{table}",) ).fetchone()["name"] if exists is not None: conn.execute( sql.SQL("DELETE FROM peripheral_sample.{} WHERE project_id=%s").format( sql.Identifier(table) ), (who.project_id,), ) def command(): return CreateTask.model_validate({ "request_id": str(uuid4()), "title": "AUTOTEST isolated sample", "summary": "not a production source", "source": {"system": "next", "task_id": str(uuid4()), "object_type": "validation", "object_id": str(uuid4()), "revision": 1}, "asset_ids": ["test-pressure"], "analysis_window": {"start": "2026-04-30T08:00:00Z", "end": "2026-04-30T09:00:00Z"}, "evidence_refs": [], "requested_check": "verification only", }) def move(store, who, task, status): return store.transition(who, task["task_id"], Transition(request_id=uuid4(), expected_revision=task["revision"], target_status=status, reason="AUTOTEST")) def feedback(task): return AddFeedback.model_validate({"request_id": str(uuid4()), "expected_revision": task["revision"], "feedback_id": str(uuid4()), "occurred_at": "2026-04-30T08:30:00Z", "applicable_time": task["analysis_window"], "asset_ids": task["asset_ids"], "finding": "instrument_fault", "notes": "AUTOTEST synthetic only", "attachment_refs": []}) def scenario(store, who, name): return store.set_scenario(who, ScenarioCommand(request_id=uuid4(), scenario=name)) def test_persistence_transitions_feedback_idempotency_and_isolation(sample): store, who = sample cmd = command() task = store.create_task(who, cmd) Task.model_validate(task) fresh = PostgresStore(store.settings) assert fresh.get_task(who, task["task_id"]) == task # A new interpreter has no service memory or open DB connections to reuse. probe = subprocess.run([sys.executable, "-c", "import sys; from samples.peripheral.store import PostgresStore; " "from samples.peripheral.config import SampleSettings; from samples.peripheral.auth import Identity; " "print(PostgresStore(SampleSettings()).get_task(Identity(sys.argv[1],sys.argv[2],True),sys.argv[3])['revision'])", who.project_id, who.user_id, task["task_id"]], capture_output=True, text=True, timeout=15, check=True) assert probe.stdout.strip() == "1" assert fresh.create_task(who, cmd) == task with pytest.raises(SampleFailure, match="同一 request_id"): store.create_task(who, cmd.model_copy(update={"title": "changed"})) outsider = Identity(str(uuid4()), who.user_id, True) with pytest.raises(SampleFailure) as exc: store.get_task(outsider, task["task_id"]) assert exc.value.code == "TASK_NOT_FOUND" with pytest.raises(SampleFailure): store.receipt(Identity(who.project_id, str(uuid4()), True), cmd.request_id) with pytest.raises(SampleFailure): store.create_task(Identity(who.project_id, who.user_id, False), command()) with pytest.raises(SampleFailure): move(store, who, task, "completed") task = move(store, who, task, "processing") stale = task.copy() cmd_feedback = feedback(task) task = store.add_feedback(who, task["task_id"], cmd_feedback) repeated = cmd_feedback.model_copy(update={"request_id": uuid4()}) assert store.add_feedback(who, task["task_id"], repeated) == task with pytest.raises(SampleFailure) as exc: store.add_feedback(who, task["task_id"], repeated.model_copy(update={"request_id": uuid4(), "notes": "different"})) assert exc.value.code == "FEEDBACK_CONFLICT" with pytest.raises(SampleFailure) as exc: move(store, who, stale, "pending_review") assert exc.value.code == "REVISION_CONFLICT" task = move(store, who, task, "pending_review") task = move(store, who, task, "processing") task = move(store, who, task, "pending_review") task = move(store, who, task, "completed") assert len(task["feedback"]) == 1 assert len(task["history"]) == 6 with store.connection(readonly=True) as conn: assert conn.execute("SELECT count(*) AS n FROM peripheral_sample.revisions WHERE project_id=%s", (who.project_id,)).fetchone()["n"] == task["revision"] def test_concurrent_same_request_and_revision_conflict(sample): store, who = sample cmd = command() with ThreadPoolExecutor(max_workers=4) as pool: results = list(pool.map(lambda _: store.create_task(who, cmd), range(4))) assert len({r["task_id"] for r in results}) == 1 task = results[0] def race(_): try: return move(store, who, task, "processing")["revision"] except SampleFailure as exc: return exc.code with ThreadPoolExecutor(max_workers=2) as pool: outcomes = list(pool.map(race, range(2))) assert sorted(map(str, outcomes)) == ["2", "REVISION_CONFLICT"] def test_fault_receipt_recovery_duplicate_conflict_and_late_state(sample): store, who = sample scenario(store, who, "response_lost") cmd = command() with pytest.raises(SampleFailure) as exc: store.create_task(who, cmd) assert exc.value.effect == "unknown" task = store.receipt(who, cmd.request_id)["result"] assert store.create_task(who, cmd) == task scenario(store, who, "unavailable") with pytest.raises(SampleFailure): store.get_task(who, task["task_id"]) assert store.receipt(who, cmd.request_id)["task_id"] == task["task_id"] scenario(store, who, "conflicting_feedback") task = move(store, who, task, "processing") task = store.add_feedback(who, task["task_id"], feedback(task)) assert {f["finding"] for f in task["feedback"]} == {"instrument_fault", "no_instrument_fault"} scenario(store, who, "duplicate_feedback") assert len(store.get_task(who, task["task_id"])["feedback"]) == 3 scenario(store, who, "late_state") assert store.get_task(who, task["task_id"])["revision"] == task["revision"] - 1 scenario(store, who, "normal") assert store.get_task(who, task["task_id"]) == task disabled = PostgresStore(SampleSettings(enable_faults=False)) with pytest.raises(SampleFailure) as exc: scenario(disabled, who, "missing_data") assert exc.value.code == "FAULTS_DISABLED" def test_historical_monitoring_readonly_mapping_and_missing_data(sample): store, who = sample points = store.points(who, limit=500) assert points["origin"] == "historical_replay" assert points["items"] point = points["items"][0] kwargs = {"point_id": point["point_id"], "start": "2026-04-30T08:00:00Z", "end": "2026-04-30T16:00:00Z"} normal = store.observations(who, **kwargs) Observations.model_validate(normal) assert normal["items"] assert {x["unit"] for x in normal["items"]} == {point["unit"]} with store.connection(readonly=True, timeseries=True) as conn: assert conn.execute("SHOW transaction_read_only").fetchone()["transaction_read_only"] == "on" scenario(store, who, "missing_data") missing = store.observations(who, **kwargs) assert missing["origin"] == "synthetic" assert all(item["value"] is None for item in missing["items"][::2]) assert store.alerts(who)["availability"] == "not_configured" def test_managed_work_orders_and_test_data_persist_revision_and_isolation(sample): store, who = sample capabilities = store.managed_resource_capabilities() if capabilities != {"work_orders": "available", "test_data": "available"}: pytest.skip("managed peripheral resource tables are not provisioned") order_command = CreateWorkOrder( request_id=uuid4(), title="演示 live test work order", summary="AUTOTEST isolated synthetic resource", status="open", priority="high", assignee="AUTOTEST", scheduled_at="2026-09-18T01:00:00Z", ) order = store.create_work_order(who, order_command) WorkOrder.model_validate(order) assert PostgresStore(store.settings).get_work_order(who, order["work_order_id"]) == order assert store.create_work_order(who, order_command) == order order = store.update_work_order(who, order["work_order_id"], UpdateWorkOrder( request_id=uuid4(), expected_revision=1, status="in_progress", priority="urgent", )) assert order["revision"] == 2 and order["status"] == "in_progress" with pytest.raises(SampleFailure) as stale_order: store.update_work_order(who, order["work_order_id"], UpdateWorkOrder( request_id=uuid4(), expected_revision=1, status="completed", )) assert stale_order.value.code == "REVISION_CONFLICT" data_command = CreateTestDataRecord( request_id=uuid4(), label="演示 live pressure", point_id="演示", observed_at="2026-09-18T00:00:00Z", value=31.2, unit="m", quality="valid", notes="AUTOTEST synthetic only", ) record = store.create_test_data(who, data_command) peripheral_schemas.TestDataRecord.model_validate(record) assert PostgresStore(store.settings).get_test_data(who, record["record_id"]) == record assert store.create_test_data(who, data_command) == record record = store.update_test_data(who, record["record_id"], UpdateTestDataRecord( request_id=uuid4(), expected_revision=1, value=None, clear_value=True, quality="missing", )) assert record["revision"] == 2 and record["value"] is None outsider = Identity(str(uuid4()), who.user_id, True) assert store.list_work_orders(outsider, 20)["items"] == [] assert store.list_test_data(outsider, 20)["items"] == [] assert [item["work_order_id"] for item in store.list_work_orders(who, 20)["items"]] == [ order["work_order_id"] ] assert [item["record_id"] for item in store.list_test_data(who, 20)["items"]] == [ record["record_id"] ] deleted_order = store.delete_work_order(who, order["work_order_id"], DeleteResource( request_id=uuid4(), expected_revision=2, )) deleted_data = store.delete_test_data(who, record["record_id"], DeleteResource( request_id=uuid4(), expected_revision=2, )) assert deleted_order["revision"] == 3 and deleted_data["revision"] == 3 assert store.list_work_orders(who, 20)["items"] == [] assert store.list_test_data(who, 20)["items"] == [] with store.connection(readonly=True) as conn: order_revisions = conn.execute( """SELECT count(*) AS n FROM peripheral_sample.work_order_revisions WHERE project_id=%s AND work_order_id=%s""", (who.project_id, order["work_order_id"]), ).fetchone()["n"] data_revisions = conn.execute( """SELECT count(*) AS n FROM peripheral_sample.test_data_record_revisions WHERE project_id=%s AND record_id=%s""", (who.project_id, record["record_id"]), ).fetchone()["n"] assert order_revisions == 3 and data_revisions == 3 def test_managed_resource_http_routes_use_provisioned_storage(sample): store, who = sample capabilities = store.managed_resource_capabilities() if capabilities != {"work_orders": "available", "test_data": "available"}: pytest.skip("managed peripheral resource tables are not provisioned") app = create_app(store.settings) app.dependency_overrides[identity] = lambda: who headers = { "Authorization": "Bearer TEST-FIXTURE-ONLY", "X-Project-Id": who.project_id, } with TestClient(app) as client: order = client.post("/api/sample/v1/work-orders", headers=headers, json={ "request_id": str(uuid4()), "title": "演示 HTTP test work order", "summary": "AUTOTEST isolated synthetic resource", "status": "open", "priority": "normal", "assignee": None, "scheduled_at": None, }) assert order.status_code == 201, order.text order_document = order.json() listed_orders = client.get("/api/sample/v1/work-orders", headers=headers) assert [item["work_order_id"] for item in listed_orders.json()["items"]] == [ order_document["work_order_id"] ] changed_order = client.patch( f"/api/sample/v1/work-orders/{order_document['work_order_id']}", headers=headers, json={ "request_id": str(uuid4()), "expected_revision": 1, "status": "in_progress", }, ) assert changed_order.status_code == 200 and changed_order.json()["revision"] == 2 record = client.post("/api/sample/v1/test-data", headers=headers, json={ "request_id": str(uuid4()), "label": "演示 HTTP pressure", "point_id": "演示", "observed_at": "2026-09-18T00:00:00Z", "value": 31.2, "unit": "m", "quality": "valid", "notes": "AUTOTEST synthetic only", }) assert record.status_code == 201, record.text record_document = record.json() listed_data = client.get("/api/sample/v1/test-data", headers=headers) assert [item["record_id"] for item in listed_data.json()["items"]] == [ record_document["record_id"] ] changed_data = client.patch( f"/api/sample/v1/test-data/{record_document['record_id']}", headers=headers, json={ "request_id": str(uuid4()), "expected_revision": 1, "value": 30.8, }, ) assert changed_data.status_code == 200 and changed_data.json()["revision"] == 2 deleted_order = client.request( "DELETE", f"/api/sample/v1/work-orders/{order_document['work_order_id']}", headers=headers, json={"request_id": str(uuid4()), "expected_revision": 2}, ) deleted_data = client.request( "DELETE", f"/api/sample/v1/test-data/{record_document['record_id']}", headers=headers, json={"request_id": str(uuid4()), "expected_revision": 2}, ) assert deleted_order.status_code == 200 and deleted_order.json()["revision"] == 3 assert deleted_data.status_code == 200 and deleted_data.json()["revision"] == 3