elliottsax/autocoder-test-planning-worker
0
1#!/usr/bin/env python32"""3Hugging Face Space: Autocoder Test Planning Worker424/7 autonomous test planning for the Once project5 6This worker continuously analyzes test requirements, discovers edge cases,7and pushes test specifications to Hugging Face Datasets.8"""9 10import gradio as gr11import asyncio12import json13import time14from datetime import datetime15from pathlib import Path16from typing import Dict, List17import os18 19# Hugging Face integration20from huggingface_hub import HfApi, create_repo, upload_file21from datasets import Dataset, load_dataset22 23# Configuration24HF_TOKEN = os.getenv("HF_TOKEN")25HF_USERNAME = os.getenv("HF_USERNAME", "autocoder")26DATASET_NAME = f"{HF_USERNAME}/autocoder-test-planning-results"27SPACE_NAME = f"{HF_USERNAME}/autocoder-test-planning-worker"28 29class TestPlanningWorker:30 """24/7 Autonomous Test Planning Worker"""31 32 def __init__(self):33 self.iteration = 034 self.running = True35 self.status = "Initializing..."36 self.latest_results = {}37 self.hf_api = HfApi()38 39 # Cumulative insights across iterations40 self.failure_modes = []41 self.integration_points = []42 self.edge_cases = []43 self.test_specifications = []44 45 # Create dataset if it doesn't exist46 try:47 self.dataset = load_dataset(DATASET_NAME, split="train")48 except:49 self.dataset = None50 self.status = "Creating new dataset..."51 52 async def perspective_1_what_testing(self):53 """What are we actually testing?"""54 return {55 "perspective": "What Are We Testing?",56 "analysis": {57 "scope": "Integration of ScriptValidator into LongFormScriptGenerator",58 "in_scope": [59 "Import statement works",60 "Validator instantiation",61 "validate_and_fix() call integration",62 "Scene array updates correctly",63 "Validation results included in output"64 ],65 "out_of_scope": [66 "ScriptValidator internals (already tested)",67 "Gemini API behavior (external)",68 "Unrelated components"69 ],70 "integration_size": "1 import + ~20 lines of code"71 }72 }73 74 async def perspective_2_failure_modes(self):75 """How can this integration fail?"""76 failure_modes = [77 {78 "mode": "Import failure",79 "probability": "LOW",80 "impact": "CRITICAL",81 "test": "test_import_succeeds()"82 },83 {84 "mode": "Validator crashes",85 "probability": "LOW",86 "impact": "HIGH",87 "test": "test_error_handling()"88 },89 {90 "mode": "Removes legitimate scenes",91 "probability": "LOW",92 "impact": "HIGH",93 "test": "test_with_perfect_script()"94 },95 {96 "mode": "Validation too lenient",97 "probability": "LOW",98 "impact": "HIGH",99 "test": "test_with_all_identical_scenes()"100 },101 {102 "mode": "Scene numbering breaks",103 "probability": "MEDIUM",104 "impact": "MEDIUM",105 "test": "test_scene_numbering_sequential()"106 }107 ]108 109 self.failure_modes = failure_modes110 return {111 "perspective": "Failure Mode Analysis",112 "failure_modes": failure_modes,113 "total_identified": len(failure_modes)114 }115 116 async def perspective_3_integration_points(self):117 """Where do systems integrate?"""118 integration_points = [119 {120 "point": "Import statement",121 "risk": "LOW",122 "test_strategy": "Unit test"123 },124 {125 "point": "Validator instantiation",126 "risk": "LOW",127 "test_strategy": "Unit test"128 },129 {130 "point": "Script data structure",131 "risk": "MEDIUM",132 "test_strategy": "Integration test"133 },134 {135 "point": "validate_and_fix call",136 "risk": "MEDIUM",137 "test_strategy": "Integration test"138 },139 {140 "point": "Scene array update",141 "risk": "HIGH",142 "test_strategy": "Critical integration test"143 },144 {145 "point": "Result dict construction",146 "risk": "MEDIUM",147 "test_strategy": "Integration test"148 },149 {150 "point": "Logging integration",151 "risk": "LOW",152 "test_strategy": "Integration test"153 }154 ]155 156 self.integration_points = integration_points157 return {158 "perspective": "Integration Points",159 "integration_points": integration_points,160 "total_identified": len(integration_points)161 }162 163 async def perspective_4_edge_cases(self):164 """What boundary conditions exist?"""165 edge_cases = [166 {"case": "Empty script (0 scenes)", "priority": "HIGH"},167 {"case": "Single scene", "priority": "HIGH"},168 {"case": "All scenes identical", "priority": "HIGH"},169 {"case": "Already-perfect script", "priority": "CRITICAL"},170 {"case": "Exactly at 90% threshold", "priority": "HIGH"},171 {"case": "Below 90% threshold", "priority": "HIGH"},172 {"case": "Case/whitespace variations", "priority": "HIGH"},173 {"case": "Empty narration", "priority": "MEDIUM"},174 {"case": "Missing fields", "priority": "MEDIUM"},175 {"case": "Special characters", "priority": "MEDIUM"}176 ]177 178 self.edge_cases = edge_cases179 return {180 "perspective": "Edge Case Analysis",181 "edge_cases": edge_cases,182 "total_identified": len(edge_cases)183 }184 185 async def perspective_5_performance(self):186 """What is the performance impact?"""187 return {188 "perspective": "Performance Analysis",189 "analysis": {190 "expected_overhead": "<1% (validation is O(n), n is small)",191 "acceptable_threshold": "<10%",192 "baseline_needed": True,193 "tests": [194 "test_baseline_performance()",195 "test_performance_with_validation()",196 "test_validation_overhead_acceptable()"197 ]198 }199 }200 201 async def perspective_6_backwards_compatibility(self):202 """Does this break existing functionality?"""203 return {204 "perspective": "Backwards Compatibility",205 "analysis": {206 "breaking_changes": [207 "Output structure (added 'validation' field)",208 "Scene count reduction (intentional)"209 ],210 "test_updates_needed": [211 "Scene count expectations",212 "Output structure assertions"213 ],214 "risk": "LOW - changes are intentional and documented"215 }216 }217 218 async def perspective_7_data_integrity(self):219 """Is data handled correctly?"""220 return {221 "perspective": "Data Integrity",222 "analysis": {223 "data_transformations": [224 "Scene deduplication",225 "Scene renumbering"226 ],227 "invariants": [228 "No legitimate scenes lost",229 "Scene order preserved",230 "All fields maintained"231 ],232 "tests": [233 "test_no_data_loss()",234 "test_field_preservation()",235 "test_order_maintained()"236 ]237 }238 }239 240 async def perspective_8_error_handling(self):241 """How do we handle failures?"""242 return {243 "perspective": "Error Handling",244 "analysis": {245 "error_scenarios": [246 "Validator throws exception",247 "Malformed script data",248 "Missing required fields"249 ],250 "recovery_strategy": "Graceful degradation - return original script",251 "tests": [252 "test_validator_exception_handled()",253 "test_malformed_data_handled()",254 "test_fallback_to_original()"255 ]256 }257 }258 259 async def perspective_9_user_experience(self):260 """How does this affect users?"""261 return {262 "perspective": "User Experience",263 "analysis": {264 "user_impact": [265 "Fewer duplicate scenes → Better quality",266 "Shorter videos → Faster processing",267 "Validation metrics → Visibility into quality"268 ],269 "potential_concerns": [270 "Unexpected scene count changes",271 "Need documentation"272 ],273 "mitigation": "Clear logging and documentation"274 }275 }276 277 async def perspective_10_security(self):278 """Are there security implications?"""279 return {280 "perspective": "Security Analysis",281 "analysis": {282 "risks": "NONE - Local data processing only",283 "attack_vectors": "N/A",284 "mitigation": "Not applicable",285 "conclusion": "No security concerns for this integration"286 }287 }288 289 async def perspective_11_test_coverage(self):290 """What is our test coverage strategy?"""291 return {292 "perspective": "Test Coverage Strategy",293 "analysis": {294 "coverage_goals": {295 "unit": "100%",296 "integration": "90%",297 "e2e": "Key scenarios"298 },299 "critical_paths": [300 "Scene deduplication",301 "Data flow correctness",302 "Error handling"303 ],304 "coverage_tools": ["pytest-cov", "coverage.py"]305 }306 }307 308 async def perspective_12_test_pyramid(self):309 """How should tests be structured?"""310 return {311 "perspective": "Test Pyramid Analysis",312 "pyramid": {313 "e2e": {"count": 2, "focus": "Full pipeline"},314 "integration": {"count": 15, "focus": "Component interaction"},315 "unit": {"count": 5, "focus": "Individual functions"},316 "performance": {"count": 3, "focus": "Speed benchmarks"}317 },318 "total_tests": 25,319 "balance": "Well-balanced pyramid"320 }321 322 async def perspective_13_mocking_strategy(self):323 """What should be mocked vs real?"""324 return {325 "perspective": "Mocking Strategy",326 "strategy": {327 "mock": [328 "Gemini API (external, expensive)",329 "File I/O (slow, side effects)"330 ],331 "real": [332 "ScriptValidator (core functionality)",333 "Data structures (need real behavior)"334 ],335 "fixtures": [336 "perfect_script",337 "buggy_script",338 "edge_case_empty"339 ]340 }341 }342 343 async def perspective_14_ci_cd(self):344 """How do tests integrate with CI/CD?"""345 return {346 "perspective": "CI/CD Integration",347 "integration": {348 "run_on": ["push", "pull_request"],349 "requirements": ["All tests must pass", "Coverage > 80%"],350 "tools": ["pytest", "pytest-asyncio", "pytest-cov"],351 "reporting": "Coverage reports uploaded to CI"352 }353 }354 355 async def perspective_15_big_picture(self):356 """Holistic view of testing strategy"""357 return {358 "perspective": "Big Picture Review",359 "summary": {360 "scope": "Small, focused integration",361 "risk": "LOW - using existing tested code",362 "coverage": "Comprehensive - 25 tests across 4 levels",363 "readiness": "Ready to implement",364 "confidence": "VERY HIGH",365 "next_steps": [366 "Implement Phase 1 (5 critical tests)",367 "Verify core integration",368 "Add edge cases",369 "Complete full suite"370 ]371 }372 }373 374 async def analyze_tests(self):375 """Perform test planning iteration"""376 self.iteration += 1377 self.status = f"Running iteration {self.iteration}..."378 379 # All 15 perspectives380 perspectives = [381 self.perspective_1_what_testing,382 self.perspective_2_failure_modes,383 self.perspective_3_integration_points,384 self.perspective_4_edge_cases,385 self.perspective_5_performance,386 self.perspective_6_backwards_compatibility,387 self.perspective_7_data_integrity,388 self.perspective_8_error_handling,389 self.perspective_9_user_experience,390 self.perspective_10_security,391 self.perspective_11_test_coverage,392 self.perspective_12_test_pyramid,393 self.perspective_13_mocking_strategy,394 self.perspective_14_ci_cd,395 self.perspective_15_big_picture396 ]397 398 # Rotate through perspectives399 perspective_index = self.iteration % len(perspectives)400 current_perspective = perspectives[perspective_index]401 402 analysis_result = await current_perspective()403 404 result = {405 "iteration": self.iteration,406 "timestamp": datetime.now().isoformat(),407 "perspective_number": perspective_index + 1,408 "perspective_name": analysis_result["perspective"],409 "analysis": analysis_result,410 "cumulative_insights": {411 "failure_modes_identified": len(self.failure_modes),412 "integration_points_mapped": len(self.integration_points),413 "edge_cases_discovered": len(self.edge_cases)414 }415 }416 417 self.latest_results = result418 419 # Push to Hugging Face Dataset420 await self.push_results(result)421 422 self.status = f"✓ Iteration {self.iteration} complete - {analysis_result['perspective']}"423 424 return result425 426 async def push_results(self, results: Dict):427 """Push results to Hugging Face Dataset"""428 if not HF_TOKEN:429 self.status += " (No HF_TOKEN - results not pushed)"430 return431 432 try:433 # Create dataset repo if it doesn't exist434 try:435 create_repo(436 repo_id=DATASET_NAME,437 token=HF_TOKEN,438 repo_type="dataset",439 exist_ok=True440 )441 except:442 pass443 444 # Save results locally445 results_file = Path("/tmp/test_plan_results.json")446 with open(results_file, 'w') as f:447 json.dump(results, f, indent=2)448 449 # Upload to dataset450 upload_file(451 path_or_fileobj=str(results_file),452 path_in_repo=f"test_plan_iteration_{self.iteration}.json",453 repo_id=DATASET_NAME,454 repo_type="dataset",455 token=HF_TOKEN456 )457 458 self.status += " (Pushed to HF Dataset)"459 460 except Exception as e:461 self.status += f" (Push failed: {e})"462 463 async def run_continuous(self):464 """Run continuous test planning loop"""465 while self.running:466 try:467 await self.analyze_tests()468 await asyncio.sleep(60) # Wait 1 minute between iterations469 except Exception as e:470 self.status = f"Error: {e}"471 await asyncio.sleep(10)472 473# Global worker instance474worker = TestPlanningWorker()475 476# Gradio Interface477def get_status():478 """Get current worker status"""479 return {480 "Status": worker.status,481 "Iteration": worker.iteration,482 "Failure Modes Identified": len(worker.failure_modes),483 "Integration Points Mapped": len(worker.integration_points),484 "Edge Cases Discovered": len(worker.edge_cases),485 "Latest Analysis": json.dumps(worker.latest_results, indent=2) if worker.latest_results else "No results yet"486 }487 488def start_worker():489 """Start the worker"""490 worker.running = True491 return "Worker is already running in background!"492 493def stop_worker():494 """Stop the worker"""495 worker.running = False496 return "Worker stopped!"497 498def get_full_log():499 """Get full test planning log"""500 if not HF_TOKEN:501 return "No HF_TOKEN configured - cannot fetch logs"502 503 try:504 # Load all results from dataset505 dataset = load_dataset(DATASET_NAME, split="train")506 return f"Total iterations: {len(dataset)}\n\nLatest results:\n{json.dumps(worker.latest_results, indent=2)}"507 except Exception as e:508 return f"Error loading dataset: {e}"509 510# Build Gradio UI511with gr.Blocks(title="Autocoder Test Planning Worker") as demo:512 gr.Markdown("""513 # 🧪 Autocoder Test Planning Worker514 515 **24/7 Autonomous Test Planning**516 517 This worker continuously analyzes test requirements for the Once project,518 discovering failure modes, edge cases, and integration points.519 520 ## Status521 """)522 523 status_display = gr.JSON(label="Current Status", value=get_status)524 525 with gr.Row():526 start_btn = gr.Button("▶ Start Worker", variant="primary")527 stop_btn = gr.Button("⏸ Stop Worker", variant="stop")528 refresh_btn = gr.Button("🔄 Refresh Status")529 530 start_output = gr.Textbox(label="Action Result")531 532 gr.Markdown("## Test Planning Log")533 log_display = gr.Textbox(label="Full Log", lines=20)534 535 gr.Markdown("""536 ## Analysis Perspectives (15)537 538 1. **What Are We Testing?** - Scope and boundaries539 2. **Failure Mode Analysis** - How can it fail?540 3. **Integration Points** - Where systems connect541 4. **Edge Case Discovery** - Boundary conditions542 5. **Performance Impact** - Speed analysis543 6. **Backwards Compatibility** - Breaking changes544 7. **Data Integrity** - Data correctness545 8. **Error Handling** - Failure recovery546 9. **User Experience** - UX implications547 10. **Security Analysis** - Vulnerability review548 11. **Test Coverage** - Coverage strategy549 12. **Test Pyramid** - Test structure550 13. **Mocking Strategy** - What to mock551 14. **CI/CD Integration** - Automation552 15. **Big Picture Review** - Holistic view553 554 ## Configuration555 556 - **Dataset**: `autocoder/autocoder-test-planning-results`557 - **Update Frequency**: Every 60 seconds558 - **Perspectives**: 15 rotating perspectives559 560 ## Setup561 562 To enable dataset pushing, configure:563 1. Add `HF_TOKEN` secret in Space settings564 2. Set `HF_USERNAME` to your username (default: `autocoder`)565 566 ## Results567 568 All test planning results are pushed to the Hugging Face Dataset and can be:569 - Downloaded programmatically570 - Viewed in the Datasets UI571 - Used by implementation workers572 """)573 574 # Button actions575 start_btn.click(fn=start_worker, outputs=start_output)576 stop_btn.click(fn=stop_worker, outputs=start_output)577 refresh_btn.click(fn=get_status, outputs=status_display)578 579 # Auto-refresh status every 5 seconds580 demo.load(fn=get_status, outputs=status_display, every=5)581 582if __name__ == "__main__":583 # Auto-start worker in background584 import threading585 threading.Thread(target=lambda: asyncio.run(worker.run_continuous()), daemon=True).start()586 587 demo.launch(server_name="0.0.0.0", server_port=7860)588