Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
549019507e |
@@ -1513,11 +1513,13 @@ class OperationalTaskService:
|
||||
body = _json(command)
|
||||
scope = f"task:{task_id}:external-feedback"
|
||||
async with self.repo.transaction():
|
||||
await self.repo.lock(f"task:{task_id}")
|
||||
task = await self._task(task_id)
|
||||
require(task.get("execution_scope") == self.execution_scope,
|
||||
"EXECUTION_SCOPE_MISMATCH", "当前环境不能同步其他执行范围的任务", status=403)
|
||||
replay = await self._request_replay(scope, command.request_id, body)
|
||||
if replay:
|
||||
return ExternalFeedbackResult(task=await self.detail(task_id), **replay)
|
||||
await self.repo.lock(f"task:{task_id}")
|
||||
task = await self._task(task_id)
|
||||
if task["task_type"] == "event_investigation":
|
||||
self._require_actionable_task(task)
|
||||
own_source = task["source"].get("external_id")
|
||||
@@ -1674,11 +1676,13 @@ class OperationalTaskService:
|
||||
body = _json(command)
|
||||
scope = f"task:{task_id}:withdraw-external-source"
|
||||
async with self.repo.transaction():
|
||||
await self.repo.lock(f"task:{task_id}")
|
||||
task = await self._task(task_id)
|
||||
require(task.get("execution_scope") == self.execution_scope,
|
||||
"EXECUTION_SCOPE_MISMATCH", "当前环境不能同步其他执行范围的任务", status=403)
|
||||
replay = await self._request_replay(scope, command.request_id, body)
|
||||
if replay:
|
||||
return await self.detail(task_id)
|
||||
await self.repo.lock(f"task:{task_id}")
|
||||
task = await self._task(task_id)
|
||||
source = task["source"]
|
||||
require(
|
||||
task["task_type"] == "external_work_order"
|
||||
|
||||
@@ -210,7 +210,9 @@ SQL 同时增加必填范围约束,不设置数据库默认值。回滚版本
|
||||
Worker 应连接所属环境的 `OPERATIONAL_WORKER_SERVER_URL` 和
|
||||
`OPERATIONAL_WORKER_AGENT_URL`。Server 在运行列表、领取、租约回收与
|
||||
SCADA recovery 候选查询上限制执行范围;直接领取其他范围返回
|
||||
`403 EXECUTION_SCOPE_MISMATCH`。这属于配置隔离,不能替代服务凭据授权:
|
||||
`403 EXECUTION_SCOPE_MISMATCH`。外部工单反馈与来源撤回也必须匹配任务的执行范围,
|
||||
并在幂等重放前校验;开发环境同步器不能更新生产任务或重置其恢复监测起点。
|
||||
这属于配置隔离,不能替代服务凭据授权:
|
||||
不要让本地 Worker 通过线上 Server 领取任务。普通任务与历史记录仍共用存储。
|
||||
|
||||
修改范围配置需要重启相应 Server。暂停的本地 Worker 不会因代码更新恢复。
|
||||
|
||||
@@ -2487,3 +2487,59 @@ def test_task_scope_is_required_in_persisted_contract():
|
||||
OperationalTaskSummary.model_validate(task)
|
||||
assert any(item["loc"] == ("execution_scope",) and item["type"] == "missing"
|
||||
for item in error.value.errors())
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize("operation", ["feedback", "withdrawal"])
|
||||
@pytest.mark.parametrize("task_scope,server_scope,allowed", [
|
||||
(None, "production", False),
|
||||
(None, "development", False),
|
||||
("production", "production", True),
|
||||
("production", "development", False),
|
||||
("development", "production", False),
|
||||
("development", "development", True),
|
||||
])
|
||||
async def test_external_sync_only_mutates_tasks_in_its_execution_scope(operation, task_scope, server_scope, allowed):
|
||||
from copy import deepcopy
|
||||
|
||||
now = datetime(2026, 11, 8, tzinfo=timezone.utc)
|
||||
context = _context()
|
||||
task = _task_document(context, now, lease=None)
|
||||
source_id = f"peripheral_sample:{uuid4()}"
|
||||
task.update(task_type="external_work_order", source={
|
||||
"kind": "external_system", "external_id": source_id,
|
||||
"external_revision": 1, "event_kind": "general",
|
||||
})
|
||||
if task_scope is None:
|
||||
task.pop("execution_scope")
|
||||
else:
|
||||
task["execution_scope"] = task_scope
|
||||
repository = TaskRepository(task)
|
||||
service = OperationalTaskService(repository, context, set(), clock=lambda: now, execution_scope=server_scope)
|
||||
if operation == "feedback":
|
||||
command = ExternalFeedbackCommand(request_id=uuid4(), source_event_id=source_id,
|
||||
source_revision=2, status="completed", occurred_at=now, detail="已关阀修复管道")
|
||||
method = "external_feedback"
|
||||
else:
|
||||
command = ExternalSourceWithdrawalCommand(request_id=uuid4(), source_event_id=source_id)
|
||||
method = "withdraw_external_source"
|
||||
before = deepcopy(repository.task)
|
||||
if not allowed:
|
||||
with pytest.raises(DomainError) as error:
|
||||
await getattr(service, method)(task["task_id"], command)
|
||||
assert error.value.code == "EXECUTION_SCOPE_MISMATCH" and error.value.status == 403
|
||||
assert repository.task == before
|
||||
assert repository.feedback == repository.evidence == repository.requests == {}
|
||||
return
|
||||
|
||||
await getattr(service, method)(task["task_id"], command)
|
||||
assert repository.task["status"] == ("completed" if operation == "feedback" else "cancelled")
|
||||
# Check scope before idempotency replay as well as before the first write.
|
||||
await getattr(service, method)(task["task_id"], command)
|
||||
before = deepcopy(repository.task)
|
||||
other = OperationalTaskService(repository, context, set(), clock=lambda: now,
|
||||
execution_scope="development" if server_scope == "production" else "production")
|
||||
with pytest.raises(DomainError) as error:
|
||||
await getattr(other, method)(task["task_id"], command)
|
||||
assert error.value.code == "EXECUTION_SCOPE_MISMATCH" and error.value.status == 403
|
||||
assert repository.task == before
|
||||
|
||||
@@ -226,7 +226,7 @@ async def test_sample_sync_rejects_cross_project_data_before_writing():
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize("targeted", [False, True])
|
||||
@pytest.mark.parametrize("order_revision", [1, 2])
|
||||
@pytest.mark.parametrize("order_revision", [1, 4])
|
||||
@pytest.mark.parametrize(
|
||||
("investigation_status", "expected_feedback_count", "routes_to_investigation"),
|
||||
[
|
||||
@@ -254,9 +254,13 @@ async def test_created_work_order_receipt_preserves_projection_and_routes_active
|
||||
"updated_at": "2026-09-23T09:00:00Z",
|
||||
"title": "采样工单", "priority": "normal",
|
||||
"work_order_type": "general", "created_at": "2026-09-23T08:00:00Z",
|
||||
"scheduled_at": None,
|
||||
"scheduled_at": "2026-09-23T08:30:00Z",
|
||||
"activity": ([{"sequence": 2, "occurred_at": "2026-09-23T08:45:00Z", "actor": "现场人员",
|
||||
"kind": "feedback", "title": "现场反馈", "detail": "复测余氯0.2mg/L"}]
|
||||
"kind": "feedback", "title": "现场反馈", "detail": "复测余氯0.2mg/L"},
|
||||
{"sequence": 3, "occurred_at": "2026-09-23T08:50:00Z", "actor": "调度员",
|
||||
"kind": "updated", "title": "工单信息更新", "detail": "更新了状态。"},
|
||||
{"sequence": 4, "occurred_at": "2026-09-23T09:00:00Z", "actor": "调度员",
|
||||
"kind": "updated", "title": "工单信息更新", "detail": "更新了优先级。"}]
|
||||
if order_revision > 1 else []),
|
||||
}
|
||||
if targeted:
|
||||
|
||||
Reference in New Issue
Block a user