49 lines
2.3 KiB
Python
49 lines
2.3 KiB
Python
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_model_api_exposes_governed_writer_without_legacy_global_enable():
|
|
source = (ROOT / "active-runtime/app/aios_model_api_staging.py").read_text(encoding="utf-8")
|
|
assert "'/createGovernedTask': ('/v2/governed-task'" in source
|
|
assert "'/commitFilesGoverned': ('/v2/commit-files'" in source
|
|
assert "path == '/api/governed-write-capabilities'" in source
|
|
assert "AIOS_GITOPS_WRITES_ENABLED" not in source or "legacy_global_writes_enabled" in source
|
|
assert "'writes_enabled': False" in source
|
|
assert "bridge_v2_json" in source
|
|
|
|
|
|
def test_model_api_routes_online_task_control_through_loopback_credential():
|
|
source = (ROOT / "active-runtime/app/aios_model_api_staging.py").read_text(encoding="utf-8")
|
|
assert "AIOS_AGENT_TASK_API_URL" in source
|
|
assert "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE" in source
|
|
assert "/createOnlineAgentTask" in source
|
|
assert "/verifyOnlineAgentTask" in source
|
|
assert "/cancelOnlineAgentTask" in source
|
|
assert "agent-task:control" in source
|
|
|
|
|
|
def test_online_task_service_keeps_control_endpoints_off_public_nginx():
|
|
nginx = (ROOT / "deploy/nginx/aios-agent-task-api.conf.example").read_text(encoding="utf-8")
|
|
assert "/agent-tasks/v1/tasks/poll" in nginx
|
|
assert "(claim|heartbeat|result)" in nginx
|
|
assert "/v1/agents/register" not in nginx.replace(
|
|
"# Control endpoints (/v1/agents/register, POST /v1/tasks, verify/cancel/events)", ""
|
|
)
|
|
|
|
|
|
def test_online_task_systemd_uses_runtime_credential_file():
|
|
unit = (ROOT / "deploy/systemd/aios-agent-task-api.service").read_text(encoding="utf-8")
|
|
assert "LoadCredential=control_token:" in unit
|
|
assert "/run/credentials/aios-agent-task-api.service/control_token" in unit
|
|
assert "ReadWritePaths=/var/lib/aios-agent-task" in unit
|
|
assert "127.0.0.1" in unit
|
|
|
|
|
|
def test_control_plane_snapshot_is_local_aggregate_not_missing_bridge_route():
|
|
source = (ROOT / "active-runtime/app/aios_model_api_staging.py").read_text(encoding="utf-8")
|
|
assert "def control_plane_snapshot(self):" in source
|
|
assert "'snapshot_source': 'model-api-local-aggregate'" in source
|
|
assert "return self.control_plane_snapshot()" in source
|
|
assert "bridge_v2_json('/v2/control-plane-snapshot'" not in source
|