import json import inspect import subprocess import sys from datetime import datetime, timedelta, timezone from uuid import UUID, uuid4 import pytest from fastapi.testclient import TestClient from app.services.content_digest import digest from samples.peripheral import auth from samples.peripheral.api import create_app, store from samples.peripheral.config import SampleSettings from samples.peripheral.demo_data import VERSION, build_demo_content from samples.peripheral.errors import SampleFailure from samples.peripheral.store import PostgresStore START = datetime(2026, 9, 17, tzinfo=timezone.utc) def test_same_seed_start_and_version_are_stable_across_processes(): content = build_demo_content("repeatable", START) assert content == build_demo_content("repeatable", START) assert content["generator_version"] == VERSION expected_digest = digest(content) script = """ import json from datetime import datetime from app.services.content_digest import digest from samples.peripheral.demo_data import build_demo_content content = build_demo_content('repeatable', datetime.fromisoformat('2026-09-17T00:00:00+00:00')) print(json.dumps({'digest': digest(content), 'ids': [x['record_id'] for x in content['observations'][:3]]})) """ result = subprocess.run( [sys.executable, "-c", script], check=True, capture_output=True, text=True, ) child = json.loads(result.stdout) assert child["digest"] == expected_digest assert child["ids"] == [item["record_id"] for item in content["observations"][:3]] assert all(UUID(item["record_id"]).version == 5 for item in content["observations"]) def test_default_canonical_digest_is_versioned_regression_anchor(): content = build_demo_content() assert digest(content) == "sha256:c630f66ad0247f70a4afe79cc5f95f4be35d54acfe85fbfe39f3166ee09a57c5" assert (len(content["points"]), len(content["observations"]), len(content["device_states"])) == (5, 240, 9) def test_seed_changes_content_and_timezone_is_normalized(): base = build_demo_content("seed-a", START) assert digest(base) != digest(build_demo_content("seed-b", START)) same_in_china = build_demo_content("seed-a", START.astimezone(timezone(timedelta(hours=8)))) assert same_in_china == base assert base["window"]["start"] == "2026-09-17T00:00:00Z" @pytest.mark.parametrize("seed", ["", "x" * 129, None, 7]) def test_invalid_seed_is_rejected(seed): with pytest.raises(ValueError, match="seed"): build_demo_content(seed, START) @pytest.mark.parametrize("start", [ datetime(2026, 9, 17), datetime(2026, 9, 17, 0, 1, tzinfo=timezone.utc), datetime(2026, 9, 17, 0, 0, 1, tzinfo=timezone.utc), ]) def test_invalid_start_is_rejected(start): with pytest.raises(ValueError, match="timezone|aligned"): build_demo_content("seed", start) def test_story_counts_null_zero_and_temporal_links_are_explicit(): content = build_demo_content("story", START) assert len(content["points"]) == 5 assert len(content["observations"]) == 240 assert len(content["device_states"]) == 9 assert len(content["alerts"]) == 2 assert len(content["operations"]) == 4 missing = [item for item in content["observations"] if item["value"] is None] zero = [item for item in content["observations"] if item["value"] == 0.0] assert len(missing) == 4 and all(item["quality"] == "missing" for item in missing) assert len(zero) == 4 and all(item["quality"] == "valid" for item in zero) states = {(item["asset_id"], item["state"]): item["observed_at"] for item in content["device_states"]} communication = next(item for item in content["alerts"] if item["code"] == "communication_lost") assert communication["asset_id"] == "DEMO-P001" assert communication["occurred_at"] == states[("DEMO-P001", "offline")] assert communication["cleared_at"] == states[("DEMO-P001", "online")] operations = {(item["asset_id"], item["operation"]): item["occurred_at"] for item in content["operations"]} assert operations[("DEMO-PUMP-01", "pump_stop")] == states[("DEMO-PUMP-01", "stopped")] assert operations[("DEMO-VALVE-01", "valve_close")] == states[("DEMO-VALVE-01", "closed")] assert len({item["record_id"] for item in content["observations"] + content["device_states"] + content["operations"]}) == 253 class FakeStore: def __init__(self, result=None, failure=None): self.result = result self.failure = failure self.reads = 0 def demo_data(self, who): self.reads += 1 if self.failure: raise self.failure return {**self.result, "project_id": who.project_id, "source_id": "test:demo", "source_revision": digest(self.result)} def test_demo_endpoint_uses_next_auth_and_allows_read_only_permission(monkeypatch): project_id, user_id = uuid4(), uuid4() fake = FakeStore(build_demo_content("api", START)) app = create_app(SampleSettings(project_code="tjwater_v2")) app.dependency_overrides[store] = lambda: fake monkeypatch.setattr(auth, "next_get", lambda request, path, params=None: { "project_id": str(project_id), "user_id": str(user_id), "network": "tjwater_v2", "permissions": ["simulation.view"], }) response = TestClient(app).get("/api/sample/v1/demo-data", headers={ "Authorization": "Bearer TEST", "X-Project-Id": str(project_id), }) assert response.status_code == 200 assert response.json()["project_id"] == str(project_id) assert fake.reads == 1 def test_demo_endpoint_requires_auth_and_reports_missing_storage_without_seeding(): fake = FakeStore(failure=SampleFailure( "STORAGE_NOT_CONFIGURED", "样板存储表尚未配置", 503, )) app = create_app(SampleSettings(project_code="tjwater_v2")) app.dependency_overrides[store] = lambda: fake client = TestClient(app) unauthorized = client.get("/api/sample/v1/demo-data") assert unauthorized.status_code == 401 assert fake.reads == 0 # Authentication is deliberately isolated here so this assertion tests the # missing-table behavior: the API reports it and never invokes a generator. from samples.peripheral.api import identity from samples.peripheral.auth import Identity app.dependency_overrides[identity] = lambda: Identity(str(uuid4()), str(uuid4()), False) unavailable = client.get("/api/sample/v1/demo-data") assert unavailable.status_code == 503 assert unavailable.json()["code"] == "STORAGE_NOT_CONFIGURED" assert fake.reads == 1 def test_demo_read_path_contains_no_schema_or_seed_writes(): source = inspect.getsource(PostgresStore.demo_data).upper() assert "SELECT " in source assert all(keyword not in source for keyword in ("CREATE ", "ALTER ", "INSERT ", "UPDATE ", "DELETE "))