75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
import scripts.smoke_1c_write_plan_safety as smoke # noqa: E402
|
|
|
|
|
|
def test_repository_gate_falls_back_to_non_mutating_path_route_probe(monkeypatch) -> None:
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake_rpc_call(
|
|
_endpoint_url: str,
|
|
method: str,
|
|
payload: dict[str, Any],
|
|
_timeout: float,
|
|
*,
|
|
transport: str,
|
|
session_id: str | None,
|
|
) -> dict[str, Any]:
|
|
calls.append((method, payload))
|
|
assert transport == "rest"
|
|
assert session_id is None
|
|
if method == "metadata.write":
|
|
if payload.get("mode") == "apply":
|
|
return {
|
|
"schema": "onec_repository_write_gate.v1",
|
|
"status": "blocked",
|
|
"error": "needs_repository_lock",
|
|
}
|
|
return {
|
|
"schema": "onec_metadata_write.v1",
|
|
"status": "blocked",
|
|
"error": "write_plan_required",
|
|
"routed_method": "metadata.write.plan",
|
|
"target_kind": "form",
|
|
}
|
|
intent = payload.get("intent") or {}
|
|
if intent.get("current_text") is not None:
|
|
return {
|
|
"status": "blocked",
|
|
"allowed": False,
|
|
"problems": [{"code": "control_fragment_drift"}],
|
|
}
|
|
if intent.get("control_fragment"):
|
|
return {
|
|
"status": "planned",
|
|
"allowed": True,
|
|
"route": {"apply_method": "metadata.module.write_apply"},
|
|
"target": {"target_kind": "module"},
|
|
}
|
|
return {
|
|
"status": "blocked",
|
|
"allowed": False,
|
|
"problems": [{"code": "missing_control_fragment"}],
|
|
}
|
|
|
|
monkeypatch.setattr(smoke, "rpc_call", fake_rpc_call)
|
|
|
|
result = smoke.run_smoke("http://adapter", "upo_test", 30, transport="rest")
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["failures"] == []
|
|
assert result["checks"]["blocked_effective_form_path"]["apply_gate"] == {
|
|
"status": "blocked",
|
|
"error": "needs_repository_lock",
|
|
}
|
|
metadata_write_modes = [payload["mode"] for method, payload in calls if method == "metadata.write"]
|
|
assert metadata_write_modes == ["apply", "plan"]
|