diff --git a/.jules/bolt.md b/.jules/bolt.md new file mode 100644 index 000000000..f831b464f --- /dev/null +++ b/.jules/bolt.md @@ -0,0 +1,4 @@ +# 2023-11-20 - [Compile Regex Patterns Once in API Routing] + +**Learning:** Recompiling regular expressions on every incoming request for path parameter matching causes an unnecessary performance bottleneck. The `_path_to_regex` method in `APIRouter` was rebuilding and recompiling its parameter extraction regex for each endpoint evaluation loop. +**Action:** Always cache or memoize compiled regex patterns in high-traffic routing logic or request parsers, either via `functools.lru_cache` or by storing them in a dictionary keyed by the endpoint path. diff --git a/scripts/agents/hermes/evaluate_orchestrators.py b/scripts/agents/hermes/evaluate_orchestrators.py index cd0f319ef..643d453cb 100644 --- a/scripts/agents/hermes/evaluate_orchestrators.py +++ b/scripts/agents/hermes/evaluate_orchestrators.py @@ -281,7 +281,7 @@ def extract_json_from_response(content: str) -> dict: def assess_script(client, script_info: dict, source_code: str) -> dict: """Use Hermes to assess a script based on stdout, stderr, and source code.""" script_name = script_info["path"].name - + prompt = f"""You are a senior principal engineer performing a code review and architectural assessment. Evaluate the following Python orchestration script based on the Codomyrmex "Thin Orchestrator" pattern. @@ -333,7 +333,7 @@ def assess_script(client, script_info: dict, source_code: str) -> dict: response = client.chat_session(prompt=prompt) if response.is_success(): eval_data = extract_json_from_response(response.content) - + print_success(f"=== Hermes Assessment for {script_name} ===") print_info(f" Adheres to pattern: {eval_data.get('adherence_assessment', {}).get('adheres', False)}") print_success("=" * 60) @@ -387,7 +387,7 @@ def main() -> int: trace_path: Optional[Path] = (_REPO_ROOT / trace_file_rel) if trace_file_rel else None if trace_path: trace_path.parent.mkdir(parents=True, exist_ok=True) - + # 1. Boot up Hermes Client if not args.dry_run: try: @@ -403,7 +403,7 @@ def main() -> int: else: client = None print_info(" [DRY RUN] Hermes assessment will be skipped.") - + print_info("═" * 60) print_info(f" Hermes Script Evaluator — Target: scripts/{args.target}") print_info(f" Saving outputs to: {output_path}") @@ -480,7 +480,7 @@ def main() -> int: script_path, timeout=evaluator_cfg.get("script_timeout", 30), ) - + # Read source code try: with open(script_path, "r", encoding="utf-8") as f: @@ -488,7 +488,7 @@ def main() -> int: except Exception as e: print_error(f" Could not read source code for {script_path.name}: {e}") continue - + # Assess (skip in dry-run mode) if args.dry_run: print_info(f" [DRY RUN] Would assess {script_path.name} with Hermes.") @@ -530,25 +530,25 @@ def main() -> int: f.write(f"# Evaluator Orchestrations Report\n") f.write(f"Target: `scripts/{args.target}`\n") f.write(f"Generated: {datetime.now().isoformat()}\n\n") - + for script_name, data in all_evaluations.items(): f.write(f"## Script: `{script_name}`\n") - + adherence = data.get("adherence_assessment", {}) pass_fail = "✅ STRICT ADHERENCE" if adherence.get("adheres", False) else "❌ NON-COMPLIANT" f.write(f"**Pattern Adherence**: {pass_fail}\n\n") f.write(f"> {adherence.get('reasoning', 'No reasoning provided.')}\n\n") - + f.write("### Technical Debt Identified:\n") for debt in data.get("technical_debt", []): f.write(f"- {debt}\n") - + f.write("\n### Underlying Method Improvements Required:\n") for imp in data.get("underlying_improvements", []): f.write(f"- {imp}\n") - + f.write("\n---\n\n") - + print_success(f"Successfully compiled overall markdown report to: {report_file_path}") except Exception as e: print_error(f"Failed to save overall evaluation report: {e}") diff --git a/src/codomyrmex/agents/hermes/hermes_client.py b/src/codomyrmex/agents/hermes/hermes_client.py index 7a0a437dd..98d6e19a8 100644 --- a/src/codomyrmex/agents/hermes/hermes_client.py +++ b/src/codomyrmex/agents/hermes/hermes_client.py @@ -286,7 +286,7 @@ def chat_session(self, prompt: str, session_id: str | None = None) -> AgentRespo history_text = "" for msg in session.messages[:-1]: # exclude the current user prompt history_text += f"[{msg['role'].upper()}]\n{msg['content']}\n\n" - + if history_text: full_prompt = ( f"Previous Conversation:\n{history_text}" @@ -302,7 +302,7 @@ def chat_session(self, prompt: str, session_id: str | None = None) -> AgentRespo if response.is_success(): session.add_message("assistant", response.content) store.save(session) - + response.metadata["session_id"] = session.session_id return response diff --git a/src/codomyrmex/api/standardization/rest_api.py b/src/codomyrmex/api/standardization/rest_api.py index 2e456e602..4ec04f38a 100644 --- a/src/codomyrmex/api/standardization/rest_api.py +++ b/src/codomyrmex/api/standardization/rest_api.py @@ -154,6 +154,7 @@ def __init__(self, prefix: str = ""): self.endpoints: dict[str, APIEndpoint] = {} self.sub_routers: list[APIRouter] = [] self.middleware: list[Callable[[APIRequest], APIResponse | None]] = [] + self._compiled_regexes: dict[str, tuple] = {} def add_endpoint(self, endpoint: APIEndpoint) -> None: """ @@ -291,6 +292,10 @@ def _path_to_regex(self, path: str) -> tuple: Returns: Tuple of (compiled_regex, parameter_names) """ + # Return cached regex if available + if path in self._compiled_regexes: + return self._compiled_regexes[path] + # Replace {param} with named capture groups param_pattern = re.compile(r"\{([^}]+)\}") param_names = param_pattern.findall(path) @@ -299,7 +304,10 @@ def _path_to_regex(self, path: str) -> tuple: regex_pattern = param_pattern.sub(r"(?P<\1>[^/]+)", path) regex_pattern = f"^{regex_pattern}$" - return re.compile(regex_pattern), param_names + compiled = re.compile(regex_pattern) + self._compiled_regexes[path] = (compiled, param_names) + + return compiled, param_names def get_all_endpoints(self) -> list[APIEndpoint]: """ @@ -457,7 +465,11 @@ def handle_request( processing_time = time.time() - start_time logger.info( - "Request completed: %s %s -> %s (%.3fs)", method, path, response.status_code.value, processing_time + "Request completed: %s %s -> %s (%.3fs)", + method, + path, + response.status_code.value, + processing_time, ) return response diff --git a/src/codomyrmex/tests/integration/calendar_integration/test_mcp_tools.py b/src/codomyrmex/tests/integration/calendar_integration/test_mcp_tools.py index 10ac3280c..8122f23ae 100644 --- a/src/codomyrmex/tests/integration/calendar_integration/test_mcp_tools.py +++ b/src/codomyrmex/tests/integration/calendar_integration/test_mcp_tools.py @@ -90,10 +90,10 @@ def test_calendar_attendee_injection(monkeypatch): """Verify that _DEFAULT_ATTENDEE is always injected as an attendee.""" test_attendee = "test@example.com" monkeypatch.setenv("CODOMYRMEX_CALENDAR_ATTENDEE", test_attendee) - - # Needs to match the env var or use monkeypatch's effect on the module. + + # Needs to match the env var or use monkeypatch's effect on the module. # But wait, mcp_tools reads os.environ at IMPORT time! - # So monkeypatching os.environ during the test execution + # So monkeypatching os.environ during the test execution # won't change mcp_tools._DEFAULT_ATTENDEE because it was evaluated at import. # We must patch mcp_tools._DEFAULT_ATTENDEE directly! import codomyrmex.calendar_integration.mcp_tools as mcp_tools_mod diff --git a/src/codomyrmex/tests/unit/agents/test_agents_hermes_client.py b/src/codomyrmex/tests/unit/agents/test_agents_hermes_client.py index 881b5f4bd..7ade0eabf 100644 --- a/src/codomyrmex/tests/unit/agents/test_agents_hermes_client.py +++ b/src/codomyrmex/tests/unit/agents/test_agents_hermes_client.py @@ -132,7 +132,7 @@ class TestHermesClientSessionIntegration: def test_chat_session_new_and_continue(self, tmp_path) -> None: """Test creating a new session and appending to it.""" db_path = tmp_path / "test_sessions.db" - + # Use echo to simulate a fast, mock-free successful execution client = HermesClient(config={ "hermes_command": "echo", @@ -170,7 +170,7 @@ def test_mcp_session_lifecycle(self, monkeypatch, tmp_path) -> None: # Patch _get_client to inject our test config from codomyrmex.agents.hermes import mcp_tools from codomyrmex.agents.hermes.hermes_client import HermesClient - + def mock_get_client(**kwargs): return HermesClient(config={ "hermes_command": "echo", @@ -179,9 +179,9 @@ def mock_get_client(**kwargs): "hermes_model": kwargs.get("model", "hermes3"), "hermes_timeout": kwargs.get("timeout", 120), }) - + monkeypatch.setattr(mcp_tools, "_get_client", mock_get_client) - + from codomyrmex.agents.hermes.mcp_tools import ( hermes_chat_session, hermes_session_clear, diff --git a/src/codomyrmex/tests/unit/audio/test_mcp_tools.py b/src/codomyrmex/tests/unit/audio/test_mcp_tools.py index e241fe912..63a26c08e 100644 --- a/src/codomyrmex/tests/unit/audio/test_mcp_tools.py +++ b/src/codomyrmex/tests/unit/audio/test_mcp_tools.py @@ -102,7 +102,7 @@ def test_audio_batch_transcribe_invalid_files( result = audio_batch_transcribe( ["/invalid/1.wav", "/invalid/2.wav"], str(temp_output_dir), model_size="tiny" ) - + if WHISPER_AVAILABLE: assert result["success"] is True assert result["result"]["processed"] == 0 @@ -118,7 +118,7 @@ def test_audio_transcribe_invalid_format(dummy_wav_file: Path) -> None: """Test transcription with invalid format (e.g. by changing suffix).""" bad_format_path = dummy_wav_file.with_suffix(".txt") dummy_wav_file.rename(bad_format_path) - + result = audio_transcribe(str(bad_format_path)) assert result["success"] is False assert "AudioFormatError" in result["error"]["type"] diff --git a/src/codomyrmex/tests/unit/auth/test_google_auth.py b/src/codomyrmex/tests/unit/auth/test_google_auth.py index 98aab51e0..c65fcee32 100644 --- a/src/codomyrmex/tests/unit/auth/test_google_auth.py +++ b/src/codomyrmex/tests/unit/auth/test_google_auth.py @@ -20,13 +20,13 @@ def test_initialization(self): with tempfile.NamedTemporaryFile() as tmp: client_secrets = tmp.name token_cache = tempfile.mktemp() - + auth = GoogleAuthenticator( client_secrets_file=client_secrets, token_cache_file=token_cache, scopes=["https://www.googleapis.com/auth/calendar"] ) - + assert auth.client_secrets_file == client_secrets assert auth.token_file == token_cache assert auth.scopes == ["https://www.googleapis.com/auth/calendar"] diff --git a/src/codomyrmex/tests/unit/bio_simulation/test_bio_simulation.py b/src/codomyrmex/tests/unit/bio_simulation/test_bio_simulation.py index 096c49681..ddadefa0e 100644 --- a/src/codomyrmex/tests/unit/bio_simulation/test_bio_simulation.py +++ b/src/codomyrmex/tests/unit/bio_simulation/test_bio_simulation.py @@ -149,11 +149,11 @@ def test_colony_idle_to_foraging(): colony = Colony(population=100) for ant in colony.ants: ant.state = AntState.IDLE - + # Run a few ticks, some should switch for _ in range(10): colony._step_tick() - + states = [ant.state for ant in colony.ants] assert AntState.FORAGING in states diff --git a/src/codomyrmex/tests/unit/build_synthesis/test_build_synthesis.py b/src/codomyrmex/tests/unit/build_synthesis/test_build_synthesis.py deleted file mode 100644 index bae7263ca..000000000 --- a/src/codomyrmex/tests/unit/build_synthesis/test_build_synthesis.py +++ /dev/null @@ -1,651 +0,0 @@ -#!/usr/bin/env python3 -""" -Unit tests for the Build Synthesis module. -""" - -import json -import os -import sys -import time - -import pytest - -# Add src to path for imports -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "src")) - -# Import build synthesis functions -try: - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - check_build_environment, - cleanup_build_artifacts, - create_build_manifest, - export_build_report, - get_build_history, - get_build_metrics, - get_supported_languages, - import_build_config, - incremental_build_check, - monitor_build_progress, - optimize_build_config, - orchestrate_build_pipeline, - parallel_build_execution, - rollback_build, - synthesize_build_artifact, - validate_build_config, - validate_build_dependencies, - validate_build_output, - ) - - FULL_BUILD_AVAILABLE = True -except ImportError: - # Fallback to basic imports - try: - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - check_build_environment, - orchestrate_build_pipeline, - synthesize_build_artifact, - validate_build_output, - ) - - FULL_BUILD_AVAILABLE = False - except ImportError: - FULL_BUILD_AVAILABLE = False - - -@pytest.mark.unit -class TestBuildSynthesis: - """Test cases for build synthesis functionality.""" - - def test_check_build_environment(self): - """Test build environment checking.""" - result = check_build_environment() - # This should work regardless of what's installed - # Function returns a dict, not a bool - assert isinstance(result, dict) - assert "python_available" in result - - def test_synthesize_python_executable(self, tmp_path): - """Test synthesis of Python executable.""" - test_dir = str(tmp_path) - source_file = os.path.join(test_dir, "test_script.py") - output_file = os.path.join(test_dir, "test_executable.py") - - # Create a simple Python script - with open(source_file, "w") as f: - f.write(""" -def main(): - print("Hello from test script!") - -if __name__ == "__main__": - main() -""") - - result = synthesize_build_artifact( - source_path=source_file, output_path=output_file, artifact_type="executable" - ) - - assert result - assert os.path.exists(output_file) - - # Check that the output contains expected content - with open(output_file) as f: - content = f.read() - assert "import" in content - # The synthesized executable may use exec() or main() depending on implementation - assert "main()" in content or "__main__" in content or "exec(" in content - - def test_validate_build_output(self, tmp_path): - """Test build output validation.""" - test_dir = str(tmp_path) - # Create a test file with proper Python code - test_file = os.path.join(test_dir, "test_output.py") - with open(test_file, "w") as f: - f.write( - "# Test Python file\nimport sys\ndef main():\n print('Hello')\n\nif __name__ == '__main__':\n main()\n" - ) - - validation = validate_build_output(test_file) - - assert validation["exists"] - assert validation["is_file"] - assert validation["size_bytes"] > 0 - assert len(validation["errors"]) == 0 - - def test_orchestrate_build_pipeline(self): - """Test build pipeline orchestration.""" - build_config = { - "dependencies": [], - "build_commands": [["python", "-c", "print('Test build command')"]], - "artifacts": [], - } - - results = orchestrate_build_pipeline(build_config) - - assert isinstance(results, dict) - assert "overall_success" in results - assert "stages" in results - assert "artifacts" in results - assert "errors" in results - - def test_synthesize_nonexistent_source(self, tmp_path): - """Test synthesis with nonexistent source file.""" - test_dir = str(tmp_path) - result = synthesize_build_artifact( - source_path="/nonexistent/file.py", - output_path=os.path.join(test_dir, "output.py"), - artifact_type="executable", - ) - - assert not result - - def test_validate_nonexistent_output(self): - """Test validation of nonexistent output file.""" - validation = validate_build_output("/nonexistent/file.py") - - assert not validation["exists"] - assert "does not exist" in validation["errors"][0] - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_supported_languages(self): - """Test getting supported languages.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - get_supported_languages, - ) - - languages = get_supported_languages() - assert isinstance(languages, list) - assert len(languages) > 0 - - # Should include common languages - common_languages = ["python", "javascript", "java", "cpp", "c"] - for lang in common_languages: - if lang in languages: - assert lang in languages - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_create_build_manifest(self): - """Test build manifest creation.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - create_build_manifest, - ) - - build_config = { - "name": "test_build", - "version": "1.0.0", - "dependencies": ["pytest", "requests"], - "build_commands": [["python", "setup.py", "build"]], - } - - manifest = create_build_manifest(build_config) - - assert isinstance(manifest, dict) - assert "build_config" in manifest - assert "timestamp" in manifest - assert "manifest_version" in manifest - assert manifest["build_config"]["name"] == "test_build" - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_optimize_build_config(self): - """Test build configuration optimization.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - optimize_build_config, - ) - - config = { - "parallel_jobs": 1, - "cache_enabled": False, - "optimization_level": "none", - } - - optimized = optimize_build_config(config) - - assert isinstance(optimized, dict) - # Should have optimization recommendations - assert "parallel_jobs" in optimized - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_validate_build_dependencies(self): - """Test build dependency validation.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - validate_build_dependencies, - ) - - dependencies = ["os", "sys", "json"] # Built-in modules - result = validate_build_dependencies(dependencies) - - assert isinstance(result, dict) - assert "valid" in result - assert "missing" in result - assert "available" in result - - # Built-in modules should be available - assert result["valid"] or len(result["available"]) > 0 - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_parallel_build_execution(self): - """Test parallel build execution.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - parallel_build_execution, - ) - - build_tasks = [ - {"name": "task1", "command": ["echo", "task1"]}, - {"name": "task2", "command": ["echo", "task2"]}, - {"name": "task3", "command": ["echo", "task3"]}, - ] - - results = parallel_build_execution(build_tasks, max_workers=2) - - assert isinstance(results, list) - assert len(results) == len(build_tasks) - - for result in results: - assert "task" in result - assert "success" in result - assert "duration" in result - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_incremental_build_check(self, tmp_path): - """Test incremental build checking.""" - test_dir = str(tmp_path) - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - incremental_build_check, - ) - - # Create test files with different modification times - old_file = os.path.join(test_dir, "old.py") - new_file = os.path.join(test_dir, "new.py") - - with open(old_file, "w") as f: - f.write("# Old file") - time.sleep(0.1) # Ensure different timestamps - - with open(new_file, "w") as f: - f.write("# New file") - - source_files = [old_file, new_file] - build_cache = {} - - needs_build = incremental_build_check(source_files, build_cache) - - assert isinstance(needs_build, bool) - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_cleanup_build_artifacts(self, tmp_path): - """Test build artifact cleanup.""" - test_dir = str(tmp_path) - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - cleanup_build_artifacts, - ) - - # Create some test artifacts - artifacts = [] - for i in range(3): - artifact = os.path.join(test_dir, f"artifact_{i}.tmp") - with open(artifact, "w") as f: - f.write(f"Test artifact {i}") - artifacts.append(artifact) - - # Clean them up - result = cleanup_build_artifacts(artifacts) - - assert result - - # Verify they're gone - for artifact in artifacts: - assert not os.path.exists(artifact) - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_get_build_metrics(self): - """Test build metrics collection.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - get_build_metrics, - ) - - # Mock build results - build_results = { - "success": True, - "duration": 45.2, - "artifacts_created": 5, - "errors": 0, - "warnings": 2, - } - - metrics = get_build_metrics(build_results) - - assert isinstance(metrics, dict) - assert "build_success_rate" in metrics - assert "average_build_time" in metrics - assert "artifact_creation_rate" in metrics - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_export_build_report(self): - """Test build report export.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - export_build_report, - ) - - build_data = { - "build_id": "test_build_123", - "timestamp": time.time(), - "duration": 30.5, - "success": True, - "artifacts": ["app.exe", "lib.dll"], - } - - # Test JSON export - json_report = export_build_report(build_data, "json") - assert isinstance(json_report, str) - - parsed = json.loads(json_report) - assert parsed["build_id"] == "test_build_123" - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_import_build_config(self, tmp_path): - """Test build configuration import.""" - test_dir = str(tmp_path) - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - import_build_config, - ) - - # Create a test config file - config_file = os.path.join(test_dir, "build_config.json") - config_data = { - "name": "test_project", - "version": "1.0.0", - "build_commands": [["make", "all"]], - } - - with open(config_file, "w") as f: - json.dump(config_data, f) - - imported_config = import_build_config(config_file) - - assert isinstance(imported_config, dict) - assert imported_config["name"] == "test_project" - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_validate_build_config(self): - """Test build configuration validation.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - validate_build_config, - ) - - # Valid config - valid_config = { - "name": "test_build", - "build_commands": [["python", "setup.py", "build"]], - "artifacts": ["dist/app"], - } - - is_valid, errors = validate_build_config(valid_config) - assert is_valid - assert len(errors) == 0 - - # Invalid config - invalid_config = { - "name": "", # Empty name - "build_commands": [], # No commands - } - - is_valid, errors = validate_build_config(invalid_config) - assert not is_valid - assert len(errors) > 0 - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_get_build_history(self): - """Test build history retrieval.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - get_build_history, - ) - - history = get_build_history(limit=10) - - assert isinstance(history, list) - # History might be empty if no builds have been recorded - assert len(history) >= 0 - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_rollback_build(self): - """Test build rollback functionality.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import rollback_build - - # Test rollback to non-existent build (should fail gracefully) - result = rollback_build("nonexistent_build_id") - - assert isinstance(result, bool) - # Should return False for non-existent build - assert not result - - @pytest.mark.skipif( - not FULL_BUILD_AVAILABLE, reason="Full build synthesis not available" - ) - def test_monitor_build_progress(self): - """Test build progress monitoring.""" - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - monitor_build_progress, - ) - - # Mock build ID - build_id = "test_build_123" - - progress = monitor_build_progress(build_id) - - assert isinstance(progress, dict) - # Progress info should be available even for non-existent builds - assert "status" in progress - - def test_concurrent_build_operations(self): - """Test concurrent build operations.""" - if not FULL_BUILD_AVAILABLE: - pytest.skip("Full build synthesis not available") - - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - parallel_build_execution, - ) - - # Create multiple build tasks - tasks = [] - for i in range(5): - task = { - "name": f"build_task_{i}", - "command": ["echo", f"Building task {i}"], - } - tasks.append(task) - - results = parallel_build_execution(tasks, max_workers=3) - - assert len(results) == len(tasks) - - # All tasks should have completed - for result in results: - assert "success" in result - assert "duration" in result - - def test_build_error_handling(self): - """Test error handling in build operations.""" - # Test with invalid build configurations - invalid_configs = [ - {"name": None}, # Invalid name - {"build_commands": "not_a_list"}, # Invalid commands - {"artifacts": None}, # Invalid artifacts - ] - - for config in invalid_configs: - try: - results = orchestrate_build_pipeline(config) - # Should handle gracefully - assert isinstance(results, dict) - assert "overall_success" in results - except Exception: - # Some errors might be raised, that's acceptable - pass - - def test_build_resource_limits(self): - """Test build operations under resource constraints.""" - if not FULL_BUILD_AVAILABLE: - pytest.skip("Full build synthesis not available") - - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - parallel_build_execution, - ) - - # Test with very limited workers - tasks = [ - {"name": "task1", "command": ["echo", "task1"]}, - {"name": "task2", "command": ["echo", "task2"]}, - ] - - results = parallel_build_execution(tasks, max_workers=1) - - assert len(results) == len(tasks) - for result in results: - assert "success" in result - - def test_build_timeout_handling(self): - """Test timeout handling in build operations.""" - # Test with potentially slow operations - slow_config = { - "build_commands": [ - ["python", "-c", "import time; time.sleep(0.1); print('Slow command')"] - ], - "timeout": 1, # Short timeout - } - - results = orchestrate_build_pipeline(slow_config) - - assert isinstance(results, dict) - # Should complete within timeout - assert "overall_success" in results - - def test_build_artifact_validation_comprehensive(self, tmp_path): - """Comprehensive test of build artifact validation.""" - test_dir = str(tmp_path) - # Test various file types and validation scenarios - test_cases = [ - ( - "python_file", - "test.py", - "# Python file\ndef hello():\n print('hello')\n", - True, - ), - ("empty_file", "empty.py", "", False), - ("binary_file", "binary.bin", b"\x00\x01\x02", False), - ("text_file", "readme.txt", "This is a readme file.", False), - ] - - for _case_name, filename, content, should_be_valid in test_cases: - filepath = os.path.join(test_dir, filename) - - if isinstance(content, str): - with open(filepath, "w") as f: - f.write(content) - else: - with open(filepath, "wb") as f: - f.write(content) - - validation = validate_build_output(filepath) - - assert isinstance(validation, dict) - assert "exists" in validation - assert validation["exists"] # File should exist - - if should_be_valid: - assert validation["is_file"] - assert validation["size_bytes"] > 0 - - def test_build_configuration_edge_cases(self): - """Test build configuration edge cases.""" - edge_configs = [ - {}, # Empty config - {"name": "test"}, # Minimal config - {"name": "test", "build_commands": None}, # None commands - {"name": "test", "build_commands": [[]]}, # Empty commands - ] - - for config in edge_configs: - try: - results = orchestrate_build_pipeline(config) - assert isinstance(results, dict) - except Exception: - # Edge cases might raise exceptions, that's acceptable - pass - - def test_build_metrics_calculation(self): - """Test build metrics calculation.""" - if not FULL_BUILD_AVAILABLE: - pytest.skip("Full build synthesis not available") - - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - get_build_metrics, - ) - - # Test with various build result scenarios - test_results = [ - {"success": True, "duration": 30.5, "artifacts_created": 3, "errors": 0}, - {"success": False, "duration": 45.2, "artifacts_created": 0, "errors": 2}, - {"success": True, "duration": 15.1, "artifacts_created": 1, "errors": 0}, - ] - - for result in test_results: - metrics = get_build_metrics(result) - assert isinstance(metrics, dict) - assert "build_success_rate" in metrics - - def test_build_report_formats(self): - """Test different build report formats.""" - if not FULL_BUILD_AVAILABLE: - pytest.skip("Full build synthesis not available") - - from codomyrmex.ci_cd_automation.build.build_orchestrator import ( - export_build_report, - ) - - test_data = { - "build_id": "test_123", - "success": True, - "duration": 25.3, - "artifacts": ["app", "lib"], - } - - formats = ( - ["json", "xml", "yaml"] - if hasattr(export_build_report, "__code__") - and "format" in export_build_report.__code__.co_varnames - else ["json"] - ) - - for fmt in formats: - try: - report = export_build_report(test_data, fmt) - assert isinstance(report, str) - assert len(report) > 0 - except Exception: - # Some formats might not be supported - pass diff --git a/src/codomyrmex/tests/unit/cache/test_namespaced_cache.py b/src/codomyrmex/tests/unit/cache/test_namespaced_cache.py index 59a4c5ad3..1b843f1ae 100644 --- a/src/codomyrmex/tests/unit/cache/test_namespaced_cache.py +++ b/src/codomyrmex/tests/unit/cache/test_namespaced_cache.py @@ -40,7 +40,7 @@ def test_delete_pattern_prefixes_pattern(self, base_cache, ns_cache): ns_cache.set("user:1", "a") ns_cache.set("user:2", "b") ns_cache.set("other", "c") - + count = ns_cache.delete_pattern("user:*") assert count == 2 assert ns_cache.get("user:1") is None @@ -49,7 +49,7 @@ def test_delete_pattern_prefixes_pattern(self, base_cache, ns_cache): def test_clear_removes_only_namespace(self, base_cache, ns_cache): ns_cache.set("k1", "v1") base_cache.set("other:k2", "v2") - + ns_cache.clear() assert ns_cache.get("k1") is None assert base_cache.get("other:k2") == "v2" @@ -57,7 +57,7 @@ def test_clear_removes_only_namespace(self, base_cache, ns_cache): def test_nested_namespaces(self, base_cache): ns1 = NamespacedCache(base_cache, "outer") ns2 = NamespacedCache(ns1, "inner") - + ns2.set("k", "v") assert base_cache.get("outer:inner:k") == "v" assert ns2.get("k") == "v" @@ -65,13 +65,13 @@ def test_nested_namespaces(self, base_cache): def test_namespaced_file_based_cache(self, tmp_path): from codomyrmex.cache.backends.file_based import FileBasedCache - + fb_cache = FileBasedCache(cache_dir=tmp_path / "fb") ns_fb = NamespacedCache(fb_cache, "app") - + ns_fb.set("config", {"theme": "dark"}) assert ns_fb.get("config") == {"theme": "dark"} assert fb_cache.exists("app:config") is True - + ns_fb.delete_pattern("*") assert ns_fb.get("config") is None diff --git a/src/codomyrmex/tests/unit/coding/review/test_code_review_functional.py b/src/codomyrmex/tests/unit/coding/review/test_code_review_functional.py index 01811bb96..f6f431da9 100644 --- a/src/codomyrmex/tests/unit/coding/review/test_code_review_functional.py +++ b/src/codomyrmex/tests/unit/coding/review/test_code_review_functional.py @@ -45,7 +45,7 @@ def complex_func(a, b): if i % 2 == 0: print(i) return a + b - + def bare_except(): try: pass @@ -112,7 +112,7 @@ def test_pyscn_analyze_complex_code(self, python_file): analyzer = PyscnAnalyzer() except ToolNotFoundError: pytest.skip("pyscn not installed") - + with open(python_file, "r") as f: code = f.read() results = analyzer.analyze_code(code, python_file) diff --git a/src/codomyrmex/tests/unit/coding/test_refactoring.py b/src/codomyrmex/tests/unit/coding/test_refactoring.py index b25bc5495..875577683 100644 --- a/src/codomyrmex/tests/unit/coding/test_refactoring.py +++ b/src/codomyrmex/tests/unit/coding/test_refactoring.py @@ -21,7 +21,7 @@ class TestRenameRefactoring: def test_rename_variable(self, sample_file): refactor = RenameRefactoring(sample_file, "x", "y") result = refactor.execute() - + assert isinstance(result, RefactoringResult) assert result.success is True assert len(result.changes) == 2 @@ -31,7 +31,7 @@ def test_rename_variable(self, sample_file): def test_rename_function(self, sample_file): refactor = RenameRefactoring(sample_file, "foo", "bar") result = refactor.execute() - + assert result.success is True assert len(result.changes) == 2 assert all(c.new_text == "bar" for c in result.changes) diff --git a/src/codomyrmex/tests/unit/p3_remediation/test_p3_file_permissions.py b/src/codomyrmex/tests/unit/p3_remediation/test_p3_file_permissions.py deleted file mode 100644 index 41bd27dab..000000000 --- a/src/codomyrmex/tests/unit/p3_remediation/test_p3_file_permissions.py +++ /dev/null @@ -1,69 +0,0 @@ -""" -TDD regression tests for P3 CodeQL overly-permissive chmod remediation. - -Verifies that ``synthesize_build_artifact`` produces files with owner-only -permissions (0o700) instead of overly permissive (0o755). - -Zero-Mock compliant — uses real file operations. -""" - -import stat - -import pytest - -from codomyrmex.ci_cd_automation.build.pipeline.build_orchestrator import ( - synthesize_build_artifact, -) - - -@pytest.mark.unit -class TestBuildArtifactPermissions: - """Verify synthesized build artifacts have secure permissions.""" - - def test_synthesized_artifact_has_owner_only_permissions(self, tmp_path): - """Synthesized artifact must have mode 0o700 (owner rwx only).""" - source = tmp_path / "source.py" - source.write_text("print('hello')\n") - - output = tmp_path / "artifact.py" - result = synthesize_build_artifact(str(source), str(output)) - - assert result is True, "Build artifact synthesis should succeed" - assert output.exists(), "Output file must exist" - - # Extract permission bits (last 9 bits) - mode = output.stat().st_mode & 0o777 - assert mode == 0o700, ( - f"Artifact permissions should be 0o700 (owner-only), got {oct(mode)}" - ) - - def test_synthesized_artifact_not_world_executable(self, tmp_path): - """Synthesized artifact must NOT be world-executable.""" - source = tmp_path / "source.py" - source.write_text("x = 1\n") - - output = tmp_path / "artifact2.py" - synthesize_build_artifact(str(source), str(output)) - - mode = output.stat().st_mode - # Check that 'others' have NO permissions at all - assert not (mode & stat.S_IROTH), "Others should not have read permission" - assert not (mode & stat.S_IWOTH), "Others should not have write permission" - assert not (mode & stat.S_IXOTH), "Others should not have execute permission" - - def test_synthesized_artifact_not_group_executable(self, tmp_path): - """Synthesized artifact must NOT be group-readable/executable.""" - source = tmp_path / "source.py" - source.write_text("y = 2\n") - - output = tmp_path / "artifact3.py" - synthesize_build_artifact(str(source), str(output)) - - mode = output.stat().st_mode - assert not (mode & stat.S_IRGRP), "Group should not have read permission" - assert not (mode & stat.S_IWGRP), "Group should not have write permission" - assert not (mode & stat.S_IXGRP), "Group should not have execute permission" - - -if __name__ == "__main__": - pytest.main([__file__, "-v"])