78 lines
3.6 KiB
Python
78 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
|
|
from backend.app.db.models import AuditEvent
|
|
from backend.app.services import action_execution, deployment_broker
|
|
|
|
|
|
@pytest.mark.parametrize("outcome", ["success", "refused", "timeout"])
|
|
def test_actual_signed_specialization_transport_preserves_outcome(monkeypatch, outcome):
|
|
client = deployment_broker.DeploymentBrokerClient()
|
|
client.base_url = "https://controller.invalid"
|
|
monkeypatch.setattr(client, "is_configured", lambda: True)
|
|
monkeypatch.setattr(client, "_verify_option", lambda: True)
|
|
monkeypatch.setattr(deployment_broker.settings, "mcp_token", "x" * 64)
|
|
observed = []
|
|
real_client = httpx.Client
|
|
|
|
def respond(request):
|
|
assert request.method == "POST"
|
|
assert request.url.path == "/v1/appops/runner-fleet/specialize"
|
|
assert len(request.headers["X-ITWorx-Deployment-Signature"]) == 64
|
|
if outcome == "timeout":
|
|
raise httpx.ReadTimeout("Bearer SECRET-CANARY", request=request)
|
|
return httpx.Response(200 if outcome == "success" else 409, json={
|
|
"status": "specialized" if outcome == "success" else "upgrade_failed",
|
|
"reason": "runner_fleet_specialization_cgroup_inventory_unavailable",
|
|
})
|
|
|
|
def transport_client(**options):
|
|
observed.append(options)
|
|
return real_client(**options, transport=httpx.MockTransport(respond))
|
|
|
|
monkeypatch.setattr(deployment_broker.httpx, "Client", transport_client)
|
|
result = client.runner_fleet_specialize(reason="qualify exact source")
|
|
assert len(observed) == 1 # No automatic replay after transport loss/refusal.
|
|
assert observed[0]["verify"] is True
|
|
assert observed[0]["timeout"].as_dict() == {
|
|
"connect": 5, "read": 3700, "write": 5, "pool": 5,
|
|
}
|
|
assert "SECRET-CANARY" not in str(result)
|
|
if outcome == "timeout":
|
|
assert result == {
|
|
"status": "unknown", "reason": "runner_specialization_outcome_unknown",
|
|
"noMutation": False, "effectsPerformed": None,
|
|
}
|
|
else:
|
|
assert result["status"] == ("specialized" if outcome == "success" else "upgrade_failed")
|
|
assert result["reason"] == "runner_fleet_specialization_cgroup_inventory_unavailable"
|
|
|
|
|
|
@pytest.mark.parametrize("raised", [False, True])
|
|
def test_specialization_unknown_is_preserved_in_job_result_and_audit(tmp_path, monkeypatch, raised):
|
|
def request(*_, **__):
|
|
if raised:
|
|
raise RuntimeError("Bearer SECRET-CANARY")
|
|
return {"status": "unknown", "reason": "runner_specialization_outcome_unknown",
|
|
"noMutation": False, "effectsPerformed": None}
|
|
|
|
monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_fleet_specialize", request)
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'unknown.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
result = action_execution._execute_runner_fleet_upgrade_v2(
|
|
session, source="chatgpt-mcp", risk="high", reason="qualify exact source",
|
|
original_policy_decision="allowed", specialized=True,
|
|
)
|
|
assert result["success"] is False
|
|
assert result["controllerStatus"] == "unknown" and result["result"] == "unknown"
|
|
assert result["effectsPerformed"] is None
|
|
if not raised:
|
|
assert result["noMutation"] is False
|
|
audit = session.get(AuditEvent, result["auditId"])
|
|
assert audit is not None and audit.result == "unknown"
|
|
assert "SECRET-CANARY" not in str(result) + str(audit)
|