706 lines
29 KiB
Python
706 lines
29 KiB
Python
# #region Test.McpRestErrorParity [C:5] [TYPE Module] [SEMANTICS test,mcp,parity,rest,error,lifecycle,baseline,rbac,disabled,investigation]
|
|
# @BRIEF 050 T044 residual: one hardcoded offline matrix proves the REST route and the MCP tool
|
|
# return the same typed error code for lifecycle, baseline, auth, disabled-automation and
|
|
# investigation-boundary cases.
|
|
# @RELATION VERIFIES -> [McpServer.ScenarioTools]
|
|
# @RELATION VERIFIES -> [McpServer.RbacServer]
|
|
# @RELATION VERIFIES -> [Api.ScenarioExecution.Routes.Start]
|
|
# @RELATION VERIFIES -> [Api.ScenarioAutomation.Routes]
|
|
# @RELATION VERIFIES -> [Api.ScenarioAnalytics.Routes]
|
|
# @RELATION VERIFIES -> [ScenarioExecution.StartError.Classify]
|
|
# @TEST_INVARIANT transport_parity -> each case's REST typed code equals the MCP typed code (or its
|
|
# documented permission equivalence), because both share classify_start_error.
|
|
# -> VERIFIED_BY: every test_*_parity case below.
|
|
# @TEST_INVARIANT production_gate_parity -> a PROD start is gated on both surfaces; the same
|
|
# principal reaches pending_approval on both. -> VERIFIED_BY: test_prod_start_*.
|
|
# @TEST_EDGE no_investigation_bypass -> the MCP catalogue exposes no case/queue tool, so the REST
|
|
# object ACL cannot be bypassed over MCP.
|
|
# @TEST_FIXTURE hardcoded graphs + publishedCatalog snapshots -> built in this module, no live stand.
|
|
# @RATIONALE The MCP tool once returned raw exception text while REST collapsed conflicts to
|
|
# RUN_START_CONFLICT; a shared classifier plus this matrix pin the two transports together.
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import secrets
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.app import app
|
|
from src.core.auth.security import get_password_hash
|
|
from src.core.database import SessionLocal
|
|
from src.dependencies import get_config_manager, get_current_user
|
|
from src.mcp_server import server as mcp_server
|
|
from src.mcp_server.server import (
|
|
_MCP_CATALOG_BY_NAME,
|
|
_access_token_context,
|
|
create_mcp_asgi_app,
|
|
)
|
|
import src.mcp_server.rbac_server as rbac_server_module
|
|
import src.mcp_server.tools_scenario as tools_scenario_module
|
|
from src.models.auth import Permission, Role, User
|
|
import src.models.scenario_investigation as investigation_models
|
|
from src.models.scenario_registry import ScenarioRegistryEntry, ScenarioRevision
|
|
from src.models.scenario_run import ScenarioRun
|
|
from src.services.dashboard_testing.scenario.templates import (
|
|
ACTION_REGISTRY_VERSION,
|
|
action_registry_fingerprint,
|
|
resolve_action_descriptor,
|
|
)
|
|
|
|
_ENVIRONMENTS = {
|
|
"env-dev": SimpleNamespace(id="env-dev", stage="DEV", is_production=False),
|
|
"env-prod": SimpleNamespace(id="env-prod", stage="PROD", is_production=True),
|
|
}
|
|
_INVESTIGATION_TOOL_TOKENS = ("investigation", "case", "queue")
|
|
_HEX40 = "c" * 40
|
|
_HEX64 = "a" * 64
|
|
_HEX64B = "b" * 64
|
|
_BASELINE_SET = "ss-parity-set"
|
|
_BASELINE_VERSION = "7"
|
|
|
|
|
|
# #region Test.McpRestErrorParity.Fixtures [C:4] [TYPE Function]
|
|
# @ingroup Test.McpRestErrorParity
|
|
# @BRIEF Hardcoded graphs, catalogue snapshots and the shared principal/DB harness.
|
|
def _config_manager() -> SimpleNamespace:
|
|
return SimpleNamespace(get_environment=lambda env_id: _ENVIRONMENTS.get(env_id))
|
|
|
|
|
|
def _patch_config(monkeypatch) -> None:
|
|
manager = _config_manager()
|
|
monkeypatch.setattr(rbac_server_module, "get_config_manager", lambda: manager)
|
|
monkeypatch.setattr(tools_scenario_module, "get_config_manager", lambda: manager)
|
|
|
|
|
|
def _step(step_id: str, tool: str, action: str, **extra) -> dict:
|
|
return {
|
|
"logical_step_id": step_id,
|
|
"tool": tool,
|
|
"action": action,
|
|
"action_descriptor": resolve_action_descriptor(
|
|
tool=tool,
|
|
action=action,
|
|
registry_version=ACTION_REGISTRY_VERSION,
|
|
registry_hash=action_registry_fingerprint(),
|
|
).snapshot(),
|
|
**extra,
|
|
}
|
|
|
|
|
|
def _plain_graph() -> dict:
|
|
return {
|
|
"schema_version": 1,
|
|
"action_registry_version": ACTION_REGISTRY_VERSION,
|
|
"action_registry_hash": action_registry_fingerprint(),
|
|
"environment_ids": ["env-dev"],
|
|
"steps": [_step("s1_assert", "assertion", "structural_assert", actual=1, expected=1)],
|
|
"dependencies": [],
|
|
}
|
|
|
|
|
|
def _baseline_graph() -> dict:
|
|
graph = _plain_graph()
|
|
graph["baselines"] = {"revenue_sum": {"reference": "baseline-v1", "policy": "exact"}}
|
|
return graph
|
|
|
|
|
|
def _human_graph() -> dict:
|
|
return {
|
|
"schema_version": 1,
|
|
"action_registry_version": ACTION_REGISTRY_VERSION,
|
|
"action_registry_hash": action_registry_fingerprint(),
|
|
"environment_ids": ["env-dev"],
|
|
"steps": [_step("s1_human", "human", "human_checkpoint")],
|
|
"dependencies": [],
|
|
}
|
|
|
|
|
|
def _approved_entry(baseline_id: str = "bl-1", **overrides) -> dict:
|
|
item = {
|
|
"entry": {
|
|
"kind": "metric",
|
|
"release_version": "v1.0.0",
|
|
"release_commit_hash": _HEX40,
|
|
"source_response_hash": _HEX64,
|
|
},
|
|
"baseline_id": baseline_id,
|
|
"baseline_revision_id": f"{baseline_id}-rev",
|
|
"coordinate_hash": _HEX64,
|
|
"entry_digest": _HEX64,
|
|
"capture_artifact_id": "artifact-1",
|
|
"capture_profile_hash": _HEX64,
|
|
"status": "approved",
|
|
}
|
|
item.update(overrides)
|
|
return item
|
|
|
|
|
|
def _catalog_snapshot(
|
|
*,
|
|
revision_digest: str = _HEX64,
|
|
entries: list[dict] | None = None,
|
|
) -> dict:
|
|
return {
|
|
"baseline_set_id": _BASELINE_SET,
|
|
"baseline_set_version": _BASELINE_VERSION,
|
|
"catalog_digest": _HEX64,
|
|
"release_id": "rel-1",
|
|
"baseline_family": _HEX64B,
|
|
"catalog_revision": {
|
|
"catalog_revision_id": "cr-1",
|
|
"catalog_digest": revision_digest,
|
|
"publication": {
|
|
"state": "published",
|
|
"commit_hash": _HEX40,
|
|
"published_receipt_id": "rcpt-1",
|
|
},
|
|
"entry_revisions": entries if entries is not None else [_approved_entry()],
|
|
},
|
|
}
|
|
|
|
|
|
def _start_body(scenario_id: str, revision_id: str, *, environment_id: str, **extra) -> dict:
|
|
return {
|
|
"scenario_id": scenario_id,
|
|
"revision_id": revision_id,
|
|
"environment_id": environment_id,
|
|
"params": {},
|
|
**extra,
|
|
}
|
|
|
|
|
|
def _mcp_call(server, tool_name: str, arguments: dict) -> dict:
|
|
async def _run():
|
|
return await server.call_tool(tool_name, arguments)
|
|
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
try:
|
|
result = loop.run_until_complete(_run())
|
|
except Exception as exc: # transport/tool error is the assertion target, not a bug
|
|
return {"status": "raised", "error": str(exc)}
|
|
finally:
|
|
loop.close()
|
|
return result[1] if isinstance(result, tuple) else result
|
|
|
|
|
|
class ParityEnv:
|
|
"""Shared global-DB harness: REST over src.app, MCP over the probe server."""
|
|
|
|
def __init__(self) -> None:
|
|
self.scenario_ids: list[str] = []
|
|
self.usernames: list[str] = []
|
|
self.role_names: list[str] = []
|
|
self.run_ids: list[str] = []
|
|
|
|
def seed_scenario(self, graph: dict) -> tuple[str, str]:
|
|
scenario_id = str(uuid4())
|
|
revision_id = str(uuid4())
|
|
with SessionLocal() as db:
|
|
db.add(ScenarioRegistryEntry(
|
|
scenario_id=scenario_id, scenario_key=f"parity-{scenario_id[:8]}",
|
|
name="parity scenario", dashboard_id=80, environment_ids=["env-dev"],
|
|
owner_id="owner", owner_username="owner",
|
|
lifecycle_status="READY", validation_status="valid",
|
|
current_revision_id=revision_id,
|
|
))
|
|
self._add_revision(db, scenario_id, revision_id, graph, "current")
|
|
db.commit()
|
|
self.scenario_ids.append(scenario_id)
|
|
return scenario_id, revision_id
|
|
|
|
def add_stale_revision(self, scenario_id: str) -> str:
|
|
revision_id = str(uuid4())
|
|
with SessionLocal() as db:
|
|
self._add_revision(db, scenario_id, revision_id, _plain_graph(), "candidate")
|
|
db.commit()
|
|
return revision_id
|
|
|
|
@staticmethod
|
|
def _add_revision(db, scenario_id: str, revision_id: str, graph: dict, activation: str) -> None:
|
|
db.add(ScenarioRevision(
|
|
revision_id=revision_id, scenario_id=scenario_id, content_hash=_HEX64,
|
|
graph_snapshot=graph, execution_template_hash="", template_version="v1",
|
|
schema_version=1, compatibility_family="default", change_summary={},
|
|
created_by="owner", activation_status=activation,
|
|
))
|
|
|
|
def seed_user(self, *perms: tuple[str, str], admin: bool = False) -> tuple[str, str]:
|
|
suffix = secrets.token_hex(4)
|
|
username = f"parity-user-{suffix}"
|
|
role_name = f"ParityRole-{suffix}"
|
|
role = Role(
|
|
name=role_name, is_admin=admin,
|
|
permissions=[Permission(resource=r, action=a) for r, a in perms],
|
|
)
|
|
user = User(username=username, password_hash=get_password_hash("pw"), is_active=True, roles=[role])
|
|
with SessionLocal() as db:
|
|
db.add(user)
|
|
db.commit()
|
|
user_id = str(user.id)
|
|
self.usernames.append(username)
|
|
self.role_names.append(role_name)
|
|
return username, user_id
|
|
|
|
def rest_client(self, username: str, user_id: str, *perms: tuple[str, str], admin: bool = False) -> TestClient:
|
|
# Object ACL (SEC-01): the REST run routes require scenario ownership. Parity fixtures seed
|
|
# the sentinel owner "owner"; when acting as a non-admin principal, transfer those seeded
|
|
# scenarios to the acting user so the tests exercise the route contract, not a 404.
|
|
if not admin:
|
|
with SessionLocal() as db:
|
|
db.query(ScenarioRegistryEntry).filter(
|
|
ScenarioRegistryEntry.owner_id == "owner"
|
|
).update(
|
|
{ScenarioRegistryEntry.owner_id: user_id, ScenarioRegistryEntry.owner_username: username},
|
|
synchronize_session=False,
|
|
)
|
|
db.commit()
|
|
principal = SimpleNamespace(
|
|
id=user_id, username=username,
|
|
roles=[SimpleNamespace(
|
|
is_admin=admin,
|
|
permissions=[SimpleNamespace(resource=r, action=a) for r, a in perms],
|
|
)],
|
|
)
|
|
app.dependency_overrides[get_current_user] = lambda: principal
|
|
app.dependency_overrides[get_config_manager] = _config_manager
|
|
return TestClient(app)
|
|
|
|
def mcp_as(self, username: str | None, principal_type: str = "user"):
|
|
server = mcp_server._build_probe_server()
|
|
token = _access_token_context.set(mcp_server.AccessToken(
|
|
token="parity-token", client_id="parity-client", scopes=["mcp"],
|
|
subject=username if principal_type == "user" else None,
|
|
claims={"principal_type": principal_type},
|
|
))
|
|
return server, token
|
|
|
|
def track_runs(self, prefix: str) -> None:
|
|
with SessionLocal() as db:
|
|
rows = db.query(ScenarioRun).filter(ScenarioRun.idempotency_key.like(f"{prefix}%")).all()
|
|
self.run_ids.extend(row.id for row in rows)
|
|
|
|
def cleanup(self) -> None:
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
app.dependency_overrides.pop(get_config_manager, None)
|
|
with SessionLocal() as db:
|
|
if self.run_ids:
|
|
db.query(ScenarioRun).filter(ScenarioRun.id.in_(self.run_ids)).delete(synchronize_session=False)
|
|
for scenario_id in self.scenario_ids:
|
|
db.query(ScenarioRegistryEntry).filter_by(scenario_id=scenario_id).delete()
|
|
db.query(ScenarioRevision).filter_by(scenario_id=scenario_id).delete()
|
|
for username in self.usernames:
|
|
user = db.query(User).filter(User.username == username).first()
|
|
if user is not None:
|
|
db.delete(user)
|
|
db.flush()
|
|
for role_name in self.role_names:
|
|
role = db.query(Role).filter(Role.name == role_name).first()
|
|
if role is not None:
|
|
db.delete(role)
|
|
db.commit()
|
|
|
|
|
|
@pytest.fixture
|
|
def parity():
|
|
env = ParityEnv()
|
|
try:
|
|
yield env
|
|
finally:
|
|
env.cleanup()
|
|
# #endregion Test.McpRestErrorParity.Fixtures
|
|
|
|
|
|
# #region Test.McpRestErrorParity.Lifecycle [C:4] [TYPE Function]
|
|
# @ingroup Test.McpRestErrorParity
|
|
# @BRIEF Lifecycle start errors carry one typed code on both transports.
|
|
def test_environment_not_configured_parity(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, revision_id = parity.seed_scenario(_plain_graph())
|
|
username, user_id = parity.seed_user(("scenario", "RUN"))
|
|
rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
|
|
body = _start_body(scenario_id, revision_id, environment_id="env-ghost")
|
|
|
|
rest_response = rest.post(
|
|
"/api/scenario-runs", headers={"Idempotency-Key": f"parity-env-{uuid4()}"}, json=body,
|
|
)
|
|
assert rest_response.status_code == 422
|
|
assert rest_response.json()["detail"]["code"] == "ENVIRONMENT_NOT_CONFIGURED"
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "start_scenario_run", {
|
|
"request": {**body, "idempotency_key": f"parity-env-{uuid4()}"},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "blocked"
|
|
assert mcp_result["error"] == "ENVIRONMENT_NOT_CONFIGURED"
|
|
|
|
|
|
def test_run_start_conflict_parity(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, _ = parity.seed_scenario(_plain_graph())
|
|
stale_revision = parity.add_stale_revision(scenario_id)
|
|
username, user_id = parity.seed_user(("scenario", "RUN"))
|
|
rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
|
|
body = _start_body(scenario_id, stale_revision, environment_id="env-dev")
|
|
|
|
rest_response = rest.post(
|
|
"/api/scenario-runs", headers={"Idempotency-Key": f"parity-conflict-{uuid4()}"}, json=body,
|
|
)
|
|
assert rest_response.status_code == 409
|
|
assert rest_response.json()["detail"]["code"] == "RUN_START_CONFLICT"
|
|
assert "revision mismatch" in rest_response.json()["detail"]["detail"]
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "start_scenario_run", {
|
|
"request": {**body, "idempotency_key": f"parity-conflict-{uuid4()}"},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "blocked"
|
|
assert mcp_result["error"] == "RUN_START_CONFLICT"
|
|
assert "revision mismatch" in mcp_result["detail"]
|
|
|
|
|
|
def test_idempotency_key_reused_parity(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, revision_id = parity.seed_scenario(_plain_graph())
|
|
username, user_id = parity.seed_user(("scenario", "RUN"))
|
|
rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
|
|
key = f"parity-idem-{uuid4()}"
|
|
parity.track_runs("parity-idem-")
|
|
|
|
first = rest.post(
|
|
"/api/scenario-runs", headers={"Idempotency-Key": key},
|
|
json=_start_body(scenario_id, revision_id, environment_id="env-dev", params={"n": 1}),
|
|
)
|
|
assert first.status_code == 201
|
|
second = rest.post(
|
|
"/api/scenario-runs", headers={"Idempotency-Key": key},
|
|
json=_start_body(scenario_id, revision_id, environment_id="env-dev", params={"n": 2}),
|
|
)
|
|
assert second.status_code == 409
|
|
assert second.json()["detail"]["code"] == "IDEMPOTENCY_KEY_REUSED"
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "start_scenario_run", {
|
|
"request": _start_body(scenario_id, revision_id, environment_id="env-dev", params={"n": 2})
|
|
| {"idempotency_key": key},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "blocked"
|
|
assert mcp_result["error"] == "IDEMPOTENCY_KEY_REUSED"
|
|
|
|
|
|
def test_prod_approval_required_maps_to_mcp_permission(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, revision_id = parity.seed_scenario(_plain_graph())
|
|
username, user_id = parity.seed_user(("scenario", "RUN"))
|
|
rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
|
|
body = _start_body(scenario_id, revision_id, environment_id="env-prod")
|
|
|
|
rest_response = rest.post(
|
|
"/api/scenario-runs", headers={"Idempotency-Key": f"parity-prod-{uuid4()}"}, json=body,
|
|
)
|
|
assert rest_response.status_code == 403
|
|
assert rest_response.json()["detail"]["code"] == "PROD_APPROVAL_REQUIRED"
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "start_scenario_run", {
|
|
"request": {**body, "idempotency_key": f"parity-prod-{uuid4()}"},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
# Documented equivalence: REST 403 PROD_APPROVAL_REQUIRED == MCP fail-closed permission denial.
|
|
assert mcp_result["status"] == "raised"
|
|
assert mcp_result["error"] == "permission_denied"
|
|
|
|
|
|
def test_prod_start_reaches_pending_approval_on_both_surfaces(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, revision_id = parity.seed_scenario(_plain_graph())
|
|
username, user_id = parity.seed_user(("scenario", "RUN"), ("scenario", "RUN_PROD"))
|
|
rest = parity.rest_client(username, user_id, ("scenario", "RUN"), ("scenario", "RUN_PROD"))
|
|
body = _start_body(scenario_id, revision_id, environment_id="env-prod")
|
|
|
|
rest_response = rest.post(
|
|
"/api/scenario-runs", headers={"Idempotency-Key": f"parity-prod-ok-{uuid4()}"}, json=body,
|
|
)
|
|
assert rest_response.status_code == 201, rest_response.text
|
|
assert rest_response.json()["status"] == "pending_approval"
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "start_scenario_run", {
|
|
"request": {**body, "idempotency_key": f"parity-prod-ok-{uuid4()}"},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "pending_approval"
|
|
parity.track_runs("parity-prod-ok-")
|
|
# #endregion Test.McpRestErrorParity.Lifecycle
|
|
|
|
|
|
# #region Test.McpRestErrorParity.Baseline [C:4] [TYPE Function]
|
|
# @ingroup Test.McpRestErrorParity
|
|
# @BRIEF Every 037 baseline-resolution code is identical on REST and MCP.
|
|
@pytest.mark.parametrize(
|
|
"code,snapshot,with_selector",
|
|
[
|
|
("BASELINE_MISSING", None, False),
|
|
("BASELINE_NOT_PUBLISHED", {"catalog": "working-tree-yaml"}, True),
|
|
("BASELINE_STALE", _catalog_snapshot(revision_digest="not-a-sha256"), True),
|
|
(
|
|
"BASELINE_AMBIGUOUS",
|
|
_catalog_snapshot(entries=[
|
|
_approved_entry("bl-dup", coordinate_hash="1" * 64),
|
|
_approved_entry("bl-dup", baseline_revision_id="bl-dup-rev-2", coordinate_hash="2" * 64),
|
|
]),
|
|
True,
|
|
),
|
|
(
|
|
"BASELINE_EVIDENCE_UNAVAILABLE",
|
|
_catalog_snapshot(entries=[
|
|
_approved_entry(**{"entry": {
|
|
"kind": "metric", "release_version": "v1.0.0",
|
|
"release_commit_hash": _HEX40, "source_response_hash": "short",
|
|
}}),
|
|
]),
|
|
True,
|
|
),
|
|
],
|
|
)
|
|
def test_baseline_error_codes_parity(parity, monkeypatch, code, snapshot, with_selector) -> None:
|
|
monkeypatch.setattr(
|
|
"src.services.dashboard_testing.execution.start_run.load_published_catalog",
|
|
lambda injected=None: snapshot,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.dashboard_testing.execution.start_run.load_published_catalog_from_config",
|
|
lambda *_args, **_kwargs: None,
|
|
)
|
|
_patch_config(monkeypatch)
|
|
scenario_id, revision_id = parity.seed_scenario(_baseline_graph())
|
|
username, user_id = parity.seed_user(("scenario", "RUN"))
|
|
rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
|
|
body = _start_body(scenario_id, revision_id, environment_id="env-dev")
|
|
if with_selector:
|
|
body["baseline_set"] = _BASELINE_SET
|
|
body["baseline_set_version"] = _BASELINE_VERSION
|
|
|
|
rest_response = rest.post(
|
|
"/api/scenario-runs", headers={"Idempotency-Key": f"parity-baseline-{uuid4()}"}, json=body,
|
|
)
|
|
assert rest_response.status_code == 422, rest_response.text
|
|
assert rest_response.json()["detail"]["code"] == code
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "start_scenario_run", {
|
|
"request": {"idempotency_key": f"parity-baseline-{uuid4()}", **{
|
|
k: v for k, v in body.items() if k != "idempotency_key"
|
|
}},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "blocked"
|
|
assert mcp_result["error"] == code
|
|
# #endregion Test.McpRestErrorParity.Baseline
|
|
|
|
|
|
# #region Test.McpRestErrorParity.DisabledAutomation [C:4] [TYPE Function]
|
|
# @ingroup Test.McpRestErrorParity
|
|
# @BRIEF Disabled/ineligible automation validation uses one code per failure on both surfaces.
|
|
def test_automation_ineligible_human_step_parity(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, revision_id = parity.seed_scenario(_human_graph())
|
|
username, user_id = parity.seed_user(admin=True)
|
|
rest = parity.rest_client(username, user_id, admin=True)
|
|
|
|
request_body = {
|
|
"scenario_id": scenario_id, "environment_id": "env-dev", "cron_expr": "0 * * * *",
|
|
"revision_policy": "pinned", "revision_id": revision_id,
|
|
}
|
|
rest_response = rest.post("/api/scenario-automation/schedules", json=request_body)
|
|
assert rest_response.status_code == 409
|
|
assert rest_response.json()["detail"]["code"] == "AUTOMATION_INELIGIBLE_HUMAN_STEP"
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "upsert_scenario_schedule", {
|
|
"request": {**request_body, "idempotency_key": f"parity-human-{uuid4()}"},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "raised"
|
|
assert "AUTOMATION_INELIGIBLE_HUMAN_STEP" in mcp_result["error"]
|
|
|
|
|
|
def test_invalid_missed_policy_parity(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, revision_id = parity.seed_scenario(_plain_graph())
|
|
username, user_id = parity.seed_user(admin=True)
|
|
rest = parity.rest_client(username, user_id, admin=True)
|
|
|
|
request_body = {
|
|
"scenario_id": scenario_id, "environment_id": "env-dev", "cron_expr": "0 * * * *",
|
|
"revision_policy": "pinned", "revision_id": revision_id,
|
|
"missed_execution_policy": "bogus",
|
|
}
|
|
rest_response = rest.post("/api/scenario-automation/schedules", json=request_body)
|
|
assert rest_response.status_code == 422
|
|
assert rest_response.json()["detail"]["code"] == "INVALID_MISSED_POLICY"
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "upsert_scenario_schedule", {
|
|
"request": {**request_body, "idempotency_key": f"parity-missed-{uuid4()}"},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "raised"
|
|
assert "INVALID_MISSED_POLICY" in mcp_result["error"]
|
|
|
|
|
|
def test_invalid_trigger_parity(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, revision_id = parity.seed_scenario(_plain_graph())
|
|
username, user_id = parity.seed_user(admin=True)
|
|
rest = parity.rest_client(username, user_id, admin=True)
|
|
|
|
request_body = {
|
|
"scenario_id": scenario_id, "environment_id": "env-dev", "trigger": "bogus",
|
|
"revision_policy": "pinned", "revision_id": revision_id,
|
|
}
|
|
rest_response = rest.post("/api/scenario-automation/trigger-rules", json=request_body)
|
|
assert rest_response.status_code == 422
|
|
assert rest_response.json()["detail"]["code"] == "INVALID_TRIGGER"
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "upsert_scenario_trigger_rule", {
|
|
"request": {**request_body, "idempotency_key": f"parity-trigger-{uuid4()}"},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "raised"
|
|
assert "INVALID_TRIGGER" in mcp_result["error"]
|
|
|
|
|
|
def test_non_active_revision_schedule_parity(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, _ = parity.seed_scenario(_plain_graph())
|
|
stale_revision = parity.add_stale_revision(scenario_id)
|
|
username, user_id = parity.seed_user(admin=True)
|
|
rest = parity.rest_client(username, user_id, admin=True)
|
|
|
|
request_body = {
|
|
"scenario_id": scenario_id, "environment_id": "env-dev", "cron_expr": "0 * * * *",
|
|
"revision_policy": "pinned", "revision_id": stale_revision,
|
|
}
|
|
rest_response = rest.post("/api/scenario-automation/schedules", json=request_body)
|
|
assert rest_response.status_code == 409
|
|
expected = "revision must be the active scenario revision"
|
|
assert rest_response.json()["detail"]["code"] == expected
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "upsert_scenario_schedule", {
|
|
"request": {**request_body, "idempotency_key": f"parity-stale-{uuid4()}"},
|
|
})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "raised"
|
|
assert expected in mcp_result["error"]
|
|
# #endregion Test.McpRestErrorParity.DisabledAutomation
|
|
|
|
|
|
# #region Test.McpRestErrorParity.AuthAndRbac [C:4] [TYPE Function]
|
|
# @ingroup Test.McpRestErrorParity
|
|
# @BRIEF Auth transport, human-gate decision authority and investigation ACL parity.
|
|
def test_service_principal_cannot_decide_human_gate_parity(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
username, user_id = parity.seed_user(("scenario", "RUN"))
|
|
rest = parity.rest_client(username, user_id, ("scenario", "RUN"))
|
|
|
|
# REST approval decision is gated on scenario RUN_PROD before the handler runs.
|
|
rest_response = rest.post(
|
|
f"/api/scenario-runs/{uuid4()}/approval/decision", json={"decision": "approve"},
|
|
)
|
|
assert rest_response.status_code == 403
|
|
|
|
# MCP: a service principal can never decide a human gate.
|
|
server, token = parity.mcp_as(None, principal_type="service")
|
|
try:
|
|
mcp_result = _mcp_call(server, "decide_approval", {"gate_id": "g-1", "decision": "approve"})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["error"] == "permission_denied"
|
|
|
|
|
|
def test_investigation_case_acl_has_no_mcp_bypass(parity, monkeypatch) -> None:
|
|
_patch_config(monkeypatch)
|
|
scenario_id, _ = parity.seed_scenario(_plain_graph())
|
|
username, user_id = parity.seed_user(("scenario:result", "VIEW"))
|
|
case_id = str(uuid4())
|
|
with SessionLocal() as db:
|
|
db.add(investigation_models.InvestigationCase(
|
|
id=case_id, fingerprint=f"fp-{case_id[:8]}", scenario_id=scenario_id,
|
|
status="open", decision_version=1, evidence_snapshot={}, linked_run_ids=[],
|
|
owner_id="other-owner",
|
|
))
|
|
db.commit()
|
|
|
|
rest = parity.rest_client(username, user_id, ("scenario:result", "VIEW"))
|
|
rest_response = rest.get(f"/api/scenario-analytics/cases/{case_id}")
|
|
assert rest_response.status_code == 403
|
|
assert rest_response.json()["detail"]["code"] == "CASE_ACL_DENIED"
|
|
|
|
# Investigation MCP tools DO exist on this release (G-INVESTIGATION-MCP CLOSED); the parity
|
|
# requirement is that they cannot BYPASS the REST case ACL. A foreign case must collapse to
|
|
# not_found through MCP exactly as REST answers 403, with no case content disclosed.
|
|
offenders = [
|
|
name for name in _MCP_CATALOG_BY_NAME
|
|
if any(token in name.lower() for token in _INVESTIGATION_TOOL_TOKENS)
|
|
]
|
|
assert "get_investigation_case" in offenders, offenders
|
|
|
|
server, token = parity.mcp_as(username)
|
|
try:
|
|
mcp_result = _mcp_call(server, "get_investigation_case", {"request": {"case_id": case_id}})
|
|
finally:
|
|
_access_token_context.reset(token)
|
|
assert mcp_result["status"] == "not_found", mcp_result
|
|
assert "fingerprint" not in mcp_result and "scenario_id" not in mcp_result, mcp_result
|
|
|
|
with SessionLocal() as db:
|
|
db.query(investigation_models.InvestigationCase).filter_by(id=case_id).delete()
|
|
db.commit()
|
|
|
|
|
|
def test_unauthenticated_transport_parity() -> None:
|
|
rest_client = TestClient(app)
|
|
rest_response = rest_client.post(
|
|
"/api/scenario-runs",
|
|
headers={"Idempotency-Key": "parity-unauth"},
|
|
json={"scenario_id": "s", "revision_id": "r", "environment_id": "env-dev", "params": {}},
|
|
)
|
|
assert rest_response.status_code == 401
|
|
|
|
mcp_client = TestClient(create_mcp_asgi_app())
|
|
mcp_response = mcp_client.post("/", json={"jsonrpc": "2.0", "id": 1, "method": "ping"})
|
|
assert mcp_response.status_code == 401
|
|
assert mcp_response.json()["error"] == "authentication_required"
|
|
# #endregion Test.McpRestErrorParity.AuthAndRbac
|
|
|
|
# #endregion Test.McpRestErrorParity
|