Spaces:
Running
Running
File size: 6,927 Bytes
6918275 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | """Observer attachment uses real service/API/client paths without controller authority."""
import threading
from unittest.mock import AsyncMock
import pytest
from flask import Flask, Response
from helpers import api, cache
from plugins.argus_supervisor.helpers import service
from plugins.argus_supervisor.helpers.a2a_client import OpenTrainingA2AClient
from plugins.argus_supervisor.helpers.campaign_store import CampaignStore
from plugins.argus_supervisor.helpers.heartbeat import Supervisor
from plugins.argus_supervisor.helpers.models import SupervisorError
@pytest.fixture
def observer_service(tmp_path, monkeypatch):
calls = []
async def transport(method, suffix, payload):
if method == "GET":
return {"name": "Controlled OpenTraining", "skills": [], "capabilities": {}}
calls.append(payload["method"])
assert payload["method"] == "tasks/get"
assert payload["params"]["id"] == "existing-task"
return {
"jsonrpc": "2.0",
"id": payload["id"],
"result": {"id": "existing-task", "status": {"state": "completed"}},
}
client = OpenTrainingA2AClient("https://training.example/a2a", transport=transport)
backend = AsyncMock()
supervisor = Supervisor(CampaignStore(tmp_path), client, backend)
monkeypatch.setenv("ARGUS_SUPERVISOR_ENABLED", "1")
monkeypatch.setattr(service, "_instance", supervisor)
for name in ("claim_control", "read_control", "mutate"):
monkeypatch.setattr(client, name, AsyncMock(side_effect=AssertionError(name)))
return supervisor, calls, backend
def create_observer():
result = service.create(
{
"campaign_id": "observer-test",
"objective": "Observe existing task",
"remote_task_id": "existing-task",
"max_attempts": 1,
}
)
assert result["mode"] == "observer"
assert result["controller_id"] is None and result["controller_epoch"] is None
assert result["remote_task_id"] == "existing-task"
return result
def assert_persisted(supervisor, calls, backend):
campaign = supervisor.store.load("observer-test")
assert campaign.remote_task_id == "existing-task"
assert campaign.remote_state == "completed" and campaign.last_seen is not None
assert campaign.mode == "observer"
assert campaign.controller_id is None and campaign.controller_epoch is None
assert calls == ["tasks/get"]
for name in ("claim_control", "read_control", "mutate"):
getattr(supervisor.client, name).assert_not_called()
backend.notify.assert_not_called()
backend.bind.assert_not_called()
return campaign
@pytest.mark.asyncio
async def test_service_observer_attach_poll_and_read_need_no_authority(
observer_service, monkeypatch
):
supervisor, calls, backend = observer_service
monkeypatch.setattr(
supervisor, "authority", AsyncMock(side_effect=AssertionError("authority"))
)
monkeypatch.setattr(
supervisor, "mutation", AsyncMock(side_effect=AssertionError("mutation"))
)
create_observer()
await service.action("observer-test", "observe", {})
assert_persisted(supervisor, calls, backend)
supervisor.authority.assert_not_called()
supervisor.mutation.assert_not_called()
# Read from a new store instance, not the in-memory campaign returned by action.
assert (
CampaignStore(supervisor.store.root).load("observer-test").remote_state
== "completed"
)
def test_real_authenticated_plugin_api_observer_golden_path(
observer_service, monkeypatch
):
supervisor, calls, backend = observer_service
monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")
monkeypatch.setattr(
supervisor, "authority", AsyncMock(side_effect=AssertionError("authority"))
)
monkeypatch.setattr(
supervisor, "mutation", AsyncMock(side_effect=AssertionError("mutation"))
)
app = Flask("observer_contract")
app.secret_key = "test-session"
app.add_url_rule("/", "serve_index", lambda: Response("index"))
app.add_url_rule("/login", "login_handler", lambda: Response("login"))
cache.clear(api.CACHE_AREA)
api.register_api_route(app, threading.RLock())
client = app.test_client()
with client.session_transaction() as session:
session["authentication"] = "hash"
session["csrf_token"] = "test-csrf"
headers = {"X-CSRF-Token": "test-csrf"}
prefix = "/api/plugins/argus_supervisor/"
try:
response = client.post(
prefix + "campaigns",
json={
"action": "create",
"campaign_id": "observer-test",
"objective": "Observe existing task",
"remote_task_id": "existing-task",
},
headers=headers,
)
assert response.status_code == 200
assert response.get_json()["campaign"]["mode"] == "observer"
response = client.post(
prefix + "campaign_action",
json={"campaign_id": "observer-test", "action": "observe"},
headers=headers,
)
assert response.status_code == 200
response = client.post(
prefix + "campaigns", json={"campaign_id": "observer-test"}, headers=headers
)
assert response.status_code == 200
persisted = response.get_json()["campaign"]
assert (
persisted["remote_state"] == "completed"
and persisted["last_seen"] is not None
)
assert (
persisted["remote_task_id"] == "existing-task"
and persisted["mode"] == "observer"
)
assert (
persisted["controller_id"] is None and persisted["controller_epoch"] is None
)
assert_persisted(supervisor, calls, backend)
supervisor.authority.assert_not_called()
supervisor.mutation.assert_not_called()
finally:
cache.clear(api.CACHE_AREA)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"operation,params",
[
("submit", {"text": "Do not send"}),
("intervene", {"instruction": "Do not send"}),
("cancel", {"reason": "Do not send"}),
("release", {}),
],
)
async def test_observer_remote_mutations_still_fail_closed(
observer_service, operation, params
):
supervisor, calls, backend = observer_service
create_observer()
with pytest.raises(SupervisorError, match="^stale_controller_epoch$"):
await service.action("observer-test", operation, params)
assert calls == []
for name in ("claim_control", "read_control", "mutate"):
getattr(supervisor.client, name).assert_not_called()
campaign = supervisor.store.load("observer-test")
assert campaign.mode == "observer" and campaign.controller_epoch is None
assert campaign.intent is None
backend.notify.assert_not_called()
|