""" 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)}")