147 lines
7.6 KiB
Python
147 lines
7.6 KiB
Python
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from samples.peripheral.api import create_app, store
|
|
from samples.peripheral.auth import Identity, identity
|
|
from samples.peripheral.schemas import CreateTestDataRecord, UpdateTestDataRecord, UpdateWorkOrder
|
|
|
|
|
|
PROJECT_ID = uuid4()
|
|
USER_ID = uuid4()
|
|
|
|
|
|
class ManagedStore:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def list_work_orders(self, who, limit, cursor):
|
|
self.calls.append(("list_work_orders", who, limit, cursor))
|
|
return {"items": [], "next_cursor": None}
|
|
|
|
def create_work_order(self, who, command):
|
|
self.calls.append(("create_work_order", who, command))
|
|
now = "2026-09-18T00:00:00Z"
|
|
return command.model_dump(mode="json", exclude={"request_id"}) | {
|
|
"work_order_id": str(uuid4()), "project_id": str(PROJECT_ID), "revision": 1,
|
|
"origin": "synthetic", "created_at": now, "updated_at": now,
|
|
"created_by": str(USER_ID), "updated_by": str(USER_ID), "activity": [],
|
|
}
|
|
|
|
def update_work_order(self, who, work_order_id, command):
|
|
self.calls.append(("update_work_order", who, work_order_id, command))
|
|
return self.create_work_order(who, type("Command", (), {"model_dump": lambda self, **kwargs: {
|
|
"title": command.title or "updated", "summary": command.summary or "summary",
|
|
"status": command.status or "open", "priority": command.priority or "normal",
|
|
"assignee": None, "scheduled_at": None, "work_order_type": "general",
|
|
"location": command.location, "scenario_code": command.scenario_code,
|
|
"feedback_requirements": command.feedback_requirements or []}})()) | {"revision": command.expected_revision + 1,
|
|
"work_order_id": str(work_order_id)}
|
|
|
|
def delete_work_order(self, who, work_order_id, command):
|
|
self.calls.append(("delete_work_order", who, work_order_id, command))
|
|
return {"resource_type": "work_order", "resource_id": str(work_order_id),
|
|
"revision": command.expected_revision + 1, "deleted": True}
|
|
|
|
def list_test_data(self, who, limit, cursor):
|
|
self.calls.append(("list_test_data", who, limit, cursor))
|
|
return {"items": [], "next_cursor": None}
|
|
|
|
def create_test_data(self, who, command):
|
|
self.calls.append(("create_test_data", who, command))
|
|
now = "2026-09-18T00:00:00Z"
|
|
return command.model_dump(mode="json", exclude={"request_id"}) | {
|
|
"record_id": str(uuid4()), "project_id": str(PROJECT_ID), "revision": 1,
|
|
"origin": "synthetic", "created_at": now, "updated_at": now,
|
|
"created_by": str(USER_ID), "updated_by": str(USER_ID),
|
|
}
|
|
|
|
def update_test_data(self, who, record_id, command):
|
|
self.calls.append(("update_test_data", who, record_id, command))
|
|
return self.create_test_data(who, CreateTestDataRecord(
|
|
request_id=command.request_id, label=command.label or "updated", point_id=command.point_id or "演示",
|
|
observed_at=command.observed_at or "2026-09-18T00:00:00Z", value=command.value,
|
|
unit=command.unit or "m", quality=command.quality or "valid", notes=command.notes or "",
|
|
)) | {"revision": command.expected_revision + 1, "record_id": str(record_id)}
|
|
|
|
def delete_test_data(self, who, record_id, command):
|
|
self.calls.append(("delete_test_data", who, record_id, command))
|
|
return {"resource_type": "test_data_record", "resource_id": str(record_id),
|
|
"revision": command.expected_revision + 1, "deleted": True}
|
|
|
|
|
|
def client_with(fake):
|
|
app = create_app()
|
|
app.dependency_overrides[store] = lambda: fake
|
|
app.dependency_overrides[identity] = lambda: Identity(str(PROJECT_ID), str(USER_ID), True)
|
|
return TestClient(app)
|
|
|
|
|
|
def test_managed_work_order_and_data_routes_are_typed_crud():
|
|
fake = ManagedStore()
|
|
with client_with(fake) as client:
|
|
assert client.get("/api/sample/v1/work-orders").json()["items"] == []
|
|
created = client.post("/api/sample/v1/work-orders", json={
|
|
"request_id": str(uuid4()), "title": "演示 测试工单", "summary": "外围联调",
|
|
"status": "open", "priority": "high", "assignee": None, "scheduled_at": None,
|
|
"work_order_type": "water_quality_complaint", "location": "演示小区",
|
|
"scenario_code": "WQ-SECONDARY-SUPPLY-001",
|
|
"feedback_requirements": ["上传检测结果"],
|
|
})
|
|
assert created.status_code == 201 and created.json()["origin"] == "synthetic"
|
|
assert created.json()["work_order_type"] == "water_quality_complaint"
|
|
assert created.json()["location"] == "演示小区"
|
|
assert created.json()["scenario_code"] == "WQ-SECONDARY-SUPPLY-001"
|
|
assert created.json()["feedback_requirements"] == ["上传检测结果"]
|
|
assert "subtasks" not in created.json()
|
|
work_order_id = created.json()["work_order_id"]
|
|
changed = client.patch(f"/api/sample/v1/work-orders/{work_order_id}", json={
|
|
"request_id": str(uuid4()), "expected_revision": 1, "status": "awaiting_reply",
|
|
"location": "复测点", "clear_scenario_code": True, "progress_note": "等待现场检测结果",
|
|
"clear_assignee": False, "clear_scheduled_at": False,
|
|
})
|
|
assert changed.status_code == 200 and changed.json()["revision"] == 2
|
|
assert changed.json()["status"] == "awaiting_reply"
|
|
update_command = next(call[3] for call in fake.calls if call[0] == "update_work_order")
|
|
assert update_command.location == "复测点"
|
|
assert update_command.clear_scenario_code and update_command.progress_note == "等待现场检测结果"
|
|
deleted = client.request("DELETE", f"/api/sample/v1/work-orders/{work_order_id}", json={
|
|
"request_id": str(uuid4()), "expected_revision": 2,
|
|
})
|
|
assert deleted.json()["resource_type"] == "work_order"
|
|
|
|
assert client.get("/api/sample/v1/test-data").json()["items"] == []
|
|
data = client.post("/api/sample/v1/test-data", json={
|
|
"request_id": str(uuid4()), "label": "演示 压力", "point_id": "演示",
|
|
"observed_at": "2026-09-18T00:00:00Z", "value": 31.2, "unit": "m",
|
|
"quality": "valid", "notes": "联调",
|
|
})
|
|
assert data.status_code == 201 and data.json()["point_id"] == "演示"
|
|
record_id = data.json()["record_id"]
|
|
changed_data = client.patch(f"/api/sample/v1/test-data/{record_id}", json={
|
|
"request_id": str(uuid4()), "expected_revision": 1, "value": 30.8, "clear_value": False,
|
|
})
|
|
assert changed_data.status_code == 200 and changed_data.json()["revision"] == 2
|
|
deleted_data = client.request("DELETE", f"/api/sample/v1/test-data/{record_id}", json={
|
|
"request_id": str(uuid4()), "expected_revision": 2,
|
|
})
|
|
assert deleted_data.json()["resource_type"] == "test_data_record"
|
|
|
|
|
|
def test_managed_update_models_reject_empty_or_conflicting_changes():
|
|
request_id = uuid4()
|
|
for model, payload in (
|
|
(UpdateWorkOrder, {"request_id": request_id, "expected_revision": 1}),
|
|
(UpdateTestDataRecord, {"request_id": request_id, "expected_revision": 1}),
|
|
(UpdateTestDataRecord, {"request_id": request_id, "expected_revision": 1,
|
|
"value": 1.0, "clear_value": True}),
|
|
(UpdateWorkOrder, {"request_id": request_id, "expected_revision": 1,
|
|
"location": "新位置", "clear_location": True}),
|
|
):
|
|
try:
|
|
model.model_validate(payload)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("invalid partial update was accepted")
|