File size: 3,551 Bytes
a3ecd30 9237011 a3ecd30 3e631d0 0e9cb33 a3ecd30 3e631d0 de6f25f 0e9cb33 9237011 a3ecd30 9237011 3e631d0 de6f25f 0e9cb33 9237011 | 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 | from pathlib import Path
import pytest
from app.agents.graph import AuditGraph
from app.config import Settings
from app.schemas import AuditReport
def test_audit_graph_exposes_current_agents_through_registry():
graph = AuditGraph(Settings())
assert [spec.node_name for spec in graph.analysis_agents] == [
"security",
"performance",
"quality",
"docs",
"config",
"dependency",
"error_handling",
"observability",
"cuda_migration",
]
assert [spec.state_key for spec in graph.analysis_agents] == [
"security_output",
"performance_output",
"quality_output",
"docs_output",
"config_output",
"dependency_output",
"error_handling_output",
"observability_output",
"cuda_migration_output",
]
assert [spec.agent.name for spec in graph.analysis_agents] == [
"Security Agent",
"Performance Agent",
"Quality Agent",
"Docs Agent",
"Config Agent",
"Dependency Agent",
"Error Handling Agent",
"Observability Agent",
"CUDA-to-ROCm Agent",
]
@pytest.mark.anyio
async def test_run_with_progress_yields_real_stages_and_report(tmp_path: Path):
source = tmp_path / "app.py"
source.write_text("API_KEY = '1234567890abcdef'\nresponse = requests.get(url)\n", encoding="utf-8")
(tmp_path / "README.md").write_text(
"# Demo\n\n## Quick Start\nInstall and run it.\n## Tests\nRun pytest.\n## Configuration\nCopy .env.example.",
encoding="utf-8",
)
graph = AuditGraph(Settings(max_files=10, max_file_size_kb=10, max_chars_per_chunk=1000))
graph.crawler.clone_and_scan = lambda repo_url: graph.crawler.scan_local_repo(repo_url, tmp_path)
graph.crawler.cleanup = lambda scan_result: None
events = []
async for event in graph.run_with_progress("https://github.com/example/project"):
events.append(event)
assert any("Crawler Agent" in event for event in events if isinstance(event, str))
assert any("Security Agent" in event for event in events if isinstance(event, str))
assert any("Performance Agent" in event for event in events if isinstance(event, str))
assert any("Quality Agent" in event for event in events if isinstance(event, str))
assert any("Docs Agent" in event for event in events if isinstance(event, str))
assert any("Config Agent" in event for event in events if isinstance(event, str))
assert any("Dependency Agent" in event for event in events if isinstance(event, str))
assert any("Error Handling Agent" in event for event in events if isinstance(event, str))
assert any("Observability Agent" in event for event in events if isinstance(event, str))
assert any("CUDA-to-ROCm Agent" in event for event in events if isinstance(event, str))
assert isinstance(events[-1], AuditReport)
assert len(events[-1].findings) >= 2
assert any(finding.agent_source == "Error Handling Agent" for finding in events[-1].findings)
assert "Security Agent" in events[-1].agents_run
assert "Performance Agent" in events[-1].agents_run
assert "Quality Agent" in events[-1].agents_run
assert "Docs Agent" in events[-1].agents_run
assert "Config Agent" in events[-1].agents_run
assert "Dependency Agent" in events[-1].agents_run
assert "Error Handling Agent" in events[-1].agents_run
assert "Observability Agent" in events[-1].agents_run
assert "CUDA-to-ROCm Agent" in events[-1].agents_run
|