155 lines
6.7 KiB
Python
155 lines
6.7 KiB
Python
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 "))
|