213 lines
6.7 KiB
Python
213 lines
6.7 KiB
Python
"""Comprehensive test suite for Google ADK Multi-Agent Architecture & PostgreSQL Persistence."""
|
|
|
|
import pytest
|
|
import uuid
|
|
from starlette.testclient import TestClient
|
|
|
|
from app.adk.agents import (
|
|
DiscoveryAgent,
|
|
ArchitectureDesignAgent,
|
|
ValidationReviewAgent,
|
|
PackagingAgent,
|
|
OrchestratorLoopAgent,
|
|
build_adk_multi_agent_system,
|
|
)
|
|
from app.adk.artifacts import PostgresArtifactRepository
|
|
from app.adk.evaluation import ADKEvaluator
|
|
from app.adk.runners import ADKAgentRunner
|
|
from app.adk.sessions import PostgresSessionService
|
|
from app.adk.tools import ADK_TOOLS
|
|
from app.adk.workflows import create_gcp_adk_workflow
|
|
from app.agent import app
|
|
from app.config import get_settings
|
|
from app.database import get_db_manager
|
|
from app.skills.loader import SkillLoader
|
|
|
|
|
|
@pytest.fixture
|
|
def skill_loader():
|
|
settings = get_settings()
|
|
loader = SkillLoader(settings.SKILLS_DIR)
|
|
loader.load_skills()
|
|
return loader
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return TestClient(app)
|
|
|
|
|
|
def test_database_manager_operations():
|
|
"""Verify PostgreSQL database manager schema initialization & CRUD operations."""
|
|
db = get_db_manager()
|
|
session_id = f"test-sess-{uuid.uuid4()}"
|
|
execution_id = f"test-exec-{uuid.uuid4()}"
|
|
|
|
# 1. Save and retrieve session
|
|
saved = db.save_session(session_id, "TestAgent", {"key": "value"})
|
|
assert saved is True
|
|
|
|
session_data = db.get_session(session_id)
|
|
assert session_data is not None
|
|
assert session_data["agent_name"] == "TestAgent"
|
|
assert session_data["state_data"]["key"] == "value"
|
|
|
|
# 2. Save workflow execution
|
|
saved_exec = db.save_workflow_execution(
|
|
execution_id=execution_id,
|
|
session_id=session_id,
|
|
status="completed",
|
|
current_phase="package",
|
|
loop_count=1,
|
|
request_summary="Unit test workflow request",
|
|
results={"passed": True},
|
|
)
|
|
assert saved_exec is True
|
|
|
|
# 3. Save artifact
|
|
art_saved = db.save_artifact(
|
|
artifact_id=f"art-{uuid.uuid4()}",
|
|
execution_id=execution_id,
|
|
artifact_type="markdown",
|
|
file_path="docs/test.md",
|
|
content="# Test Content",
|
|
)
|
|
assert art_saved is True
|
|
|
|
# 4. Record review log
|
|
log_saved = db.record_orchestrator_log(
|
|
log_id=f"log-{uuid.uuid4()}",
|
|
execution_id=execution_id,
|
|
iteration=1,
|
|
review_status="APPROVED",
|
|
reviewer_agent="ValidationReviewAgent",
|
|
feedback="Passed all checks",
|
|
)
|
|
assert log_saved is True
|
|
|
|
|
|
def test_adk_tools():
|
|
"""Verify google.adk.tools wrappers."""
|
|
assert len(ADK_TOOLS) == 6
|
|
tool_names = [t.name for t in ADK_TOOLS]
|
|
assert "validate_mermaid_diagram" in tool_names
|
|
assert "validate_terraform_syntax" in tool_names
|
|
assert "validate_repository_artifacts" in tool_names
|
|
assert "developerknowledge_search_documents" in tool_names
|
|
assert "developerknowledge_get_documents" in tool_names
|
|
assert "developerknowledge_answer_query" in tool_names
|
|
|
|
|
|
def test_adk_postgres_session_service():
|
|
"""Verify PostgresSessionService creating, getting, and saving session state."""
|
|
session_service = PostgresSessionService()
|
|
session_id = f"adk-session-{uuid.uuid4()}"
|
|
|
|
sess = session_service.create_session("GCP_ADK_Agent", session_id=session_id)
|
|
assert sess.session_id == session_id
|
|
assert sess.agent_name == "GCP_ADK_Agent"
|
|
|
|
sess.state["counter"] = 42
|
|
saved = session_service.save_session(sess)
|
|
assert saved is True
|
|
|
|
fetched = session_service.get_session(session_id)
|
|
assert fetched is not None
|
|
assert fetched.state.get("counter") == 42
|
|
|
|
|
|
def test_adk_orchestrator_loop_agent(skill_loader):
|
|
"""Verify OrchestratorLoopAgent multi-agent loop review cycle."""
|
|
orchestrator = build_adk_multi_agent_system(skill_loader, max_iterations=3)
|
|
assert orchestrator.name == "OrchestratorLoopAgent"
|
|
assert orchestrator.max_iterations == 3
|
|
|
|
initial_state = {
|
|
"execution_id": str(uuid.uuid4()),
|
|
"workflow_request": "Event-driven regional HTTP application",
|
|
"target_dir": ".",
|
|
"active_skills": [],
|
|
}
|
|
|
|
final_state = orchestrator.execute(initial_state)
|
|
assert final_state.get("validation_passed") is True
|
|
assert final_state.get("loop_completed_successfully") is True
|
|
assert len(final_state.get("active_skills", [])) == 5
|
|
|
|
|
|
def test_adk_workflow_composition(skill_loader):
|
|
"""Verify google.adk.workflows Workflow composition."""
|
|
workflow = create_gcp_adk_workflow(skill_loader, max_iterations=2)
|
|
assert workflow.name == "GCP_Solution_Architecture_Workflow"
|
|
assert len(workflow.steps) == 1
|
|
|
|
initial_state = {
|
|
"execution_id": str(uuid.uuid4()),
|
|
"workflow_request": "Event-driven architecture test",
|
|
"target_dir": ".",
|
|
"active_skills": [],
|
|
}
|
|
|
|
result = workflow.execute(initial_state)
|
|
assert result.get("validation_passed") is True
|
|
|
|
|
|
def test_adk_agent_runner():
|
|
"""Verify ADKAgentRunner execution and artifact persistence."""
|
|
runner = ADKAgentRunner()
|
|
session_id = f"runner-session-{uuid.uuid4()}"
|
|
|
|
res = runner.run_execution(
|
|
session_id=session_id,
|
|
request_summary="High scale ingestion workflow",
|
|
target_dir=".",
|
|
)
|
|
|
|
assert res["status"] == "success"
|
|
assert res["validation_passed"] is True
|
|
assert "solution_guide" in res["artifacts"]
|
|
|
|
# Verify session persists in Postgres DB
|
|
session_service = PostgresSessionService()
|
|
sess = session_service.get_session(session_id)
|
|
assert sess is not None
|
|
|
|
|
|
def test_adk_evaluator():
|
|
"""Verify ADKEvaluator executing benchmark cases."""
|
|
evaluator = ADKEvaluator()
|
|
case = {
|
|
"id": "case-test",
|
|
"workflow_request": "Event-driven regional HTTP application",
|
|
"rubric": {
|
|
"required_products": ["Cloud Run", "Pub/Sub", "Cloud Storage"],
|
|
"requires_mermaid": True,
|
|
"requires_terraform": True,
|
|
"requires_guide_sections": ["Functional requirements", "Selected products"],
|
|
},
|
|
}
|
|
|
|
res = evaluator.evaluate_benchmark_case(case)
|
|
assert res["passed"] is True
|
|
assert res["percentage"] >= 80.0
|
|
|
|
|
|
def test_starlette_session_api_endpoint(client):
|
|
"""Verify Starlette API GET /sessions/{session_id} route."""
|
|
session_id = f"api-sess-{uuid.uuid4()}"
|
|
|
|
# First run workflow to populate session
|
|
payload = {
|
|
"session_id": session_id,
|
|
"request": "Test session retrieval route",
|
|
}
|
|
gen_resp = client.post("/generate", json=payload)
|
|
assert gen_resp.status_code == 200
|
|
|
|
# Query session
|
|
sess_resp = client.get(f"/sessions/{session_id}")
|
|
assert sess_resp.status_code == 200
|
|
sess_data = sess_resp.json()
|
|
assert sess_data["session_id"] == session_id
|
|
assert sess_data["agent_name"] == "OrchestratorLoopAgent"
|