jiang
be6317eb48
Add project-bound OAuth MCP queries and controlled analysis with scoped evidence, limits, timeouts and a nearby-model spatial query using EPSG:4326 inputs. Default nearby searches to 200 meters and five results. Integrate investigation messaging/presentation, health/readiness and typed history; remove duplicate web search and fix online simulation and compressed baseline writes. Include project migrations 023-025; keep local credentials and demo seeding excluded. Validation: 319 focused pytest checks; real PostGIS temporary-table regression; read-only Jinqian bridge radius comparison; cross-language MCP discovery/calls.
71 lines
3.0 KiB
Python
71 lines
3.0 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from samples.peripheral.auth import Identity
|
|
from samples.peripheral.config import SampleSettings
|
|
from samples.peripheral.schemas import CreateWorkOrder, UpdateWorkOrder, WorkOrder
|
|
from samples.peripheral.store import PostgresStore
|
|
|
|
|
|
def test_create_and_update_record_flat_order_activity(monkeypatch):
|
|
store = PostgresStore(SampleSettings())
|
|
actor = Identity(str(uuid4()), str(uuid4()), True)
|
|
saved = {}
|
|
|
|
monkeypatch.setattr(store, "_mutate", lambda who, key, command, operation: operation(None, "normal"))
|
|
|
|
def save(_conn, _who, document):
|
|
saved.clear()
|
|
saved.update(WorkOrder.model_validate(document).model_dump(mode="json"))
|
|
return dict(saved)
|
|
|
|
monkeypatch.setattr(store, "_save_work_order", save)
|
|
monkeypatch.setattr(store, "_managed_document", lambda *_args: dict(saved))
|
|
|
|
created = store.create_work_order(actor, CreateWorkOrder(
|
|
request_id=uuid4(), title="阀门预约复核", summary="现场复核开度",
|
|
location="北区阀门", assignee="北区班组",
|
|
feedback_requirements=["回传开度照片"],
|
|
))
|
|
assert created["feedback_requirements"] == ["回传开度照片"]
|
|
assert created["activity"][0]["kind"] == "created"
|
|
assert "subtasks" not in created
|
|
|
|
updated = store.update_work_order(actor, created["work_order_id"], UpdateWorkOrder(
|
|
request_id=uuid4(), expected_revision=1, status="awaiting_reply",
|
|
progress_note="已复核开度，等待回执确认",
|
|
))
|
|
assert updated["status"] == "awaiting_reply"
|
|
assert [entry["sequence"] for entry in updated["activity"]] == [1, 2]
|
|
assert updated["activity"][1]["kind"] == "feedback"
|
|
assert updated["activity"][1]["detail"] == "已复核开度，等待回执确认"
|
|
assert "progress_note" not in updated
|
|
|
|
|
|
@pytest.mark.parametrize(("work_order_type", "detail"), [
|
|
("general", "预约工单已创建。"),
|
|
("water_quality_complaint", "投诉工单已创建。"),
|
|
("burst_report", "爆管上报工单已创建。"),
|
|
])
|
|
def test_created_activity_distinguishes_work_order_type(monkeypatch, work_order_type, detail):
|
|
store = PostgresStore(SampleSettings())
|
|
actor = Identity(str(uuid4()), str(uuid4()), True)
|
|
monkeypatch.setattr(store, "_mutate", lambda who, key, command, operation: operation(None, "normal"))
|
|
monkeypatch.setattr(store, "_save_work_order", lambda _conn, _who, document: document)
|
|
|
|
created = store.create_work_order(actor, CreateWorkOrder(
|
|
request_id=uuid4(), title="测试工单", summary="验证创建活动语义",
|
|
work_order_type=work_order_type,
|
|
))
|
|
|
|
assert created["activity"][0]["detail"] == detail
|
|
|
|
|
|
def test_legacy_order_reads_flat_without_claiming_unrecorded_progress():
|
|
legacy = {"work_order_id": str(uuid4()), "subtasks": [{"stages": [{"status": "done"}]}]}
|
|
projected = PostgresStore._active_managed_document(legacy)
|
|
assert "subtasks" not in projected
|
|
assert projected["feedback_requirements"] == []
|
|
assert projected["activity"] == []
|