"""
Tests for Code Audit Governance system in amicus-mcp.
These tests validate the code audit workflow including:
- Submitting code audit requests
- Getting pending audits
- Approving code audits (with self-approval prevention)
- Rejecting code audits (with required feedback)
- Getting audit status
- Listing code audits with filters
"""
import pytest
import time
from pathlib import Path
from unittest.mock import patch
from amicus.server import (
submit_code_audit_request as _submit_code_audit_request,
get_pending_audits as _get_pending_audits,
approve_code_audit as _approve_code_audit,
reject_code_audit as _reject_code_audit,
get_audit_status as _get_audit_status,
list_code_audits as _list_code_audits
)
from amicus.core import read_with_lock, write_with_lock, get_state_file
# Access the underlying functions from FunctionTool wrappers
submit_code_audit_request = _submit_code_audit_request.fn
get_pending_audits = _get_pending_audits.fn
approve_code_audit = _approve_code_audit.fn
reject_code_audit = _reject_code_audit.fn
get_audit_status = _get_audit_status.fn
list_code_audits = _list_code_audits.fn
class TestCodeAuditSubmission:
"""Test suite for submitting code audit requests."""
def test_submit_code_audit_request_creates_audit_with_correct_structure(self, temp_context_dir):
"""Test that submit_code_audit_request creates an audit with the correct structure."""
# Setup: Initialize state
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test: Submit a code audit request
result = submit_code_audit_request(
requesting_agent="Node-Dev-001",
files=["src/module.py", "tests/test_module.py"],
description="Add new feature to module",
proposed_changes="Add function foo() that returns bar",
priority="high"
)
# Verify: Result contains audit ID and status
assert "Code audit request submitted:" in result
assert "audit-" in result
assert "PENDING" in result
assert "src/module.py" in result
assert "HIGH" in result.upper()
# Verify: State was updated with correct structure
state_data = read_with_lock(state_file)
assert "code_audits" in state_data
audits = state_data["code_audits"]
assert len(audits) == 1
# Get the audit ID from result
audit_id = result.split("Code audit request submitted: ")[1].split("\n")[0]
# Verify: Audit has correct structure
audit = audits[audit_id]
assert audit["id"] == audit_id
assert audit["requesting_agent"] == "Node-Dev-001"
assert audit["reviewer_agent"] is None
assert audit["files"] == ["src/module.py", "tests/test_module.py"]
assert audit["description"] == "Add new feature to module"
assert audit["proposed_changes"] == "Add function foo() that returns bar"
assert audit["priority"] == "high"
assert audit["status"] == "pending"
assert audit["feedback"] is None
assert "created_at" in audit
assert audit["reviewed_at"] is None
def test_submit_code_audit_request_validates_file_paths(self, temp_context_dir):
"""Test that submit_code_audit_request validates file paths are within project root."""
# Setup: Initialize state
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test: Submit audit with unsafe path
result = submit_code_audit_request(
requesting_agent="Node-Dev-001",
files=["/etc/passwd"],
description="Malicious change",
proposed_changes="Modify system file",
priority="high"
)
# Verify: Request is rejected
assert "ERROR" in result
assert "outside the project root" in result
assert "Audit rejected" in result
# Verify: No audit was created
state_data = read_with_lock(state_file)
audits = state_data.get("code_audits", {})
assert len(audits) == 0
def test_submit_multiple_code_audits(self, temp_context_dir):
"""Test that multiple audits can be submitted and tracked separately."""
# Setup: Initialize state
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test: Submit multiple audits
result1 = submit_code_audit_request(
requesting_agent="Node-Dev-001",
files=["file1.py"],
description="Change 1",
proposed_changes="Changes to file1",
priority="high"
)
result2 = submit_code_audit_request(
requesting_agent="Node-Dev-002",
files=["file2.py"],
description="Change 2",
proposed_changes="Changes to file2",
priority="medium"
)
# Verify: Both audits were created
state_data = read_with_lock(state_file)
audits = state_data["code_audits"]
assert len(audits) == 2
# Extract audit IDs
audit_id1 = result1.split("Code audit request submitted: ")[1].split("\n")[0]
audit_id2 = result2.split("Code audit request submitted: ")[1].split("\n")[0]
assert audit_id1 != audit_id2
assert audits[audit_id1]["requesting_agent"] == "Node-Dev-001"
assert audits[audit_id2]["requesting_agent"] == "Node-Dev-002"
class TestGetPendingAudits:
"""Test suite for retrieving pending code audits."""
def test_get_pending_audits_returns_pending_audits(self, temp_context_dir):
"""Test that get_pending_audits returns only pending audits."""
# Setup: Create state with mixed audit statuses
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"files": ["file1.py"],
"description": "Pending change 1",
"proposed_changes": "Change 1",
"priority": "high",
"status": "pending",
"created_at": now - 300
},
"audit-002": {
"id": "audit-002",
"requesting_agent": "Node-Dev-002",
"files": ["file2.py"],
"description": "Approved change",
"proposed_changes": "Change 2",
"priority": "medium",
"status": "approved",
"created_at": now - 200,
"reviewed_at": now - 100,
"reviewer_agent": "Node-Reviewer-001"
},
"audit-003": {
"id": "audit-003",
"requesting_agent": "Node-Dev-003",
"files": ["file3.py"],
"description": "Pending change 2",
"proposed_changes": "Change 3",
"priority": "medium",
"status": "pending",
"created_at": now - 100
}
}
}
write_with_lock(state_file, initial_state)
# Test: Get pending audits
result = get_pending_audits()
# Verify: Only pending audits are returned
assert "Pending Code Audits" in result
assert "audit-001" in result
assert "audit-003" in result
assert "audit-002" not in result # Should not show approved audit
assert "Node-Dev-001" in result
assert "Node-Dev-003" in result
assert "Total pending: 2" in result
def test_get_pending_audits_empty(self, temp_context_dir):
"""Test that get_pending_audits handles empty state gracefully."""
# Setup: Initialize state with no audits
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test: Get pending audits
result = get_pending_audits()
# Verify: Returns appropriate message
assert "No pending code audits" in result
def test_get_pending_audits_sorts_by_priority_and_age(self, temp_context_dir):
"""Test that get_pending_audits sorts by priority (high first) then age (oldest first)."""
# Setup: Create state with audits of different priorities and ages
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-low-old": {
"id": "audit-low-old",
"requesting_agent": "Node-Dev-001",
"files": ["file1.py"],
"description": "Low priority old",
"proposed_changes": "Change 1",
"priority": "low",
"status": "pending",
"created_at": now - 400
},
"audit-high-new": {
"id": "audit-high-new",
"requesting_agent": "Node-Dev-002",
"files": ["file2.py"],
"description": "High priority new",
"proposed_changes": "Change 2",
"priority": "high",
"status": "pending",
"created_at": now - 100
},
"audit-high-old": {
"id": "audit-high-old",
"requesting_agent": "Node-Dev-003",
"files": ["file3.py"],
"description": "High priority old",
"proposed_changes": "Change 3",
"priority": "high",
"status": "pending",
"created_at": now - 300
},
"audit-medium": {
"id": "audit-medium",
"requesting_agent": "Node-Dev-004",
"files": ["file4.py"],
"description": "Medium priority",
"proposed_changes": "Change 4",
"priority": "medium",
"status": "pending",
"created_at": now - 200
}
}
}
write_with_lock(state_file, initial_state)
# Test: Get pending audits
result = get_pending_audits()
# Verify: Audits are sorted correctly (high priority first, then by age)
lines = result.split("\n")
audit_lines = [line for line in lines if "Audit ID:" in line]
# Extract audit IDs in order
audit_order = []
for line in audit_lines:
audit_id = line.split("Audit ID: ")[1].strip()
audit_order.append(audit_id)
# Verify order: high-old, high-new, medium, low-old
assert audit_order[0] == "audit-high-old" # High priority, oldest
assert audit_order[1] == "audit-high-new" # High priority, newer
assert audit_order[2] == "audit-medium" # Medium priority
assert audit_order[3] == "audit-low-old" # Low priority
class TestApproveCodeAudit:
"""Test suite for approving code audits."""
def test_approve_code_audit_success(self, temp_context_dir):
"""Test that approve_code_audit successfully approves a pending audit."""
# Setup: Create state with pending audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": None,
"files": ["file1.py"],
"description": "Test change",
"proposed_changes": "Add feature",
"priority": "high",
"status": "pending",
"feedback": None,
"created_at": now,
"reviewed_at": None
}
}
}
write_with_lock(state_file, initial_state)
# Test: Approve the audit
result = approve_code_audit(
audit_id="audit-001",
reviewer_agent="Node-Reviewer-001",
feedback="Looks good, proceed with changes"
)
# Verify: Success message
assert "APPROVED" in result
assert "audit-001" in result
assert "Node-Reviewer-001" in result
assert "may proceed" in result
assert "file1.py" in result
assert "Looks good, proceed with changes" in result
# Verify: State was updated correctly
state_data = read_with_lock(state_file)
audit = state_data["code_audits"]["audit-001"]
assert audit["status"] == "approved"
assert audit["reviewer_agent"] == "Node-Reviewer-001"
assert audit["feedback"] == "Looks good, proceed with changes"
assert audit["reviewed_at"] is not None
assert audit["reviewed_at"] > now
def test_approve_code_audit_cannot_approve_own_audit(self, temp_context_dir):
"""Test that an agent cannot approve their own audit request."""
# Setup: Create state with pending audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": None,
"files": ["file1.py"],
"description": "Test change",
"proposed_changes": "Add feature",
"priority": "high",
"status": "pending",
"feedback": None,
"created_at": now,
"reviewed_at": None
}
}
}
write_with_lock(state_file, initial_state)
# Test: Try to approve own audit
result = approve_code_audit(
audit_id="audit-001",
reviewer_agent="Node-Dev-001", # Same as requesting agent
feedback="Approving my own work"
)
# Verify: Error message
assert "ERROR" in result
assert "Cannot approve your own audit request" in result
# Verify: Audit remains pending
state_data = read_with_lock(state_file)
audit = state_data["code_audits"]["audit-001"]
assert audit["status"] == "pending"
assert audit["reviewer_agent"] is None
def test_approve_code_audit_not_found(self, temp_context_dir):
"""Test that approve_code_audit handles non-existent audit ID."""
# Setup: Initialize state with no audits
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time(),
"code_audits": {}
}
write_with_lock(state_file, initial_state)
# Test: Try to approve non-existent audit
result = approve_code_audit(
audit_id="audit-nonexistent",
reviewer_agent="Node-Reviewer-001"
)
# Verify: Error message
assert "ERROR" in result
assert "not found" in result
def test_approve_code_audit_already_approved(self, temp_context_dir):
"""Test that approve_code_audit handles already approved audit."""
# Setup: Create state with already approved audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": "Node-Reviewer-001",
"files": ["file1.py"],
"description": "Test change",
"proposed_changes": "Add feature",
"priority": "high",
"status": "approved",
"feedback": "Already approved",
"created_at": now - 100,
"reviewed_at": now
}
}
}
write_with_lock(state_file, initial_state)
# Test: Try to approve again
result = approve_code_audit(
audit_id="audit-001",
reviewer_agent="Node-Reviewer-002"
)
# Verify: Error message
assert "ERROR" in result
assert "not pending" in result
assert "approved" in result
def test_approve_code_audit_without_feedback(self, temp_context_dir):
"""Test that approve_code_audit works without optional feedback."""
# Setup: Create state with pending audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": None,
"files": ["file1.py"],
"description": "Test change",
"proposed_changes": "Add feature",
"priority": "high",
"status": "pending",
"feedback": None,
"created_at": now,
"reviewed_at": None
}
}
}
write_with_lock(state_file, initial_state)
# Test: Approve without feedback
result = approve_code_audit(
audit_id="audit-001",
reviewer_agent="Node-Reviewer-001"
)
# Verify: Success without feedback in message
assert "APPROVED" in result
assert "Feedback:" not in result
# Verify: State was updated correctly
state_data = read_with_lock(state_file)
audit = state_data["code_audits"]["audit-001"]
assert audit["status"] == "approved"
class TestRejectCodeAudit:
"""Test suite for rejecting code audits."""
def test_reject_code_audit_success(self, temp_context_dir):
"""Test that reject_code_audit successfully rejects a pending audit."""
# Setup: Create state with pending audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": None,
"files": ["file1.py"],
"description": "Test change",
"proposed_changes": "Add feature",
"priority": "high",
"status": "pending",
"feedback": None,
"created_at": now,
"reviewed_at": None
}
}
}
write_with_lock(state_file, initial_state)
# Test: Reject the audit with feedback
result = reject_code_audit(
audit_id="audit-001",
reviewer_agent="Node-Reviewer-001",
feedback="Code needs better error handling and unit tests"
)
# Verify: Success message
assert "REJECTED" in result
assert "audit-001" in result
assert "Node-Reviewer-001" in result
assert "must address feedback" in result
assert "Code needs better error handling" in result
# Verify: State was updated correctly
state_data = read_with_lock(state_file)
audit = state_data["code_audits"]["audit-001"]
assert audit["status"] == "rejected"
assert audit["reviewer_agent"] == "Node-Reviewer-001"
assert audit["feedback"] == "Code needs better error handling and unit tests"
assert audit["reviewed_at"] is not None
assert audit["reviewed_at"] > now
def test_reject_code_audit_requires_feedback(self, temp_context_dir):
"""Test that reject_code_audit requires feedback to be provided."""
# Setup: Create state with pending audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": None,
"files": ["file1.py"],
"description": "Test change",
"proposed_changes": "Add feature",
"priority": "high",
"status": "pending",
"feedback": None,
"created_at": now,
"reviewed_at": None
}
}
}
write_with_lock(state_file, initial_state)
# Test: Try to reject without feedback
result = reject_code_audit(
audit_id="audit-001",
reviewer_agent="Node-Reviewer-001",
feedback="" # Empty feedback
)
# Verify: Error message
assert "ERROR" in result
assert "Feedback is required" in result
# Verify: Audit remains pending
state_data = read_with_lock(state_file)
audit = state_data["code_audits"]["audit-001"]
assert audit["status"] == "pending"
assert audit["reviewer_agent"] is None
def test_reject_code_audit_whitespace_only_feedback(self, temp_context_dir):
"""Test that reject_code_audit rejects whitespace-only feedback."""
# Setup: Create state with pending audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": None,
"files": ["file1.py"],
"description": "Test change",
"proposed_changes": "Add feature",
"priority": "high",
"status": "pending",
"feedback": None,
"created_at": now,
"reviewed_at": None
}
}
}
write_with_lock(state_file, initial_state)
# Test: Try to reject with whitespace-only feedback
result = reject_code_audit(
audit_id="audit-001",
reviewer_agent="Node-Reviewer-001",
feedback=" \n\t " # Only whitespace
)
# Verify: Error message
assert "ERROR" in result
assert "Feedback is required" in result
# Verify: Audit remains pending
state_data = read_with_lock(state_file)
audit = state_data["code_audits"]["audit-001"]
assert audit["status"] == "pending"
def test_reject_code_audit_cannot_reject_own_audit(self, temp_context_dir):
"""Test that an agent cannot reject their own audit request."""
# Setup: Create state with pending audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": None,
"files": ["file1.py"],
"description": "Test change",
"proposed_changes": "Add feature",
"priority": "high",
"status": "pending",
"feedback": None,
"created_at": now,
"reviewed_at": None
}
}
}
write_with_lock(state_file, initial_state)
# Test: Try to reject own audit
result = reject_code_audit(
audit_id="audit-001",
reviewer_agent="Node-Dev-001", # Same as requesting agent
feedback="Rejecting my own work"
)
# Verify: Error message
assert "ERROR" in result
assert "Cannot reject your own audit request" in result
# Verify: Audit remains pending
state_data = read_with_lock(state_file)
audit = state_data["code_audits"]["audit-001"]
assert audit["status"] == "pending"
assert audit["reviewer_agent"] is None
class TestGetAuditStatus:
"""Test suite for getting audit status."""
def test_get_audit_status_pending(self, temp_context_dir):
"""Test that get_audit_status returns correct status for pending audit."""
# Setup: Create state with pending audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": None,
"files": ["file1.py", "file2.py"],
"description": "Add new feature",
"proposed_changes": "Implement feature X with Y",
"priority": "high",
"status": "pending",
"feedback": None,
"created_at": now - 300,
"reviewed_at": None
}
}
}
write_with_lock(state_file, initial_state)
# Test: Get audit status
result = get_audit_status("audit-001")
# Verify: Status is displayed correctly
assert "audit-001" in result
assert "PENDING" in result
assert "Node-Dev-001" in result
assert "file1.py, file2.py" in result
assert "Add new feature" in result
assert "HIGH" in result
assert "Awaiting review" in result
assert "Implement feature X with Y" in result
def test_get_audit_status_approved(self, temp_context_dir):
"""Test that get_audit_status returns correct status for approved audit."""
# Setup: Create state with approved audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": "Node-Reviewer-001",
"files": ["file1.py"],
"description": "Add new feature",
"proposed_changes": "Implementation details",
"priority": "high",
"status": "approved",
"feedback": "Looks good, proceed",
"created_at": now - 300,
"reviewed_at": now - 100
}
}
}
write_with_lock(state_file, initial_state)
# Test: Get audit status
result = get_audit_status("audit-001")
# Verify: Status is displayed correctly
assert "APPROVED" in result
assert "Node-Reviewer-001" in result
assert "Looks good, proceed" in result
assert "may proceed with the file changes" in result
def test_get_audit_status_rejected(self, temp_context_dir):
"""Test that get_audit_status returns correct status for rejected audit."""
# Setup: Create state with rejected audit
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"reviewer_agent": "Node-Reviewer-001",
"files": ["file1.py"],
"description": "Add new feature",
"proposed_changes": "Implementation details",
"priority": "high",
"status": "rejected",
"feedback": "Needs better error handling and tests",
"created_at": now - 300,
"reviewed_at": now - 100
}
}
}
write_with_lock(state_file, initial_state)
# Test: Get audit status
result = get_audit_status("audit-001")
# Verify: Status is displayed correctly
assert "REJECTED" in result
assert "Node-Reviewer-001" in result
assert "Needs better error handling and tests" in result
assert "Address the feedback and submit a new audit request" in result
def test_get_audit_status_not_found(self, temp_context_dir):
"""Test that get_audit_status handles non-existent audit ID."""
# Setup: Initialize state with no audits
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time(),
"code_audits": {}
}
write_with_lock(state_file, initial_state)
# Test: Try to get status of non-existent audit
result = get_audit_status("audit-nonexistent")
# Verify: Error message
assert "ERROR" in result
assert "not found" in result
class TestListCodeAudits:
"""Test suite for listing code audits with filters."""
def test_list_code_audits_all(self, temp_context_dir):
"""Test that list_code_audits lists all audits without filters."""
# Setup: Create state with multiple audits
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"files": ["file1.py"],
"description": "Change 1",
"proposed_changes": "Changes 1",
"priority": "high",
"status": "pending",
"created_at": now - 300
},
"audit-002": {
"id": "audit-002",
"requesting_agent": "Node-Dev-002",
"files": ["file2.py"],
"description": "Change 2",
"proposed_changes": "Changes 2",
"priority": "medium",
"status": "approved",
"created_at": now - 200,
"reviewer_agent": "Node-Reviewer-001"
},
"audit-003": {
"id": "audit-003",
"requesting_agent": "Node-Dev-003",
"files": ["file3.py"],
"description": "Change 3",
"proposed_changes": "Changes 3",
"priority": "low",
"status": "rejected",
"created_at": now - 100,
"reviewer_agent": "Node-Reviewer-001"
}
}
}
write_with_lock(state_file, initial_state)
# Test: List all audits
result = list_code_audits()
# Verify: All audits are listed
assert "Code Audits" in result
assert "audit-001" in result
assert "audit-002" in result
assert "audit-003" in result
assert "1 pending, 1 approved, 1 rejected" in result
def test_list_code_audits_filter_by_status(self, temp_context_dir):
"""Test that list_code_audits filters by status correctly."""
# Setup: Create state with multiple audits
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"files": ["file1.py"],
"description": "Change 1",
"proposed_changes": "Changes 1",
"priority": "high",
"status": "pending",
"created_at": now - 300
},
"audit-002": {
"id": "audit-002",
"requesting_agent": "Node-Dev-002",
"files": ["file2.py"],
"description": "Change 2",
"proposed_changes": "Changes 2",
"priority": "medium",
"status": "approved",
"created_at": now - 200,
"reviewer_agent": "Node-Reviewer-001"
},
"audit-003": {
"id": "audit-003",
"requesting_agent": "Node-Dev-003",
"files": ["file3.py"],
"description": "Change 3",
"proposed_changes": "Changes 3",
"priority": "low",
"status": "pending",
"created_at": now - 100
}
}
}
write_with_lock(state_file, initial_state)
# Test: Filter by status
result = list_code_audits(status_filter="pending")
# Verify: Only pending audits are listed
assert "audit-001" in result
assert "audit-003" in result
assert "audit-002" not in result
def test_list_code_audits_filter_by_agent(self, temp_context_dir):
"""Test that list_code_audits filters by agent correctly."""
# Setup: Create state with multiple audits
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"files": ["file1.py"],
"description": "Change 1",
"proposed_changes": "Changes 1",
"priority": "high",
"status": "pending",
"created_at": now - 300
},
"audit-002": {
"id": "audit-002",
"requesting_agent": "Node-Dev-002",
"files": ["file2.py"],
"description": "Change 2",
"proposed_changes": "Changes 2",
"priority": "medium",
"status": "approved",
"created_at": now - 200,
"reviewer_agent": "Node-Reviewer-001"
},
"audit-003": {
"id": "audit-003",
"requesting_agent": "Node-Dev-001",
"files": ["file3.py"],
"description": "Change 3",
"proposed_changes": "Changes 3",
"priority": "low",
"status": "rejected",
"created_at": now - 100,
"reviewer_agent": "Node-Reviewer-002"
}
}
}
write_with_lock(state_file, initial_state)
# Test: Filter by requesting agent
result = list_code_audits(agent_filter="Node-Dev-001")
# Verify: Only audits by Node-Dev-001 are listed
assert "audit-001" in result
assert "audit-003" in result
assert "audit-002" not in result
# Test: Filter by reviewing agent
result2 = list_code_audits(agent_filter="Node-Reviewer-001")
# Verify: Only audits reviewed by Node-Reviewer-001 are listed
assert "audit-002" in result2
assert "audit-001" not in result2
def test_list_code_audits_respects_limit(self, temp_context_dir):
"""Test that list_code_audits respects the limit parameter."""
# Setup: Create state with many audits
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
audits = {}
for i in range(25):
audits[f"audit-{i:03d}"] = {
"id": f"audit-{i:03d}",
"requesting_agent": "Node-Dev-001",
"files": [f"file{i}.py"],
"description": f"Change {i}",
"proposed_changes": f"Changes {i}",
"priority": "medium",
"status": "pending",
"created_at": now - (i * 10)
}
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": audits
}
write_with_lock(state_file, initial_state)
# Test: List with limit
result = list_code_audits(limit=5)
# Verify: Only 5 audits are listed
lines = [line for line in result.split("\n") if line.strip().startswith("⏳") or line.strip().startswith("✅") or line.strip().startswith("❌")]
assert len(lines) == 5
def test_list_code_audits_empty(self, temp_context_dir):
"""Test that list_code_audits handles empty state gracefully."""
# Setup: Initialize state with no audits
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
# Test: List audits
result = list_code_audits()
# Verify: Returns appropriate message
assert "No code audits found" in result
def test_list_code_audits_no_matches_with_filters(self, temp_context_dir):
"""Test that list_code_audits returns appropriate message when filters match nothing."""
# Setup: Create state with audits
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
now = time.time()
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": now,
"code_audits": {
"audit-001": {
"id": "audit-001",
"requesting_agent": "Node-Dev-001",
"files": ["file1.py"],
"description": "Change 1",
"proposed_changes": "Changes 1",
"priority": "high",
"status": "pending",
"created_at": now
}
}
}
write_with_lock(state_file, initial_state)
# Test: List with filters that match nothing
result = list_code_audits(status_filter="approved", agent_filter="Node-Dev-999")
# Verify: Returns appropriate message
assert "No code audits found matching filters" in result
if __name__ == "__main__":
pytest.main([__file__, "-v"])
