b4ae08ef53
Preflight and updater tests no longer probe 8081/8082 or GitHub. Hermetic runs fail-fast on non-loopback sockets. GUI helpers are importable without customtkinter. CI installs the web extra, and pytest-timeout plus a wall-clock wrapper bound the suite.
169 lines
7.4 KiB
Python
169 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from antigravity_provider.router.router_config import (
|
|
RolePolicy,
|
|
RouterConfig,
|
|
RouterProfileConfig,
|
|
save_router_config,
|
|
)
|
|
from antigravity_provider.router.workflow_service import WorkflowService
|
|
|
|
|
|
@pytest.fixture
|
|
def workflow_service(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
config = RouterConfig(
|
|
profiles={
|
|
"account-a": RouterProfileConfig(
|
|
profile_id="account-a",
|
|
provider="openai-codex",
|
|
preferred_models=["model-real-from-config"],
|
|
)
|
|
},
|
|
roles={"developer": RolePolicy(role_name="developer", preferred_chain=["account-a"])},
|
|
)
|
|
save_router_config(config)
|
|
return WorkflowService(tmp_path / "workflow_state.json")
|
|
|
|
|
|
def test_router_roles_migrate_to_agents_and_create_real_files(workflow_service, tmp_path):
|
|
snapshot = workflow_service.snapshot()
|
|
assert "developer" in [agent["id"] for agent in snapshot["agents"]]
|
|
agent = next(item for item in snapshot["agents"] if item["id"] == "developer")
|
|
assert agent["agent_file"] == "agents/developer.md"
|
|
assert agent["agent_file_exists"] is True
|
|
assert (tmp_path / "agents" / "developer.md").is_file()
|
|
assert agent["execution_config"]["account"] == "account-a"
|
|
assert agent["execution_config"]["model"] == "model-real-from-config"
|
|
|
|
|
|
def test_create_update_file_and_restart_persistence(workflow_service, tmp_path):
|
|
created = workflow_service.create_agent({
|
|
"name": "Security Reviewer",
|
|
"role": "security-reviewer",
|
|
"account": "account-a",
|
|
"model": "model-real-from-config",
|
|
"description": "Проверяет безопасность",
|
|
})
|
|
workflow_service.save_agent_file(created.id, "# Security\n\nOnly measured facts.")
|
|
workflow_service.update_agent(created.id, {"timeout": 91, "position": {"x": 33, "y": 44}})
|
|
|
|
restarted = WorkflowService(tmp_path / "workflow_state.json")
|
|
item = next(agent for agent in restarted.snapshot()["agents"] if agent["id"] == created.id)
|
|
assert item["timeout"] == 91
|
|
assert item["position"] == {"x": 33, "y": 44}
|
|
assert restarted.read_agent_file(created.id)["content"].endswith("Only measured facts.")
|
|
|
|
|
|
def test_delete_requires_explicit_confirmation_when_referenced(workflow_service):
|
|
workflow_service.create_agent({"name": "Test Reviewer", "role": "test-reviewer", "account": "account-a"})
|
|
workflow_service.save_workflow({
|
|
"start_agent_id": "developer",
|
|
"max_iterations": 3,
|
|
"edges": [{"id": "review", "source": "developer", "target": "test-reviewer", "condition": "SUCCESS"}],
|
|
})
|
|
warning = workflow_service.delete_agent("test-reviewer")
|
|
assert warning["confirmation_required"] is True
|
|
assert warning["consequences"]["workflow_edges"] == ["review"]
|
|
result = workflow_service.delete_agent("test-reviewer", force=True)
|
|
assert result["deleted"] is True
|
|
assert workflow_service.workflow.edges == []
|
|
|
|
|
|
def test_cycles_are_valid_and_iteration_limit_is_persisted(workflow_service):
|
|
workflow_service.create_agent({"name": "Test Reviewer", "role": "test-reviewer", "account": "account-a"})
|
|
definition = workflow_service.save_workflow({
|
|
"start_agent_id": "developer",
|
|
"max_iterations": 2,
|
|
"edges": [
|
|
{"source": "developer", "target": "test-reviewer", "condition": "SUCCESS"},
|
|
{"source": "test-reviewer", "target": "developer", "condition": "REVIEW_FAILED"},
|
|
],
|
|
})
|
|
assert definition.max_iterations == 2
|
|
assert definition.edges[1].condition == "REVIEW_FAILED"
|
|
payload = json.loads(workflow_service.state_path.read_text(encoding="utf-8"))
|
|
assert payload["workflow"]["max_iterations"] == 2
|
|
|
|
|
|
def test_invalid_edge_and_unknown_model_are_rejected(workflow_service):
|
|
with pytest.raises(ValueError, match="отсутствующего агента"):
|
|
workflow_service.save_workflow({"edges": [{"source": "developer", "target": "missing"}]})
|
|
with pytest.raises(ValueError, match="не доступна"):
|
|
workflow_service.update_agent("developer", {"account": "account-a", "model": "invented-model"})
|
|
|
|
|
|
def test_interrupted_run_is_reported_not_silently_completed(workflow_service, tmp_path):
|
|
workflow_service.run.update({"id": "run-1", "status": "running", "current_agent_id": "developer"})
|
|
workflow_service._save()
|
|
restarted = WorkflowService(tmp_path / "workflow_state.json")
|
|
assert restarted.run["status"] == "interrupted"
|
|
assert "перезапуском" in restarted.run["error"]
|
|
assert restarted.events[-1].type == "WORKFLOW_INTERRUPTED"
|
|
|
|
|
|
def test_live_cycle_stops_with_explicit_iteration_limit_event(workflow_service, monkeypatch):
|
|
workflow_service.create_agent({"name": "Loop Reviewer", "role": "loop-reviewer", "account": "account-a"})
|
|
workflow_service.save_workflow({
|
|
"start_agent_id": "developer",
|
|
"max_iterations": 2,
|
|
"edges": [
|
|
{"source": "developer", "target": "loop-reviewer", "condition": "SUCCESS"},
|
|
{"source": "loop-reviewer", "target": "developer", "condition": "REVIEW_FAILED"},
|
|
],
|
|
})
|
|
|
|
class FakeEngine:
|
|
def reload_config(self):
|
|
return None
|
|
|
|
def route_request(self, request, role=None, session_id=None):
|
|
status = "REVIEW_FAILED" if role == "loop-reviewer" else "SUCCESS"
|
|
return {
|
|
"choices": [{"message": {"content": status}}],
|
|
"router_metadata": {
|
|
"provider": "measured-provider",
|
|
"profile_id": "account-a",
|
|
"selection_trace": {"selected_model": "model-real-from-config"},
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr("antigravity_provider.router.router_engine.get_router_engine", lambda: FakeEngine())
|
|
workflow_service.start("Проверить реальный цикл")
|
|
thread = workflow_service._thread
|
|
assert thread is not None
|
|
thread.join(timeout=3)
|
|
assert not thread.is_alive(), "Workflow execution thread leaked after join"
|
|
|
|
assert workflow_service.run["status"] == "failed"
|
|
assert workflow_service.run["error"] == "Достигнут предел итераций: 2"
|
|
assert any(event.type == "WORKFLOW_MAX_ITERATIONS" for event in workflow_service.events)
|
|
|
|
|
|
def test_provider_error_text_reaches_run_and_events(workflow_service, monkeypatch):
|
|
provider_text = "Provider Error: authentication token missing for account-a"
|
|
|
|
class ErrorEngine:
|
|
def reload_config(self):
|
|
return None
|
|
|
|
def route_request(self, request, role=None, session_id=None):
|
|
return {"choices": [{"message": {"content": f"ERROR\n{provider_text}"}}]}
|
|
|
|
monkeypatch.setattr("antigravity_provider.router.router_engine.get_router_engine", lambda: ErrorEngine())
|
|
workflow_service.workflow.start_agent_id = "developer"
|
|
workflow_service.workflow.edges = []
|
|
workflow_service.start("Проверить ошибку")
|
|
thread = workflow_service._thread
|
|
assert thread is not None
|
|
thread.join(timeout=3)
|
|
assert not thread.is_alive(), "Workflow execution thread leaked after join"
|
|
|
|
assert workflow_service.run["status"] == "failed"
|
|
assert provider_text in workflow_service.run["error"]
|
|
assert any(provider_text in (event.error or "") for event in workflow_service.events)
|
|
assert workflow_service.snapshot()["agents"][0]["runtime_state"] == "error"
|