237 lines
9.5 KiB
Python
237 lines
9.5 KiB
Python
"""
|
|
Test suite for Brand Monitor capstone reference.
|
|
|
|
Tests the full pipeline: scan → analyze → report
|
|
Uses mock LLM for offline testing.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
# Add mock LLM
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / 'labs' / '_shared'))
|
|
try:
|
|
from mock_llm import MockAnthropic
|
|
MOCK_AVAILABLE = True
|
|
except ImportError:
|
|
MOCK_AVAILABLE = False
|
|
|
|
|
|
class TestScannerAgent:
|
|
"""Tests for the Scanner agent."""
|
|
|
|
def test_read_brand_config(self):
|
|
"""Scanner should be able to read brand configuration."""
|
|
config_path = Path(__file__).parent.parent / 'config' / 'brand.yaml'
|
|
assert config_path.exists(), "brand.yaml not found"
|
|
content = config_path.read_text()
|
|
assert 'name:' in content
|
|
assert 'AcmeCorp' in content
|
|
|
|
def test_scan_output_format(self):
|
|
"""Scanner output should be valid JSONL with required fields."""
|
|
scan_path = Path(__file__).parent.parent / 'data' / 'sample_scan.jsonl'
|
|
assert scan_path.exists(), "sample_scan.jsonl not found"
|
|
|
|
with open(scan_path) as f:
|
|
for line in f:
|
|
record = json.loads(line)
|
|
assert 'service' in record, f"Missing 'service' in: {line}"
|
|
assert 'query' in record, f"Missing 'query' in: {line}"
|
|
assert 'response' in record, f"Missing 'response' in: {line}"
|
|
assert 'mentioned' in record, f"Missing 'mentioned' in: {line}"
|
|
assert isinstance(record['mentioned'], bool), "'mentioned' must be boolean"
|
|
|
|
print(f" PASS: {sum(1 for _ in open(scan_path))} scan records valid")
|
|
|
|
|
|
class TestAnalyzerAgent:
|
|
"""Tests for the Analyzer agent."""
|
|
|
|
def test_analysis_output_format(self):
|
|
"""Analyzer output should be valid JSONL with sentiment classification."""
|
|
analysis_path = Path(__file__).parent.parent / 'data' / 'sample_analysis.jsonl'
|
|
assert analysis_path.exists(), "sample_analysis.jsonl not found"
|
|
|
|
valid_sentiments = {'positive', 'negative', 'neutral'}
|
|
with open(analysis_path) as f:
|
|
for line in f:
|
|
record = json.loads(line)
|
|
assert 'sentiment' in record, f"Missing 'sentiment' in: {line}"
|
|
assert record['sentiment'] in valid_sentiments, \
|
|
f"Invalid sentiment '{record['sentiment']}' in: {line}"
|
|
assert 'confidence' in record, f"Missing 'confidence' in: {line}"
|
|
assert 0 <= record['confidence'] <= 1, \
|
|
f"Confidence out of range: {record['confidence']}"
|
|
|
|
print(f" PASS: {sum(1 for _ in open(analysis_path))} analysis records valid")
|
|
|
|
def test_sentiment_distribution(self):
|
|
"""Analysis should have a reasonable distribution of sentiments."""
|
|
analysis_path = Path(__file__).parent.parent / 'data' / 'sample_analysis.jsonl'
|
|
sentiments = []
|
|
with open(analysis_path) as f:
|
|
for line in f:
|
|
record = json.loads(line)
|
|
sentiments.append(record['sentiment'])
|
|
|
|
positive_pct = sentiments.count('positive') / len(sentiments) * 100
|
|
negative_pct = sentiments.count('negative') / len(sentiments) * 100
|
|
|
|
print(f" Distribution: {positive_pct:.0f}% positive, {negative_pct:.0f}% negative")
|
|
assert negative_pct < 50, "Too many negative mentions (may trigger false alert)"
|
|
|
|
|
|
class TestReporterAgent:
|
|
"""Tests for the Reporter agent."""
|
|
|
|
def test_report_generation(self):
|
|
"""Reporter should be able to generate a summary from analysis data."""
|
|
analysis_path = Path(__file__).parent.parent / 'data' / 'sample_analysis.jsonl'
|
|
assert analysis_path.exists()
|
|
|
|
# Calculate summary stats (what reporter should do)
|
|
sentiments = []
|
|
services = set()
|
|
with open(analysis_path) as f:
|
|
for line in f:
|
|
record = json.loads(line)
|
|
sentiments.append(record['sentiment'])
|
|
services.add(record['service'])
|
|
|
|
total = len(sentiments)
|
|
positive_pct = sentiments.count('positive') / total * 100
|
|
negative_pct = sentiments.count('negative') / total * 100
|
|
|
|
report = f"""# Brand Monitor Report — Test
|
|
|
|
## Summary
|
|
- Total mentions: {total}
|
|
- Positive: {sentiments.count('positive')} ({positive_pct:.0f}%)
|
|
- Negative: {sentiments.count('negative')} ({negative_pct:.0f}%)
|
|
- Neutral: {sentiments.count('neutral')} ({100-positive_pct-negative_pct:.0f}%)
|
|
- Services: {', '.join(sorted(services))}
|
|
"""
|
|
print(f" Generated report ({len(report)} chars)")
|
|
assert 'positive' in report.lower()
|
|
assert str(total) in report
|
|
|
|
def test_negative_alert_threshold(self):
|
|
"""Report should flag if negative mentions exceed threshold."""
|
|
analysis_path = Path(__file__).parent.parent / 'data' / 'sample_analysis.jsonl'
|
|
sentiments = []
|
|
with open(analysis_path) as f:
|
|
for line in f:
|
|
record = json.loads(line)
|
|
sentiments.append(record['sentiment'])
|
|
|
|
negative_pct = sentiments.count('negative') / len(sentiments) * 100
|
|
threshold = 20 # from brand.yaml
|
|
|
|
if negative_pct > threshold:
|
|
print(f" ALERT: {negative_pct:.0f}% negative (threshold: {threshold}%)")
|
|
else:
|
|
print(f" OK: {negative_pct:.0f}% negative (threshold: {threshold}%)")
|
|
|
|
|
|
class TestSecurityArchitecture:
|
|
"""Tests for security architecture."""
|
|
|
|
def test_damage_control_rules(self):
|
|
"""Damage control rules should be valid YAML with required sections."""
|
|
import yaml
|
|
rules_path = Path(__file__).parent.parent / 'config' / 'damage-control-rules.yaml'
|
|
assert rules_path.exists()
|
|
|
|
with open(rules_path) as f:
|
|
rules = yaml.safe_load(f)
|
|
|
|
assert 'zeroAccessPaths' in rules, "Missing zeroAccessPaths"
|
|
assert 'readOnlyPaths' in rules, "Missing readOnlyPaths"
|
|
assert 'noDeletePaths' in rules, "Missing noDeletePaths"
|
|
assert 'bashToolPatterns' in rules, "Missing bashToolPatterns"
|
|
print(f" PASS: {len(rules['bashToolPatterns'])} bash patterns defined")
|
|
|
|
def test_agent_domain_permissions(self):
|
|
"""Each agent should have domain permissions defined."""
|
|
agents_dir = Path(__file__).parent.parent / 'agents'
|
|
assert agents_dir.exists()
|
|
|
|
agent_files = list(agents_dir.glob('*.md'))
|
|
assert len(agent_files) >= 3, f"Expected 3+ agents, found {len(agent_files)}"
|
|
|
|
import frontmatter
|
|
for agent_file in agent_files:
|
|
post = frontmatter.load(agent_file)
|
|
assert 'domain' in post.metadata, f"{agent_file.name} missing domain permissions"
|
|
assert 'tools' in post.metadata, f"{agent_file.name} missing tool definitions"
|
|
|
|
print(f" PASS: {len(agent_files)} agents with domain permissions")
|
|
|
|
|
|
class TestCostEstimate:
|
|
"""Tests for cost estimation."""
|
|
|
|
def test_cost_calculation(self):
|
|
"""Cost estimate should be within expected range."""
|
|
# Scanner: Gemini Flash at $0.15/$0.60 per M tokens
|
|
# Analyzer: Claude Sonnet at $3/$15 per M tokens
|
|
# Reporter: Claude Opus at $15/$75 per M tokens
|
|
|
|
costs = {
|
|
'scanner': {'input': 2000, 'output': 500, 'model': ('gemini-flash', 0.15, 0.60)},
|
|
'analyzer': {'input': 3000, 'output': 800, 'model': ('claude-sonnet', 3.00, 15.00)},
|
|
'reporter': {'input': 5000, 'output': 2000, 'model': ('claude-opus', 15.00, 75.00)},
|
|
}
|
|
|
|
total = 0
|
|
for name, details in costs.items():
|
|
in_tok = details['input']
|
|
out_tok = details['output']
|
|
in_price, out_price = details['model'][1], details['model'][2]
|
|
|
|
cost = (in_tok * in_price / 1_000_000) + (out_tok * out_price / 1_000_000)
|
|
total += cost
|
|
print(f" {name}: ${cost:.6f}")
|
|
|
|
print(f" Total per run: ${total:.4f}")
|
|
assert total < 1.0, f"Cost ${total:.2f} exceeds $1.00 budget"
|
|
print(f" Daily (4 runs): ${total*4:.2f}")
|
|
print(f" Monthly (30 days): ${total*4*30:.2f}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("=" * 50)
|
|
print("Brand Monitor Capstone — Test Suite")
|
|
print("=" * 50)
|
|
print()
|
|
|
|
tests = [
|
|
("Scanner: Read config", TestScannerAgent().test_read_brand_config),
|
|
("Scanner: Output format", TestScannerAgent().test_scan_output_format),
|
|
("Analyzer: Output format", TestAnalyzerAgent().test_analysis_output_format),
|
|
("Analyzer: Sentiment dist", TestAnalyzerAgent().test_sentiment_distribution),
|
|
("Reporter: Generate report", TestReporterAgent().test_report_generation),
|
|
("Reporter: Alert threshold", TestReporterAgent().test_negative_alert_threshold),
|
|
("Security: Damage control", TestSecurityArchitecture().test_damage_control_rules),
|
|
("Security: Domain perms", TestSecurityArchitecture().test_agent_domain_permissions),
|
|
("Cost: Calculation", TestCostEstimate().test_cost_calculation),
|
|
]
|
|
|
|
passed = 0
|
|
failed = 0
|
|
for name, test_fn in tests:
|
|
try:
|
|
test_fn()
|
|
print(f" [{chr(10003)}] {name}")
|
|
passed += 1
|
|
except Exception as e:
|
|
print(f" [x] {name}: {e}")
|
|
failed += 1
|
|
print()
|
|
|
|
print(f"Results: {passed} passed, {failed} failed out of {len(tests)}")
|