diff --git a/docs/adr/0001-product-contract.md b/docs/adr/0001-product-contract.md index 34aa97a1..4ef2c8a8 100644 --- a/docs/adr/0001-product-contract.md +++ b/docs/adr/0001-product-contract.md @@ -139,13 +139,16 @@ collection is prohibited. ## Golden pipelines -Three executable acceptance specifications, all hermetic: +Four executable acceptance specifications. The first two are hermetic; the last +two are `live` and skip without a credential: 1. **`basic`** — deterministic local tools, sequential steps, template interpolation, typed outputs. -2. **`control-flow`** — conditional branching and parallel fan-out with - dependency ordering, plus a deliberately failing step to verify failure - propagation and exit codes. +2. **`control-flow`** — parallel fan-out and a dependent fan-in join, with + template interpolation across step results, plus a deliberately failing step + to verify failure propagation and exit codes. Conditional branching and + loops are *not* covered by this fixture and are not yet part of the + supported contract; see #333 (`on_false` / `on_success`) and #320. 3. **`live-anthropic`** — the same shape as `basic` but with one real Anthropic call. Marked `live`, skipped unless `ANTHROPIC_API_KEY` is set. 4. **`live-dartmouth`** — the same shape, against a free Dartmouth Chat model. diff --git a/docs/adr/0002-retire-pipeline-integration-infrastructure.md b/docs/adr/0002-retire-pipeline-integration-infrastructure.md new file mode 100644 index 00000000..c4b2fa5f --- /dev/null +++ b/docs/adr/0002-retire-pipeline-integration-infrastructure.md @@ -0,0 +1,113 @@ +# ADR 0002: Retire the pipeline integration testing subsystem + +**Status:** accepted +**Date:** 2026-08-01 +**Closes:** #435 +**Relates to:** #430 (retire the frozen competing implementations), #429 +(consolidate the two parallel model-adapter layers), #354 (legacy suite), +#241 / #104 (`run` and `validate` disagree) + +## Decision + +Delete, rather than repair: + +- `src/orchestrator/testing/pipeline_integration_infrastructure.py` (958 lines) +- `src/orchestrator/testing/pipeline_integration_demo.py` (544 lines) +- `tests/integration/test_pipeline_integration_infrastructure.py` (721 lines) + +Replace only the intent with six tests against the canonical compiler and +orchestrator, in `tests/test_pipeline_model_contracts.py`. + +## Why not repair it + +The module could not construct its own central model. Its constructor passed +four separate keyword arguments that no core dataclass defines — +`ModelCapabilities(max_output_tokens=)`, `ModelRequirements(network_access=)`, +`ModelMetrics(tokens_per_second=)`, `ModelCost(input_cost_per_token=)`. +Correcting one revealed the next. It was written against an API that has never +existed, and therefore had never run: **27 tests, permanently red**. + +Renaming those arguments was explicitly rejected. It would have made the first +layer pass while preserving a second testing architecture — its own model, +provider, validator, scoring system and orchestration facade — inside the +shipped package, with no user-facing consumer. That is the same +parallel-implementation problem #430 exists to remove, and repairing it would +have grown the surface that work has to retire. + +Supporting facts, all verified rather than assumed: + +- **No external consumers.** `rg` for the module names and for every symbol + they export (`PipelineTestModel`, `PipelineTestProvider`, + `PipelineIntegrationResult`, …) returns nothing outside the three files + themselves. `src/orchestrator/testing/__init__.py` does not import them. +- **It duplicated working infrastructure.** `tests/test_infrastructure.py` + already provides a deterministic model and provider. +- **It shipped pretend models.** It hardcoded fictional `openai/gpt-4` and + `anthropic/claude-*` entries into a package that end users install. +- **It contradicted project policy.** It was built around `mock_responses`, + while CONTRIBUTING states "No mock objects. Ever." + +Git history preserves the code. An archive copy inside the tree would add +noise without reducing risk. + +## What replaced it + +Six tests, hermetic, using the existing deterministic model: + +| Test | Pins | +|-|-| +| `test_a_valid_pipeline_compiles` | the canonical compiler accepts a well-formed pipeline | +| `test_a_malformed_pipeline_is_refused` | validation failure raises, not compiles-to-broken | +| `test_a_model_pipeline_executes` | a model step actually runs | +| `test_structured_output_pipeline_returns_an_object` | structured output is a mapping | +| `test_execution_failure_is_reported_not_swallowed` | a failing step reports failure | +| `test_a_model_pipeline_compiles` | **xfail(strict)** — documents #241 | + +Test-only models and providers stay under `tests/`. They are not product code +and must not be shipped again. + +## Defects this uncovered + +Deleting the parallel architecture forced the replacement tests through the +canonical path, which immediately exposed three real bugs that the retired +subsystem had been routing around: + +1. **`select_model()` results were looked up again.** Five call sites in + `hybrid_control_system.py` and `model_based_control_system.py` did + `model = get_model(await select_model(...))`. `select_model` already + returns a `Model`, so `get_model()`'s `":" in model_name` check raised + `TypeError: argument of type 'Model' is not a container`. Every + registry-selected `action: generate` step failed. +2. **`create_test_orchestrator()` built the wrong registry.** Two classes + share the name `ModelRegistry`; only `models.registry` has + `register_provider`, and only `models.model_registry` has + `can_provide_models`, which `Orchestrator.__init__` calls. The helper used + the first, so it raised on every call and could not construct an + orchestrator at all. (The duplicate-registry split itself is #429.) +3. **The deterministic model advertised the wrong task names.** It offered + `text-generation`, while the control system asks for `generate` and real + providers advertise `generate` — so it was ineligible for every generate + step and selection failed with `NoEligibleModelsError`. + +None of these were visible while a second testing stack stood in for the real +one. That is the strongest argument for the deletion: the parallel +architecture was not merely unused, it was hiding the state of the supported +path. + +## Consequences + +- 2,223 lines removed; 6 tests added, in the **blocking** layer rather than a + permanently red one. +- 27 tests were **retired, not fixed**. #354's backlog shrinks by that amount + for that reason, and must not be read as progress on the legacy suite. +- `validate` still rejects model pipelines that `run` accepts. That divergence + is now pinned by a strict xfail instead of being absorbed by a bespoke + validator, and closing #241 will make that test flip to a visible failure, + prompting its removal. + +## If systematic example-pipeline validation is wanted again + +Build it around the canonical `orchestrator validate` path so it returns the +same diagnostics as the CLI. It must not introduce another model, provider, +validator, scoring system, or orchestration facade — that is what was just +removed. diff --git a/src/orchestrator/control_systems/hybrid_control_system.py b/src/orchestrator/control_systems/hybrid_control_system.py index 6d4db4c2..2400f077 100644 --- a/src/orchestrator/control_systems/hybrid_control_system.py +++ b/src/orchestrator/control_systems/hybrid_control_system.py @@ -971,13 +971,13 @@ async def _handle_prompt_optimization_placeholder(self, task: Task, context: Dic goal = task.parameters.get("optimization_goal", "quality") # Use the model to enhance the prompt - model_name = await self.model_registry.select_model({"tasks": ["generate"]}) - if not model_name: + # select_model returns a Model, so there is nothing further to look up. + model = await self.model_registry.select_model({"tasks": ["generate"]}) + if not model: return { "success": False, "error": "No model available for prompt optimization" } - model = self.model_registry.get_model(model_name) optimization_prompt = f"""Optimize this prompt for {task_type}: Original prompt: "{prompt}" @@ -1137,9 +1137,9 @@ async def _handle_task_delegation(self, task: Task, context: Dict[str, Any]) -> if match: # Use a model to resolve the AUTO tag auto_prompt = match.group(1) - model_name = await self.model_registry.select_model({"tasks": ["generate"]}) - if model_name: - model = self.model_registry.get_model(model_name) + # select_model returns a Model already. + model = await self.model_registry.select_model({"tasks": ["generate"]}) + if model: response = await model.generate( f"{auto_prompt}\nBased on the task: '{params.get('task', '')}'\nRespond with only: simple, moderate, or complex" ) @@ -1208,8 +1208,8 @@ async def _handle_analyze_text(self, task: Task, context: Dict[str, Any]) -> Any "tasks": ["analyze", "generate"], "context_window": len(prompt.encode()) // 4 # Rough token estimate } - model_name = await self.model_registry.select_model(requirements) - model = self.model_registry.get_model(model_name) + # select_model returns a Model already. + model = await self.model_registry.select_model(requirements) else: # Get specific model model = self.model_registry.get_model(model_spec) @@ -1318,13 +1318,16 @@ async def _handle_generate_text(self, task: Task, context: Dict[str, Any]) -> An elif isinstance(model_spec, str) and model_spec.startswith(" Any: # If still no model, select one based on task requirements if not model: requirements = self._get_task_requirements(task) - model_name = await self.model_registry.select_model(requirements) - if model_name: - model = await self.model_registry.get_model(model_name) + # select_model returns a Model, not a name -- passing its result + # back into get_model() raised TypeError from get_model()'s + # `":" in model_name` check. + model = await self.model_registry.select_model(requirements) # Extract the actual action/prompt from the task if task.action in ["generate", "generate_text"] and task.parameters.get( @@ -228,11 +238,11 @@ async def _execute_task_impl(self, task: Task, context: Dict[str, Any]) -> Any: try: # Check if this is a structured generation task - if task.action == "generate-structured": + if task.action in STRUCTURED_ACTIONS: # Use structured generation if not task.parameters or "schema" not in task.parameters: raise ValueError( - f"Task '{task.id}' with action 'generate-structured' requires a 'schema' parameter" + f"Task '{task.id}' with action '{task.action}' requires a 'schema' parameter" ) temperature = ( @@ -442,8 +452,8 @@ def _build_prompt( def _parse_result(self, result: Any, task: Task) -> Any: """Parse model result based on expected format.""" - # For generate-structured actions, the result should already be a structured object - if task.action == "generate-structured": + # For structured actions, the result should already be a structured object + if task.action in STRUCTURED_ACTIONS: # If it's still a string, try to parse it as JSON if isinstance(result, str): try: diff --git a/src/orchestrator/testing/pipeline_integration_demo.py b/src/orchestrator/testing/pipeline_integration_demo.py deleted file mode 100644 index faea74e8..00000000 --- a/src/orchestrator/testing/pipeline_integration_demo.py +++ /dev/null @@ -1,545 +0,0 @@ -#!/usr/bin/env python3 -"""Pipeline Integration Infrastructure Demonstration Script. - -This script demonstrates how to use the new PipelineTestModel/PipelineTestProvider -infrastructure for systematic pipeline validation, applying the proven patterns -from the successful testing epic (#374) to pipeline validation. - -Usage: - python -m src.orchestrator.testing.pipeline_integration_demo [options] - -Examples: - # Validate all example pipelines - python -m src.orchestrator.testing.pipeline_integration_demo --validate-all - - # Validate specific pipeline - python -m src.orchestrator.testing.pipeline_integration_demo --pipeline simple_data_processing - - # Run integration tests - python -m src.orchestrator.testing.pipeline_integration_demo --test-infrastructure - - # Generate performance report - python -m src.orchestrator.testing.pipeline_integration_demo --performance-report -""" - -import asyncio -import argparse -import json -import logging -import sys -import time -from pathlib import Path -from typing import Any, Dict, List, Optional - -# Setup logging -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') -logger = logging.getLogger(__name__) - -try: - from .pipeline_integration_infrastructure import ( - PipelineTestModel, - PipelineTestProvider, - PipelineIntegrationValidator, - create_pipeline_test_orchestrator, - create_pipeline_integration_validator - ) - from ..orchestrator import Orchestrator - from ..models.registry import ModelRegistry -except ImportError as e: - logger.error(f"Import error: {e}") - logger.error("Make sure you're running this from the orchestrator root directory") - sys.exit(1) - - -class PipelineIntegrationDemo: - """Demonstration class for pipeline integration infrastructure.""" - - def __init__(self, examples_dir: Optional[Path] = None): - """Initialize the demonstration.""" - self.examples_dir = examples_dir or Path("examples") - self.validator = None - self.results_dir = Path("outputs/integration_demo") - self.results_dir.mkdir(parents=True, exist_ok=True) - - logger.info(f"Pipeline Integration Demo initialized") - logger.info(f"Examples directory: {self.examples_dir}") - logger.info(f"Results directory: {self.results_dir}") - - def demonstrate_test_model_capabilities(self) -> Dict[str, Any]: - """Demonstrate PipelineTestModel capabilities.""" - logger.info("=== Demonstrating PipelineTestModel Capabilities ===") - - results = {} - - try: - # Create test model with custom configuration - model = PipelineTestModel( - name="demo-pipeline-model", - mock_responses={ - "demo_pipeline": "This is a custom response for the demo pipeline", - "quality_test": "Pipeline quality is excellent with score 98/100" - }, - pipeline_validation_enabled=True - ) - - logger.info(f"Created PipelineTestModel: {model.name}") - logger.info(f"Capabilities: {len(model.capabilities.supported_tasks)} supported tasks") - logger.info(f"Context window: {model.capabilities.context_window}") - logger.info(f"Validation enabled: {model.pipeline_validation_enabled}") - - results["model_created"] = True - results["model_name"] = model.name - results["supported_tasks"] = model.capabilities.supported_tasks - results["context_window"] = model.capabilities.context_window - - # Test various model capabilities - asyncio.run(self._test_model_features(model, results)) - - except Exception as e: - logger.error(f"Error demonstrating test model: {e}") - results["error"] = str(e) - - return results - - async def _test_model_features(self, model: PipelineTestModel, results: Dict[str, Any]): - """Test various model features.""" - - # Test health check - health = await model.health_check() - logger.info(f"Health check: {'PASS' if health else 'FAIL'}") - results["health_check"] = health - - # Test text generation - response = await model.generate( - "Validate this pipeline for quality and correctness", - pipeline_context={'pipeline_name': 'demo_pipeline'} - ) - logger.info(f"Generated response: {response[:100]}...") - results["text_generation"] = len(response) > 0 - - # Test structured output - schema = { - "type": "object", - "properties": { - "validation_status": {"type": "string"}, - "quality_score": {"type": "number"}, - "issues": {"type": "array"}, - "recommendations": {"type": "array"} - } - } - - structured_response = await model.generate_structured( - "Generate comprehensive pipeline validation report", - schema, - pipeline_context={'pipeline_name': 'demo_pipeline'} - ) - - logger.info(f"Structured response keys: {list(structured_response.keys())}") - results["structured_output"] = isinstance(structured_response, dict) - results["structured_response_sample"] = structured_response - - # Test cost estimation - cost = await model.estimate_cost(200, 100) - logger.info(f"Cost estimate: ${cost}") - results["cost_estimation"] = isinstance(cost, (int, float)) - - # Get validation summary - summary = model.get_validation_summary() - logger.info(f"Validation summary: {summary}") - results["validation_summary"] = summary - - def demonstrate_test_provider_capabilities(self) -> Dict[str, Any]: - """Demonstrate PipelineTestProvider capabilities.""" - logger.info("=== Demonstrating PipelineTestProvider Capabilities ===") - - results = {} - - try: - # Create test provider - provider = PipelineTestProvider("demo-provider") - - logger.info(f"Created PipelineTestProvider: {provider.name}") - logger.info(f"Available models: {len(provider.available_models)}") - logger.info(f"Provider info: {provider.get_provider_info()}") - - results["provider_created"] = True - results["provider_name"] = provider.name - results["available_models"] = provider.available_models - results["provider_info"] = provider.get_provider_info() - - # Test provider features - asyncio.run(self._test_provider_features(provider, results)) - - except Exception as e: - logger.error(f"Error demonstrating test provider: {e}") - results["error"] = str(e) - - return results - - async def _test_provider_features(self, provider: PipelineTestProvider, results: Dict[str, Any]): - """Test various provider features.""" - - # Test model support - test_models = ["pipeline-test-model", "openai/gpt-4", "nonexistent-model"] - support_results = {} - - for model_name in test_models: - supported = provider.supports_model(model_name) - support_results[model_name] = supported - logger.info(f"Model support '{model_name}': {'YES' if supported else 'NO'}") - - results["model_support"] = support_results - - # Test getting model capabilities - try: - capabilities = provider.get_model_capabilities("pipeline-test-model") - logger.info(f"Model capabilities retrieved successfully") - results["capabilities_retrieval"] = True - except Exception as e: - logger.error(f"Failed to get capabilities: {e}") - results["capabilities_retrieval"] = False - - # Test model retrieval and usage tracking - initial_stats = provider.get_usage_statistics() - logger.info(f"Initial usage stats: {initial_stats}") - - # Get a model and use it - model = await provider.get_model("pipeline-test-model") - await model.generate("Test usage tracking") - - updated_stats = provider.get_usage_statistics() - logger.info(f"Updated usage stats: {updated_stats}") - - results["usage_tracking"] = updated_stats['total_requests'] > initial_stats['total_requests'] - - # Test provider initialization - await provider.initialize() - results["initialization"] = provider.is_initialized - - def demonstrate_orchestrator_integration(self) -> Dict[str, Any]: - """Demonstrate integration with orchestrator framework.""" - logger.info("=== Demonstrating Orchestrator Integration ===") - - results = {} - - try: - # Create test orchestrator using utility function - orchestrator = create_pipeline_test_orchestrator() - - logger.info(f"Created test orchestrator successfully") - logger.info(f"Model registry: {orchestrator.model_registry is not None}") - logger.info(f"Control system: {orchestrator.control_system is not None}") - - results["orchestrator_created"] = True - - # Verify test provider is registered - providers = orchestrator.model_registry.get_registered_providers() - test_provider_found = any(p.name == "pipeline-test-provider" for p in providers) - - logger.info(f"Test provider registered: {'YES' if test_provider_found else 'NO'}") - logger.info(f"Total providers: {len(providers)}") - - results["test_provider_registered"] = test_provider_found - results["total_providers"] = len(providers) - - # Test model availability through registry - try: - # This would typically require the orchestrator to be fully set up - logger.info("Orchestrator integration test completed") - results["orchestrator_functional"] = True - except Exception as e: - logger.warning(f"Orchestrator functionality test failed: {e}") - results["orchestrator_functional"] = False - results["orchestrator_error"] = str(e) - - except Exception as e: - logger.error(f"Error in orchestrator integration: {e}") - results["error"] = str(e) - - return results - - async def demonstrate_integration_validator(self, pipeline_name: Optional[str] = None) -> Dict[str, Any]: - """Demonstrate PipelineIntegrationValidator capabilities.""" - logger.info("=== Demonstrating Integration Validator ===") - - results = {} - - try: - # Create integration validator - self.validator = create_pipeline_integration_validator(self.examples_dir) - - logger.info(f"Created PipelineIntegrationValidator") - logger.info(f"Examples directory: {self.validator.examples_dir}") - logger.info(f"Test provider: {self.validator.test_provider.name}") - - results["validator_created"] = True - - if pipeline_name: - # Validate specific pipeline - pipeline_path = self.examples_dir / f"{pipeline_name}.yaml" - if pipeline_path.exists(): - logger.info(f"Validating specific pipeline: {pipeline_name}") - result = await self.validator.validate_pipeline_integration(pipeline_name, pipeline_path) - - logger.info(f"Validation result:") - logger.info(f" - Pipeline: {result.pipeline_name}") - logger.info(f" - Integration score: {result.integration_score:.1f}") - logger.info(f" - Validation passed: {result.validation_passed}") - logger.info(f" - Execution successful: {result.execution_successful}") - logger.info(f" - Execution time: {result.execution_time:.2f}s") - logger.info(f" - Issues: {len(result.issues)}") - logger.info(f" - Recommendations: {len(result.recommendations)}") - - results["single_pipeline_validation"] = { - "pipeline_name": result.pipeline_name, - "integration_score": result.integration_score, - "validation_passed": result.validation_passed, - "execution_successful": result.execution_successful, - "execution_time": result.execution_time, - "issues": result.issues, - "recommendations": result.recommendations[:3] # First 3 recommendations - } - else: - logger.warning(f"Pipeline file not found: {pipeline_path}") - results["single_pipeline_validation"] = {"error": "Pipeline file not found"} - else: - # Validate all pipelines (limited for demo) - logger.info("Validating all example pipelines...") - - # For demo, limit to first few pipelines to avoid long execution - yaml_files = list(self.examples_dir.glob("*.yaml"))[:3] # First 3 pipelines - logger.info(f"Found {len(yaml_files)} pipeline files (limited to 3 for demo)") - - validation_results = {} - - for pipeline_path in yaml_files: - pipeline_name = pipeline_path.stem - try: - result = await self.validator.validate_pipeline_integration(pipeline_name, pipeline_path) - validation_results[pipeline_name] = { - "integration_score": result.integration_score, - "validation_passed": result.validation_passed, - "execution_successful": result.execution_successful, - "issues_count": len(result.issues) - } - logger.info(f" {pipeline_name}: score {result.integration_score:.1f}") - except Exception as e: - logger.warning(f"Failed to validate {pipeline_name}: {e}") - validation_results[pipeline_name] = {"error": str(e)} - - results["all_pipelines_validation"] = validation_results - - # Get integration summary - summary = self.validator.get_integration_summary() - logger.info(f"Integration summary:") - logger.info(f" - Total validations: {summary.get('total_validations', 0)}") - logger.info(f" - Success rate: {summary.get('validation_success_rate', 0):.1f}%") - logger.info(f" - Average score: {summary.get('average_integration_score', 0):.1f}") - - results["integration_summary"] = summary - - except Exception as e: - logger.error(f"Error in integration validator demo: {e}") - results["error"] = str(e) - - return results - - def run_infrastructure_tests(self) -> Dict[str, Any]: - """Run comprehensive infrastructure tests.""" - logger.info("=== Running Infrastructure Tests ===") - - results = { - "test_model_capabilities": self.demonstrate_test_model_capabilities(), - "test_provider_capabilities": self.demonstrate_test_provider_capabilities(), - "orchestrator_integration": self.demonstrate_orchestrator_integration() - } - - # Calculate overall test success - total_tests = 0 - passed_tests = 0 - - for category, category_results in results.items(): - if isinstance(category_results, dict): - for key, value in category_results.items(): - if isinstance(value, bool): - total_tests += 1 - if value: - passed_tests += 1 - - success_rate = (passed_tests / max(1, total_tests)) * 100 - logger.info(f"Infrastructure tests completed: {passed_tests}/{total_tests} passed ({success_rate:.1f}%)") - - results["test_summary"] = { - "total_tests": total_tests, - "passed_tests": passed_tests, - "success_rate": success_rate - } - - return results - - def generate_performance_report(self) -> Dict[str, Any]: - """Generate performance analysis report.""" - logger.info("=== Generating Performance Report ===") - - results = {} - - try: - if not self.validator: - self.validator = create_pipeline_integration_validator(self.examples_dir) - - # Get provider statistics - provider_stats = self.validator.test_provider.get_usage_statistics() - results["provider_statistics"] = provider_stats - - # Get integration summary if available - if self.validator.integration_results: - summary = self.validator.get_integration_summary() - results["integration_summary"] = summary - - # Calculate performance metrics - execution_times = [r.execution_time for r in self.validator.integration_results if r.execution_time > 0] - if execution_times: - results["performance_metrics"] = { - "average_execution_time": sum(execution_times) / len(execution_times), - "min_execution_time": min(execution_times), - "max_execution_time": max(execution_times), - "total_pipelines_tested": len(execution_times) - } - - logger.info(f"Performance report generated successfully") - - except Exception as e: - logger.error(f"Error generating performance report: {e}") - results["error"] = str(e) - - return results - - def save_results(self, results: Dict[str, Any], filename: str): - """Save results to JSON file.""" - try: - output_file = self.results_dir / f"{filename}_{int(time.time())}.json" - with open(output_file, 'w') as f: - json.dump(results, f, indent=2, default=str) - logger.info(f"Results saved to: {output_file}") - except Exception as e: - logger.error(f"Failed to save results: {e}") - - -async def main(): - """Main demonstration function.""" - parser = argparse.ArgumentParser(description="Pipeline Integration Infrastructure Demo") - parser.add_argument("--examples-dir", type=Path, help="Path to examples directory") - parser.add_argument("--validate-all", action="store_true", help="Validate all example pipelines") - parser.add_argument("--pipeline", type=str, help="Validate specific pipeline") - parser.add_argument("--test-infrastructure", action="store_true", help="Run infrastructure tests") - parser.add_argument("--performance-report", action="store_true", help="Generate performance report") - parser.add_argument("--save-results", action="store_true", help="Save results to file") - - args = parser.parse_args() - - # Create demo instance - demo = PipelineIntegrationDemo(args.examples_dir) - - if args.test_infrastructure: - logger.info("Running infrastructure tests...") - results = demo.run_infrastructure_tests() - - if args.save_results: - demo.save_results(results, "infrastructure_tests") - - # Print summary - summary = results.get("test_summary", {}) - print(f"\n{'='*60}") - print(f"INFRASTRUCTURE TEST RESULTS") - print(f"{'='*60}") - print(f"Total tests: {summary.get('total_tests', 0)}") - print(f"Passed: {summary.get('passed_tests', 0)}") - print(f"Success rate: {summary.get('success_rate', 0):.1f}%") - - elif args.validate_all: - logger.info("Running validation on all pipelines...") - results = await demo.demonstrate_integration_validator() - - if args.save_results: - demo.save_results(results, "pipeline_validation") - - # Print summary - if "integration_summary" in results: - summary = results["integration_summary"] - print(f"\n{'='*60}") - print(f"PIPELINE VALIDATION RESULTS") - print(f"{'='*60}") - print(f"Total pipelines: {summary.get('total_validations', 0)}") - print(f"Validation success rate: {summary.get('validation_success_rate', 0):.1f}%") - print(f"Average integration score: {summary.get('average_integration_score', 0):.1f}") - - elif args.pipeline: - logger.info(f"Running validation on pipeline: {args.pipeline}") - results = await demo.demonstrate_integration_validator(args.pipeline) - - if args.save_results: - demo.save_results(results, f"pipeline_{args.pipeline}") - - # Print summary - if "single_pipeline_validation" in results: - result = results["single_pipeline_validation"] - print(f"\n{'='*60}") - print(f"PIPELINE VALIDATION: {args.pipeline}") - print(f"{'='*60}") - print(f"Integration score: {result.get('integration_score', 0):.1f}") - print(f"Validation passed: {result.get('validation_passed', False)}") - print(f"Execution successful: {result.get('execution_successful', False)}") - print(f"Issues: {len(result.get('issues', []))}") - print(f"Recommendations: {len(result.get('recommendations', []))}") - - if result.get('recommendations'): - print(f"\nTop recommendations:") - for i, rec in enumerate(result['recommendations'][:3], 1): - print(f" {i}. {rec}") - - elif args.performance_report: - logger.info("Generating performance report...") - results = demo.generate_performance_report() - - if args.save_results: - demo.save_results(results, "performance_report") - - print(f"\n{'='*60}") - print(f"PERFORMANCE REPORT") - print(f"{'='*60}") - - if "provider_statistics" in results: - stats = results["provider_statistics"] - print(f"Provider requests: {stats.get('total_requests', 0)}") - print(f"Available models: {stats.get('total_models', 0)}") - - if "performance_metrics" in results: - metrics = results["performance_metrics"] - print(f"Average execution time: {metrics.get('average_execution_time', 0):.2f}s") - print(f"Pipelines tested: {metrics.get('total_pipelines_tested', 0)}") - - else: - # Run full demo - logger.info("Running full pipeline integration demo...") - - print(f"\n{'='*60}") - print(f"PIPELINE INTEGRATION INFRASTRUCTURE DEMO") - print(f"{'='*60}") - - # Run infrastructure tests - infra_results = demo.run_infrastructure_tests() - infra_summary = infra_results.get("test_summary", {}) - print(f"Infrastructure tests: {infra_summary.get('passed_tests', 0)}/{infra_summary.get('total_tests', 0)} passed") - - # Run pipeline validation - validation_results = await demo.demonstrate_integration_validator() - if "integration_summary" in validation_results: - val_summary = validation_results["integration_summary"] - print(f"Pipeline validations: {val_summary.get('total_validations', 0)} pipelines tested") - print(f"Average score: {val_summary.get('average_integration_score', 0):.1f}") - - print(f"\nDemo completed successfully!") - - -if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file diff --git a/src/orchestrator/testing/pipeline_integration_infrastructure.py b/src/orchestrator/testing/pipeline_integration_infrastructure.py deleted file mode 100644 index fa8af0fa..00000000 --- a/src/orchestrator/testing/pipeline_integration_infrastructure.py +++ /dev/null @@ -1,959 +0,0 @@ -"""Pipeline integration infrastructure using proven TestModel/TestProvider patterns. - -This module extends the successful test infrastructure from Issue #374 to provide -systematic pipeline validation capabilities for the validate-all-example-pipelines epic. - -Key Features: -- PipelineTestModel extending proven TestModel patterns -- PipelineTestProvider extending MockTestProvider patterns -- Integration with existing PipelineTestSuite infrastructure -- Systematic validation using orchestrator framework patterns -""" - -from __future__ import annotations - -import asyncio -import json -import logging -import tempfile -import time -from pathlib import Path -from typing import Any, Dict, List, Optional, Set, Tuple, Union -from dataclasses import dataclass, field - -from ..core.model import Model, ModelCapabilities, ModelRequirements, ModelMetrics, ModelCost -from ..orchestrator import Orchestrator -from ..models.registry import ModelRegistry -from .pipeline_test_suite import PipelineTestSuite, PipelineTestResult, ExecutionResult -from .pipeline_validator import PipelineValidator - -logger = logging.getLogger(__name__) - - -@dataclass -class PipelineIntegrationResult: - """Result of pipeline integration validation.""" - - pipeline_name: str - validation_passed: bool - execution_successful: bool - integration_score: float - - # Detailed results - test_model_performance: Dict[str, Any] = field(default_factory=dict) - provider_integration_status: Dict[str, Any] = field(default_factory=dict) - orchestrator_compatibility: Dict[str, Any] = field(default_factory=dict) - - # Issues and recommendations - issues: List[str] = field(default_factory=list) - warnings: List[str] = field(default_factory=list) - recommendations: List[str] = field(default_factory=list) - - # Execution metadata - execution_time: float = 0.0 - memory_usage_mb: float = 0.0 - estimated_cost: float = 0.0 - - -class PipelineTestModel(Model): - """Extended TestModel specifically designed for pipeline validation. - - This class extends the proven TestModel patterns from the successful - testing infrastructure to provide pipeline-specific validation capabilities. - """ - - def __init__( - self, - name: str = "pipeline-test-model", - provider: str = "pipeline-test-provider", - capabilities: Optional[ModelCapabilities] = None, - requirements: Optional[ModelRequirements] = None, - metrics: Optional[ModelMetrics] = None, - cost: Optional[ModelCost] = None, - pipeline_validation_enabled: bool = True, - mock_responses: Optional[Dict[str, str]] = None - ) -> None: - """Initialize pipeline test model with enhanced validation capabilities.""" - - # Enhanced capabilities for pipeline testing - if capabilities is None: - capabilities = ModelCapabilities( - supported_tasks=[ - "text-generation", "analysis", "validation", "structured-output", - "pipeline-execution", "template-resolution", "quality-assessment" - ], - context_window=32768, # Larger context for pipeline processing - supports_function_calling=True, - supports_structured_output=True, - supports_streaming=False, # Simplified for testing - max_output_tokens=4096 - ) - - if requirements is None: - requirements = ModelRequirements( - memory_gb=0.2, # Slightly more memory for pipeline processing - cpu_cores=1, - gpu_memory_gb=0, # CPU-only for testing - network_access=True # May need network for validation - ) - - if metrics is None: - metrics = ModelMetrics( - tokens_per_second=50.0, - quality_score=0.95, - reliability_score=0.99, - latency_p50_ms=100.0, - latency_p95_ms=250.0 - ) - - if cost is None: - cost = ModelCost( - is_free=True, - input_cost_per_token=0.0, - output_cost_per_token=0.0, - fixed_cost_per_request=0.0 - ) - - super().__init__(name, provider, capabilities, requirements, metrics, cost) - - # Pipeline-specific configuration - self.pipeline_validation_enabled = pipeline_validation_enabled - self.mock_responses = mock_responses or {} - self.execution_count = 0 - self.validation_history: List[Dict[str, Any]] = [] - - async def generate( - self, - prompt: str, - max_tokens: Optional[int] = None, - temperature: float = 0.7, - **kwargs: Any - ) -> str: - """Generate pipeline-aware text response.""" - - self.execution_count += 1 - - # Check for pipeline-specific mock responses - pipeline_context = kwargs.get('pipeline_context', {}) - pipeline_name = pipeline_context.get('pipeline_name', 'unknown') - - if pipeline_name in self.mock_responses: - response = self.mock_responses[pipeline_name] - else: - # Generate contextual response based on prompt content - if "validation" in prompt.lower(): - response = f"Pipeline validation response for {pipeline_name}: All validations passed successfully." - elif "analysis" in prompt.lower(): - response = f"Pipeline analysis for {pipeline_name}: Structure is valid, templates resolved correctly." - elif "quality" in prompt.lower(): - response = f"Quality assessment for {pipeline_name}: High quality output with score 95/100." - else: - response = f"Pipeline test response for {pipeline_name}: {prompt[:100]}..." - - # Record validation attempt - validation_record = { - 'timestamp': time.time(), - 'pipeline_name': pipeline_name, - 'prompt_type': self._classify_prompt(prompt), - 'response_generated': True, - 'execution_count': self.execution_count - } - self.validation_history.append(validation_record) - - return response - - async def generate_structured( - self, - prompt: str, - schema: Dict[str, Any], - max_tokens: Optional[int] = None, - temperature: float = 0.7, - **kwargs: Any - ) -> Dict[str, Any]: - """Generate structured response for pipeline validation.""" - - pipeline_context = kwargs.get('pipeline_context', {}) - pipeline_name = pipeline_context.get('pipeline_name', 'unknown') - - # Generate structured validation result - if schema.get('type') == 'object': - properties = schema.get('properties', {}) - - # Build structured response based on schema - response = {} - - if 'validation_status' in properties: - response['validation_status'] = 'passed' - - if 'quality_score' in properties: - response['quality_score'] = 95.0 - - if 'issues' in properties: - response['issues'] = [] - - if 'recommendations' in properties: - response['recommendations'] = [ - f"Pipeline {pipeline_name} is well-structured", - "Consider adding error handling for robustness" - ] - - if 'execution_metadata' in properties: - response['execution_metadata'] = { - 'execution_time': 0.5, - 'tokens_used': 150, - 'api_calls': 1, - 'estimated_cost': 0.001 - } - - # Fill any remaining properties with defaults - for prop_name, prop_schema in properties.items(): - if prop_name not in response: - if prop_schema.get('type') == 'string': - response[prop_name] = f"Test value for {prop_name}" - elif prop_schema.get('type') == 'number': - response[prop_name] = 1.0 - elif prop_schema.get('type') == 'boolean': - response[prop_name] = True - elif prop_schema.get('type') == 'array': - response[prop_name] = [] - elif prop_schema.get('type') == 'object': - response[prop_name] = {} - - return response - - # Fallback structured response - return { - "pipeline_validation": True, - "test_output": f"Structured response for pipeline {pipeline_name}", - "schema_validated": True - } - - async def health_check(self) -> bool: - """Enhanced health check for pipeline testing.""" - - # Basic health check - if not super().health_check(): - return False - - # Pipeline-specific health checks - if self.pipeline_validation_enabled: - # Check if we can perform basic validation operations - try: - test_validation = await self.generate("validation test", pipeline_context={'pipeline_name': 'health_check'}) - return "validation" in test_validation.lower() - except Exception: - return False - - return True - - async def estimate_cost( - self, - input_tokens: int, - output_tokens: int - ) -> float: - """Pipeline-aware cost estimation.""" - - # Always free for testing, but track usage - usage_record = { - 'input_tokens': input_tokens, - 'output_tokens': output_tokens, - 'estimated_cost': 0.0, - 'timestamp': time.time() - } - - # Add to validation history for tracking - self.validation_history.append({ - 'type': 'cost_estimation', - 'usage': usage_record - }) - - return 0.0 - - def _classify_prompt(self, prompt: str) -> str: - """Classify the type of prompt for better response generation.""" - prompt_lower = prompt.lower() - - if any(word in prompt_lower for word in ['validate', 'validation', 'verify']): - return 'validation' - elif any(word in prompt_lower for word in ['analyze', 'analysis', 'assess']): - return 'analysis' - elif any(word in prompt_lower for word in ['quality', 'score', 'rating']): - return 'quality' - elif any(word in prompt_lower for word in ['template', 'resolve', 'render']): - return 'template' - else: - return 'general' - - def get_validation_summary(self) -> Dict[str, Any]: - """Get summary of validation activities.""" - - return { - 'total_executions': self.execution_count, - 'validation_attempts': len([v for v in self.validation_history if v.get('type') != 'cost_estimation']), - 'unique_pipelines': len(set(v.get('pipeline_name', 'unknown') for v in self.validation_history if 'pipeline_name' in v)), - 'prompt_types': { - prompt_type: len([v for v in self.validation_history if v.get('prompt_type') == prompt_type]) - for prompt_type in set(v.get('prompt_type') for v in self.validation_history if 'prompt_type' in v) - }, - 'average_executions_per_pipeline': self.execution_count / max(1, len(set(v.get('pipeline_name', 'unknown') for v in self.validation_history if 'pipeline_name' in v))) - } - - -class PipelineTestProvider: - """Enhanced TestProvider specifically designed for pipeline integration testing. - - This class extends the proven MockTestProvider patterns to provide - comprehensive pipeline validation capabilities. - """ - - def __init__(self, name: str = "pipeline-test-provider"): - """Initialize pipeline test provider with enhanced capabilities.""" - - self.name = name - self.is_initialized = True - - # Create enhanced test models for different validation scenarios - self._models = { - # Core pipeline test model - "pipeline-test-model": PipelineTestModel(), - - # Specialized models for different validation types - "pipeline-validation-model": PipelineTestModel( - name="pipeline-validation-model", - pipeline_validation_enabled=True - ), - - "pipeline-quality-model": PipelineTestModel( - name="pipeline-quality-model", - mock_responses={ - "quality_assessment": "Excellent pipeline quality with comprehensive validation", - "template_validation": "All templates resolved successfully" - } - ), - - # Common model aliases that tests might expect - "openai/gpt-3.5-turbo": PipelineTestModel(name="gpt-3.5-turbo", provider="openai"), - "openai/gpt-4": PipelineTestModel(name="gpt-4", provider="openai"), - "anthropic/claude-3": PipelineTestModel(name="claude-3", provider="anthropic"), - "anthropic/claude-sonnet-4-20250514": PipelineTestModel(name="claude-sonnet-4", provider="anthropic"), - } - - # Pipeline validation configuration - self.validation_enabled = True - self.integration_tracking = True - self.execution_history: List[Dict[str, Any]] = [] - - # Performance tracking - self.model_usage_stats: Dict[str, Dict[str, Any]] = {} - - @property - def available_models(self) -> List[str]: - """List all available models including pipeline-specific ones.""" - return list(self._models.keys()) - - def supports_model(self, model_name: str) -> bool: - """Check if provider supports model with enhanced pipeline validation.""" - return model_name in self._models - - def get_model_capabilities(self, model_name: str) -> ModelCapabilities: - """Get enhanced capabilities for pipeline validation.""" - if not self.supports_model(model_name): - raise ValueError(f"Model '{model_name}' not supported") - - model = self._models[model_name] - return model.capabilities - - def get_model_requirements(self, model_name: str) -> ModelRequirements: - """Get model requirements with pipeline-specific considerations.""" - if not self.supports_model(model_name): - raise ValueError(f"Model '{model_name}' not supported") - - model = self._models[model_name] - return model.requirements - - def get_model_cost(self, model_name: str) -> ModelCost: - """Get cost information for pipeline testing.""" - if not self.supports_model(model_name): - raise ValueError(f"Model '{model_name}' not supported") - - model = self._models[model_name] - return model.cost - - def get_provider_info(self) -> Dict[str, Any]: - """Get enhanced provider information with pipeline capabilities.""" - return { - "name": self.name, - "type": "pipeline-test", - "models": len(self._models), - "initialized": self.is_initialized, - "validation_enabled": self.validation_enabled, - "integration_tracking": self.integration_tracking, - "supported_pipeline_features": [ - "template-resolution", - "quality-assessment", - "execution-validation", - "structured-output", - "cost-estimation" - ] - } - - async def get_model(self, model_name: str, **kwargs) -> PipelineTestModel: - """Get model instance with pipeline context.""" - if not self.supports_model(model_name): - raise ValueError(f"Model '{model_name}' not supported") - - model = self._models[model_name] - - # Track model usage - if model_name not in self.model_usage_stats: - self.model_usage_stats[model_name] = { - 'requests': 0, - 'total_execution_time': 0.0, - 'last_used': None - } - - self.model_usage_stats[model_name]['requests'] += 1 - self.model_usage_stats[model_name]['last_used'] = time.time() - - return model - - async def initialize(self) -> None: - """Initialize provider with pipeline validation support.""" - - # Verify all models are properly configured - for model_name, model in self._models.items(): - try: - health = await model.health_check() - if not health: - logger.warning(f"Model {model_name} failed health check") - except Exception as e: - logger.warning(f"Error during model {model_name} initialization: {e}") - - self.is_initialized = True - logger.info(f"PipelineTestProvider initialized with {len(self._models)} models") - - def get_usage_statistics(self) -> Dict[str, Any]: - """Get provider usage statistics for analysis.""" - - total_requests = sum(stats['requests'] for stats in self.model_usage_stats.values()) - - return { - 'total_models': len(self._models), - 'total_requests': total_requests, - 'model_usage': self.model_usage_stats.copy(), - 'most_used_model': max(self.model_usage_stats.items(), key=lambda x: x[1]['requests'])[0] if self.model_usage_stats else None, - 'provider_uptime': time.time() - (min(stats.get('last_used', time.time()) for stats in self.model_usage_stats.values()) if self.model_usage_stats else time.time()) - } - - def reset_usage_statistics(self) -> None: - """Reset usage statistics for clean testing.""" - self.model_usage_stats.clear() - for model in self._models.values(): - if hasattr(model, 'execution_count'): - model.execution_count = 0 - if hasattr(model, 'validation_history'): - model.validation_history.clear() - - -class PipelineIntegrationValidator: - """Systematic pipeline integration validator using proven infrastructure patterns.""" - - def __init__( - self, - examples_dir: Optional[Path] = None, - orchestrator: Optional[Orchestrator] = None, - test_provider: Optional[PipelineTestProvider] = None, - enable_comprehensive_validation: bool = True - ): - """Initialize pipeline integration validator.""" - - self.examples_dir = examples_dir or Path("examples") - self.test_provider = test_provider or PipelineTestProvider() - - # Create model registry with test provider - self.model_registry = ModelRegistry() - self.model_registry.register_provider(self.test_provider) - - # Create orchestrator with test infrastructure - self.orchestrator = orchestrator or self._create_test_orchestrator() - - # Initialize validation components - self.pipeline_validator = PipelineValidator() - self.pipeline_test_suite = PipelineTestSuite( - examples_dir=self.examples_dir, - model_registry=self.model_registry, - orchestrator=self.orchestrator, - enable_llm_quality_review=enable_comprehensive_validation, - enable_enhanced_template_validation=enable_comprehensive_validation, - enable_performance_monitoring=True - ) - - # Results tracking - self.integration_results: List[PipelineIntegrationResult] = [] - - logger.info("PipelineIntegrationValidator initialized with systematic validation infrastructure") - - def _create_test_orchestrator(self) -> Orchestrator: - """Create orchestrator with pipeline test infrastructure.""" - - from ..control_systems.hybrid_control_system import HybridControlSystem - - # Create control system with test model registry - control_system = HybridControlSystem(model_registry=self.model_registry) - - # Create orchestrator with all required components - orchestrator = Orchestrator( - model_registry=self.model_registry, - control_system=control_system, - max_concurrent_tasks=5, # Limit for testing - debug_templates=True # Enable template debugging - ) - - return orchestrator - - async def validate_pipeline_integration( - self, - pipeline_name: str, - pipeline_path: Path - ) -> PipelineIntegrationResult: - """Validate a single pipeline's integration with the test infrastructure.""" - - start_time = time.time() - logger.info(f"Validating pipeline integration: {pipeline_name}") - - # Initialize result - result = PipelineIntegrationResult( - pipeline_name=pipeline_name, - validation_passed=False, - execution_successful=False, - integration_score=0.0 - ) - - try: - # 1. Basic pipeline validation using existing validator - validation_results = self.pipeline_validator.validate_pipeline_file(pipeline_path) - result.validation_passed = validation_results.get('valid', False) - - if not result.validation_passed: - result.issues.extend(validation_results.get('issues', [])) - result.warnings.extend(validation_results.get('warnings', [])) - - # 2. Test model integration - model_performance = await self._test_model_integration(pipeline_name) - result.test_model_performance = model_performance - - # 3. Provider integration validation - provider_status = await self._test_provider_integration(pipeline_name) - result.provider_integration_status = provider_status - - # 4. Orchestrator compatibility test - orchestrator_compat = await self._test_orchestrator_compatibility(pipeline_path) - result.orchestrator_compatibility = orchestrator_compat - result.execution_successful = orchestrator_compat.get('execution_successful', False) - - # 5. Calculate integration score - result.integration_score = self._calculate_integration_score(result) - - # 6. Generate recommendations - result.recommendations = self._generate_recommendations(result) - - except Exception as e: - logger.error(f"Error validating pipeline integration for {pipeline_name}: {e}") - result.issues.append(f"Integration validation error: {str(e)}") - - result.execution_time = time.time() - start_time - self.integration_results.append(result) - - logger.info(f"Pipeline integration validation completed for {pipeline_name} " - f"(score: {result.integration_score:.2f}, time: {result.execution_time:.1f}s)") - - return result - - async def _test_model_integration(self, pipeline_name: str) -> Dict[str, Any]: - """Test integration with PipelineTestModel.""" - - model_performance = { - 'model_available': False, - 'health_check_passed': False, - 'generation_test_passed': False, - 'structured_output_test_passed': False, - 'cost_estimation_working': False - } - - try: - # Get pipeline test model - model = await self.test_provider.get_model("pipeline-test-model") - model_performance['model_available'] = True - - # Test health check - health = await model.health_check() - model_performance['health_check_passed'] = health - - # Test text generation - response = await model.generate( - "Test pipeline validation", - pipeline_context={'pipeline_name': pipeline_name} - ) - model_performance['generation_test_passed'] = len(response) > 0 - - # Test structured output - schema = { - "type": "object", - "properties": { - "validation_status": {"type": "string"}, - "quality_score": {"type": "number"} - } - } - structured_response = await model.generate_structured( - "Generate validation result", - schema, - pipeline_context={'pipeline_name': pipeline_name} - ) - model_performance['structured_output_test_passed'] = isinstance(structured_response, dict) - - # Test cost estimation - cost = await model.estimate_cost(100, 50) - model_performance['cost_estimation_working'] = isinstance(cost, (int, float)) - - # Get model validation summary - model_performance['validation_summary'] = model.get_validation_summary() - - except Exception as e: - model_performance['error'] = str(e) - logger.warning(f"Model integration test failed for {pipeline_name}: {e}") - - return model_performance - - async def _test_provider_integration(self, pipeline_name: str) -> Dict[str, Any]: - """Test integration with PipelineTestProvider.""" - - provider_status = { - 'provider_initialized': False, - 'models_available': 0, - 'all_models_healthy': False, - 'usage_tracking_working': False - } - - try: - # Check provider initialization - provider_status['provider_initialized'] = self.test_provider.is_initialized - - # Check available models - available_models = self.test_provider.available_models - provider_status['models_available'] = len(available_models) - - # Test model health checks - healthy_models = 0 - for model_name in available_models[:5]: # Test first 5 models - try: - model = await self.test_provider.get_model(model_name) - if await model.health_check(): - healthy_models += 1 - except Exception: - pass - - provider_status['all_models_healthy'] = healthy_models > 0 - provider_status['healthy_models_count'] = healthy_models - - # Test usage tracking - initial_stats = self.test_provider.get_usage_statistics() - test_model = await self.test_provider.get_model("pipeline-test-model") - await test_model.generate("usage tracking test") - updated_stats = self.test_provider.get_usage_statistics() - - provider_status['usage_tracking_working'] = ( - updated_stats['total_requests'] > initial_stats['total_requests'] - ) - - except Exception as e: - provider_status['error'] = str(e) - logger.warning(f"Provider integration test failed for {pipeline_name}: {e}") - - return provider_status - - async def _test_orchestrator_compatibility(self, pipeline_path: Path) -> Dict[str, Any]: - """Test orchestrator compatibility with test infrastructure.""" - - orchestrator_compat = { - 'orchestrator_available': False, - 'pipeline_loadable': False, - 'execution_successful': False, - 'templates_resolved': False, - 'outputs_generated': False - } - - try: - orchestrator_compat['orchestrator_available'] = self.orchestrator is not None - - # Test pipeline loading (compilation) - try: - # Use pipeline test suite for execution - pipeline_info = type('PipelineInfo', (), { - 'name': pipeline_path.stem, - 'path': pipeline_path, - 'estimated_runtime': 30.0, - 'input_requirements': {} - })() - - # Test execution with timeout - execution_result = await asyncio.wait_for( - self.pipeline_test_suite._test_pipeline_execution(pipeline_info), - timeout=60.0 - ) - - orchestrator_compat['pipeline_loadable'] = True - orchestrator_compat['execution_successful'] = execution_result.success - orchestrator_compat['execution_time'] = execution_result.execution_time - - if execution_result.error: - orchestrator_compat['error'] = str(execution_result.error) - - # Check if outputs were generated - orchestrator_compat['outputs_generated'] = bool(execution_result.outputs) - - # Basic template resolution check - if execution_result.success: - # If execution succeeded, templates were likely resolved - orchestrator_compat['templates_resolved'] = True - - except asyncio.TimeoutError: - orchestrator_compat['error'] = 'Pipeline execution timed out' - except Exception as e: - orchestrator_compat['error'] = str(e) - orchestrator_compat['pipeline_loadable'] = 'YAML' not in str(e) # Can load if not YAML error - - except Exception as e: - orchestrator_compat['error'] = str(e) - logger.warning(f"Orchestrator compatibility test failed: {e}") - - return orchestrator_compat - - def _calculate_integration_score(self, result: PipelineIntegrationResult) -> float: - """Calculate overall integration score (0.0-100.0).""" - - score = 0.0 - - # Basic validation (25 points) - if result.validation_passed: - score += 25.0 - - # Model integration (25 points) - model_perf = result.test_model_performance - if model_perf.get('model_available', False): - score += 5.0 - if model_perf.get('health_check_passed', False): - score += 5.0 - if model_perf.get('generation_test_passed', False): - score += 10.0 - if model_perf.get('structured_output_test_passed', False): - score += 5.0 - - # Provider integration (25 points) - provider_status = result.provider_integration_status - if provider_status.get('provider_initialized', False): - score += 5.0 - if provider_status.get('models_available', 0) > 0: - score += 10.0 - if provider_status.get('all_models_healthy', False): - score += 5.0 - if provider_status.get('usage_tracking_working', False): - score += 5.0 - - # Orchestrator compatibility (25 points) - orchestrator_compat = result.orchestrator_compatibility - if orchestrator_compat.get('orchestrator_available', False): - score += 5.0 - if orchestrator_compat.get('pipeline_loadable', False): - score += 10.0 - if orchestrator_compat.get('execution_successful', False): - score += 10.0 - - # Penalty for issues - issue_penalty = min(10.0, len(result.issues) * 2.0) - score = max(0.0, score - issue_penalty) - - return score - - def _generate_recommendations(self, result: PipelineIntegrationResult) -> List[str]: - """Generate recommendations based on integration results.""" - - recommendations = [] - - # Basic validation recommendations - if not result.validation_passed: - recommendations.append("Fix pipeline validation issues before integration testing") - - # Model integration recommendations - model_perf = result.test_model_performance - if not model_perf.get('model_available', False): - recommendations.append("Ensure PipelineTestModel is properly registered in provider") - if not model_perf.get('health_check_passed', False): - recommendations.append("Check model health check implementation") - - # Provider integration recommendations - provider_status = result.provider_integration_status - if provider_status.get('models_available', 0) == 0: - recommendations.append("Register test models with PipelineTestProvider") - if not provider_status.get('usage_tracking_working', False): - recommendations.append("Verify usage statistics tracking is functional") - - # Orchestrator compatibility recommendations - orchestrator_compat = result.orchestrator_compatibility - if not orchestrator_compat.get('execution_successful', False): - if 'error' in orchestrator_compat: - recommendations.append(f"Fix orchestrator execution issue: {orchestrator_compat['error']}") - else: - recommendations.append("Debug pipeline execution failure with test orchestrator") - - # Performance recommendations - if result.execution_time > 30.0: - recommendations.append("Consider optimizing pipeline for faster test execution") - - # General recommendations - if result.integration_score < 70.0: - recommendations.append("Integration score is low - review all validation components") - elif result.integration_score < 90.0: - recommendations.append("Good integration - address remaining issues for optimal performance") - - return recommendations - - async def validate_all_examples(self) -> Dict[str, PipelineIntegrationResult]: - """Validate integration for all example pipelines.""" - - logger.info("Starting systematic validation of all example pipelines") - - results = {} - - # Discover all YAML files in examples directory - if not self.examples_dir.exists(): - logger.error(f"Examples directory not found: {self.examples_dir}") - return results - - yaml_files = list(self.examples_dir.glob("*.yaml")) + list(self.examples_dir.glob("*.yml")) - - logger.info(f"Found {len(yaml_files)} pipeline files to validate") - - for pipeline_path in yaml_files: - pipeline_name = pipeline_path.stem - - try: - result = await self.validate_pipeline_integration(pipeline_name, pipeline_path) - results[pipeline_name] = result - - status = "PASS" if result.integration_score >= 70.0 else "FAIL" - logger.info(f"{status}: {pipeline_name} (score: {result.integration_score:.1f})") - - except Exception as e: - logger.error(f"Failed to validate {pipeline_name}: {e}") - - # Create failed result - failed_result = PipelineIntegrationResult( - pipeline_name=pipeline_name, - validation_passed=False, - execution_successful=False, - integration_score=0.0 - ) - failed_result.issues.append(f"Validation failed: {str(e)}") - results[pipeline_name] = failed_result - - # Generate summary - total_pipelines = len(results) - passed_pipelines = len([r for r in results.values() if r.integration_score >= 70.0]) - avg_score = sum(r.integration_score for r in results.values()) / max(1, total_pipelines) - - logger.info(f"Pipeline integration validation completed:") - logger.info(f" Total pipelines: {total_pipelines}") - logger.info(f" Passed: {passed_pipelines} ({passed_pipelines/max(1, total_pipelines)*100:.1f}%)") - logger.info(f" Average score: {avg_score:.1f}") - - return results - - def get_integration_summary(self) -> Dict[str, Any]: - """Get comprehensive integration validation summary.""" - - if not self.integration_results: - return {"message": "No validation results available"} - - # Calculate statistics - total_validations = len(self.integration_results) - successful_validations = len([r for r in self.integration_results if r.validation_passed]) - successful_executions = len([r for r in self.integration_results if r.execution_successful]) - - avg_integration_score = sum(r.integration_score for r in self.integration_results) / total_validations - avg_execution_time = sum(r.execution_time for r in self.integration_results) / total_validations - - # Categorize results - high_quality = len([r for r in self.integration_results if r.integration_score >= 90.0]) - good_quality = len([r for r in self.integration_results if 70.0 <= r.integration_score < 90.0]) - needs_work = len([r for r in self.integration_results if r.integration_score < 70.0]) - - # Common issues - all_issues = [] - for result in self.integration_results: - all_issues.extend(result.issues) - - issue_frequency = {} - for issue in all_issues: - issue_type = issue.split(':')[0] if ':' in issue else issue - issue_frequency[issue_type] = issue_frequency.get(issue_type, 0) + 1 - - return { - 'total_validations': total_validations, - 'successful_validations': successful_validations, - 'successful_executions': successful_executions, - 'validation_success_rate': successful_validations / total_validations * 100, - 'execution_success_rate': successful_executions / total_validations * 100, - 'average_integration_score': avg_integration_score, - 'average_execution_time': avg_execution_time, - 'quality_distribution': { - 'high_quality': high_quality, - 'good_quality': good_quality, - 'needs_work': needs_work - }, - 'common_issues': dict(sorted(issue_frequency.items(), key=lambda x: x[1], reverse=True)[:10]), - 'provider_statistics': self.test_provider.get_usage_statistics() - } - - -# Utility functions for creating test infrastructure -def create_pipeline_test_orchestrator() -> Orchestrator: - """Create orchestrator with pipeline test infrastructure for external use.""" - - # Create test provider and registry - test_provider = PipelineTestProvider() - registry = ModelRegistry() - registry.register_provider(test_provider) - - # Create orchestrator with test components - from ..control_systems.hybrid_control_system import HybridControlSystem - - control_system = HybridControlSystem(model_registry=registry) - - orchestrator = Orchestrator( - model_registry=registry, - control_system=control_system, - max_concurrent_tasks=5, - debug_templates=True - ) - - return orchestrator - - -def create_pipeline_integration_validator(examples_dir: Optional[Path] = None) -> PipelineIntegrationValidator: - """Create pipeline integration validator with default configuration.""" - - return PipelineIntegrationValidator( - examples_dir=examples_dir or Path("examples"), - enable_comprehensive_validation=True - ) - - -# Export key classes and functions -__all__ = [ - 'PipelineTestModel', - 'PipelineTestProvider', - 'PipelineIntegrationValidator', - 'PipelineIntegrationResult', - 'create_pipeline_test_orchestrator', - 'create_pipeline_integration_validator' -] \ No newline at end of file diff --git a/tests/integration/test_pipeline_integration_infrastructure.py b/tests/integration/test_pipeline_integration_infrastructure.py deleted file mode 100644 index 544c2633..00000000 --- a/tests/integration/test_pipeline_integration_infrastructure.py +++ /dev/null @@ -1,722 +0,0 @@ -"""Integration tests for pipeline integration infrastructure. - -This module provides comprehensive tests for the new PipelineTestModel/PipelineTestProvider -patterns, demonstrating systematic pipeline validation using proven infrastructure. - -Tests cover: -- PipelineTestModel functionality and validation capabilities -- PipelineTestProvider integration and model management -- PipelineIntegrationValidator systematic validation -- Integration with existing orchestrator framework -- End-to-end pipeline execution validation -""" - -import asyncio -import pytest -import tempfile -from pathlib import Path -from typing import Any, Dict, List -from unittest.mock import Mock, patch - -from orchestrator.testing.pipeline_integration_infrastructure import ( - PipelineTestModel, - PipelineTestProvider, - PipelineIntegrationValidator, - PipelineIntegrationResult, - create_pipeline_test_orchestrator, - create_pipeline_integration_validator -) -from orchestrator.core.model import ModelCapabilities, ModelRequirements, ModelMetrics, ModelCost - - -class TestPipelineTestModel: - """Test suite for PipelineTestModel extending proven TestModel patterns.""" - - def test_pipeline_test_model_initialization(self): - """Test that PipelineTestModel initializes correctly with enhanced capabilities.""" - - model = PipelineTestModel() - - # Verify basic model properties - assert model.name == "pipeline-test-model" - assert model.provider == "pipeline-test-provider" - - # Verify enhanced capabilities for pipeline testing - assert model.capabilities.context_window == 32768 - assert "pipeline-execution" in model.capabilities.supported_tasks - assert "template-resolution" in model.capabilities.supported_tasks - assert "quality-assessment" in model.capabilities.supported_tasks - - # Verify pipeline-specific configuration - assert model.pipeline_validation_enabled is True - assert model.execution_count == 0 - assert isinstance(model.validation_history, list) - - def test_pipeline_test_model_custom_initialization(self): - """Test PipelineTestModel with custom configuration.""" - - custom_capabilities = ModelCapabilities( - supported_tasks=["custom-task"], - context_window=16384, - supports_function_calling=False, - supports_structured_output=True - ) - - mock_responses = { - "test_pipeline": "Custom response for test pipeline" - } - - model = PipelineTestModel( - name="custom-pipeline-model", - capabilities=custom_capabilities, - mock_responses=mock_responses, - pipeline_validation_enabled=False - ) - - assert model.name == "custom-pipeline-model" - assert model.capabilities.context_window == 16384 - assert model.capabilities.supports_function_calling is False - assert model.mock_responses == mock_responses - assert model.pipeline_validation_enabled is False - - @pytest.mark.asyncio - async def test_pipeline_test_model_text_generation(self): - """Test text generation with pipeline context.""" - - model = PipelineTestModel() - - # Test basic generation - response = await model.generate("Test prompt") - assert isinstance(response, str) - assert len(response) > 0 - assert model.execution_count == 1 - - # Test generation with pipeline context - pipeline_context = { - 'pipeline_name': 'test_pipeline' - } - - response = await model.generate( - "Validate this pipeline", - pipeline_context=pipeline_context - ) - - assert "test_pipeline" in response - assert model.execution_count == 2 - assert len(model.validation_history) == 2 - - # Verify validation history tracking - latest_record = model.validation_history[-1] - assert latest_record['pipeline_name'] == 'test_pipeline' - assert latest_record['prompt_type'] == 'validation' - - @pytest.mark.asyncio - async def test_pipeline_test_model_mock_responses(self): - """Test mock response functionality.""" - - mock_responses = { - "specific_pipeline": "Specific mock response for this pipeline" - } - - model = PipelineTestModel(mock_responses=mock_responses) - - # Test mock response usage - response = await model.generate( - "Any prompt", - pipeline_context={'pipeline_name': 'specific_pipeline'} - ) - - assert response == "Specific mock response for this pipeline" - - # Test fallback for unknown pipeline - response = await model.generate( - "Validation request", - pipeline_context={'pipeline_name': 'unknown_pipeline'} - ) - - assert "unknown_pipeline" in response - assert "validation" in response.lower() - - @pytest.mark.asyncio - async def test_pipeline_test_model_structured_output(self): - """Test structured output generation for pipeline validation.""" - - model = PipelineTestModel() - - schema = { - "type": "object", - "properties": { - "validation_status": {"type": "string"}, - "quality_score": {"type": "number"}, - "issues": {"type": "array"}, - "recommendations": {"type": "array"} - } - } - - response = await model.generate_structured( - "Generate validation result", - schema, - pipeline_context={'pipeline_name': 'test_pipeline'} - ) - - # Verify structured response - assert isinstance(response, dict) - assert 'validation_status' in response - assert 'quality_score' in response - assert 'issues' in response - assert 'recommendations' in response - - # Verify response content - assert response['validation_status'] == 'passed' - assert isinstance(response['quality_score'], (int, float)) - assert isinstance(response['issues'], list) - assert isinstance(response['recommendations'], list) - assert len(response['recommendations']) > 0 - - @pytest.mark.asyncio - async def test_pipeline_test_model_health_check(self): - """Test enhanced health check functionality.""" - - # Test with validation enabled - model = PipelineTestModel(pipeline_validation_enabled=True) - health = await model.health_check() - assert health is True - - # Test with validation disabled - model_no_validation = PipelineTestModel(pipeline_validation_enabled=False) - health = await model_no_validation.health_check() - assert health is True - - @pytest.mark.asyncio - async def test_pipeline_test_model_cost_estimation(self): - """Test cost estimation with usage tracking.""" - - model = PipelineTestModel() - - cost = await model.estimate_cost(100, 50) - assert cost == 0.0 # Always free for testing - - # Verify usage tracking - cost_records = [record for record in model.validation_history if record.get('type') == 'cost_estimation'] - assert len(cost_records) == 1 - - usage = cost_records[0]['usage'] - assert usage['input_tokens'] == 100 - assert usage['output_tokens'] == 50 - assert usage['estimated_cost'] == 0.0 - - def test_pipeline_test_model_validation_summary(self): - """Test validation summary functionality.""" - - model = PipelineTestModel() - - # Initially empty - summary = model.get_validation_summary() - assert summary['total_executions'] == 0 - assert summary['validation_attempts'] == 0 - - # Add some validation history manually for testing - model.execution_count = 5 - model.validation_history = [ - {'pipeline_name': 'pipeline1', 'prompt_type': 'validation'}, - {'pipeline_name': 'pipeline1', 'prompt_type': 'analysis'}, - {'pipeline_name': 'pipeline2', 'prompt_type': 'validation'}, - {'type': 'cost_estimation'}, # Should not count as validation attempt - ] - - summary = model.get_validation_summary() - assert summary['total_executions'] == 5 - assert summary['validation_attempts'] == 3 # Excludes cost_estimation - assert summary['unique_pipelines'] == 2 - assert summary['prompt_types']['validation'] == 2 - assert summary['prompt_types']['analysis'] == 1 - - -class TestPipelineTestProvider: - """Test suite for PipelineTestProvider extending proven MockTestProvider patterns.""" - - def test_pipeline_test_provider_initialization(self): - """Test provider initialization with enhanced capabilities.""" - - provider = PipelineTestProvider() - - assert provider.name == "pipeline-test-provider" - assert provider.is_initialized is True - assert provider.validation_enabled is True - assert provider.integration_tracking is True - - # Verify enhanced model registry - models = provider.available_models - assert len(models) >= 7 # At least 7 models including aliases - assert "pipeline-test-model" in models - assert "pipeline-validation-model" in models - assert "pipeline-quality-model" in models - assert "openai/gpt-4" in models - assert "anthropic/claude-sonnet-4-20250514" in models - - def test_pipeline_test_provider_model_support(self): - """Test model support checking.""" - - provider = PipelineTestProvider() - - # Test supported models - assert provider.supports_model("pipeline-test-model") is True - assert provider.supports_model("openai/gpt-4") is True - assert provider.supports_model("nonexistent-model") is False - - def test_pipeline_test_provider_model_capabilities(self): - """Test model capabilities retrieval.""" - - provider = PipelineTestProvider() - - # Test pipeline-specific model capabilities - capabilities = provider.get_model_capabilities("pipeline-test-model") - assert isinstance(capabilities, ModelCapabilities) - assert capabilities.context_window == 32768 - assert "pipeline-execution" in capabilities.supported_tasks - - # Test unsupported model - with pytest.raises(ValueError, match="Model 'unknown' not supported"): - provider.get_model_capabilities("unknown") - - def test_pipeline_test_provider_model_requirements(self): - """Test model requirements retrieval.""" - - provider = PipelineTestProvider() - - requirements = provider.get_model_requirements("pipeline-test-model") - assert isinstance(requirements, ModelRequirements) - assert requirements.memory_gb == 0.2 - assert requirements.cpu_cores == 1 - - def test_pipeline_test_provider_model_cost(self): - """Test model cost information.""" - - provider = PipelineTestProvider() - - cost = provider.get_model_cost("pipeline-test-model") - assert isinstance(cost, ModelCost) - assert cost.is_free is True - - def test_pipeline_test_provider_info(self): - """Test provider information with pipeline capabilities.""" - - provider = PipelineTestProvider() - - info = provider.get_provider_info() - assert info["name"] == "pipeline-test-provider" - assert info["type"] == "pipeline-test" - assert info["validation_enabled"] is True - assert info["integration_tracking"] is True - assert "supported_pipeline_features" in info - - features = info["supported_pipeline_features"] - assert "template-resolution" in features - assert "quality-assessment" in features - assert "execution-validation" in features - - @pytest.mark.asyncio - async def test_pipeline_test_provider_get_model(self): - """Test model retrieval with usage tracking.""" - - provider = PipelineTestProvider() - - # Get model and verify it's the right type - model = await provider.get_model("pipeline-test-model") - assert isinstance(model, PipelineTestModel) - - # Verify usage tracking - stats = provider.get_usage_statistics() - assert stats['total_requests'] == 1 - assert "pipeline-test-model" in stats['model_usage'] - assert stats['model_usage']["pipeline-test-model"]['requests'] == 1 - - @pytest.mark.asyncio - async def test_pipeline_test_provider_async_initialization(self): - """Test the async initialize() call. - - Renamed: this shared a name with the constructor test above, so it - silently replaced it and that test never ran. The two check different - things and both are wanted. - """ - provider = PipelineTestProvider() - await provider.initialize() - - assert provider.is_initialized is True - - def test_pipeline_test_provider_usage_statistics(self): - """Test usage statistics functionality.""" - - provider = PipelineTestProvider() - - # Initial state - stats = provider.get_usage_statistics() - assert stats['total_requests'] == 0 - assert stats['total_models'] >= 7 - - # Reset statistics - provider.reset_usage_statistics() - stats = provider.get_usage_statistics() - assert stats['total_requests'] == 0 - assert len(stats['model_usage']) == 0 - - -class TestPipelineIntegrationValidator: - """Test suite for PipelineIntegrationValidator systematic validation.""" - - def test_pipeline_integration_validator_initialization(self): - """Test validator initialization with test infrastructure.""" - - with tempfile.TemporaryDirectory() as temp_dir: - examples_dir = Path(temp_dir) - - validator = PipelineIntegrationValidator(examples_dir=examples_dir) - - assert validator.examples_dir == examples_dir - assert isinstance(validator.test_provider, PipelineTestProvider) - assert validator.model_registry is not None - assert validator.orchestrator is not None - assert validator.pipeline_validator is not None - assert validator.pipeline_test_suite is not None - - @pytest.mark.asyncio - async def test_pipeline_integration_validator_model_integration_test(self): - """Test model integration validation.""" - - validator = PipelineIntegrationValidator() - - result = await validator._test_model_integration("test_pipeline") - - # Verify all model integration tests - assert result['model_available'] is True - assert result['health_check_passed'] is True - assert result['generation_test_passed'] is True - assert result['structured_output_test_passed'] is True - assert result['cost_estimation_working'] is True - assert 'validation_summary' in result - - @pytest.mark.asyncio - async def test_pipeline_integration_validator_provider_integration_test(self): - """Test provider integration validation.""" - - validator = PipelineIntegrationValidator() - - result = await validator._test_provider_integration("test_pipeline") - - # Verify provider integration tests - assert result['provider_initialized'] is True - assert result['models_available'] >= 7 - assert result['all_models_healthy'] is True - assert result['healthy_models_count'] > 0 - assert result['usage_tracking_working'] is True - - @pytest.mark.asyncio - async def test_pipeline_integration_validator_single_pipeline_validation(self): - """Test validation of a single pipeline.""" - - with tempfile.TemporaryDirectory() as temp_dir: - examples_dir = Path(temp_dir) - - # Create a test pipeline file - test_pipeline = examples_dir / "test_pipeline.yaml" - test_pipeline.write_text(""" -# Test Pipeline -name: Test Pipeline -description: Simple test pipeline for validation - -steps: - - id: test_step - tool: filesystem - action: read - parameters: - path: "test.txt" -""") - - validator = PipelineIntegrationValidator(examples_dir=examples_dir) - - result = await validator.validate_pipeline_integration("test_pipeline", test_pipeline) - - # Verify result structure - assert isinstance(result, PipelineIntegrationResult) - assert result.pipeline_name == "test_pipeline" - assert isinstance(result.integration_score, float) - assert result.integration_score >= 0.0 - assert result.integration_score <= 100.0 - assert isinstance(result.test_model_performance, dict) - assert isinstance(result.provider_integration_status, dict) - assert isinstance(result.orchestrator_compatibility, dict) - assert isinstance(result.recommendations, list) - - def test_pipeline_integration_validator_integration_score_calculation(self): - """Test integration score calculation logic.""" - - validator = PipelineIntegrationValidator() - - # Create test result with perfect scores - perfect_result = PipelineIntegrationResult( - pipeline_name="perfect", - validation_passed=True, - execution_successful=True, - integration_score=0.0, - test_model_performance={ - 'model_available': True, - 'health_check_passed': True, - 'generation_test_passed': True, - 'structured_output_test_passed': True, - 'cost_estimation_working': True - }, - provider_integration_status={ - 'provider_initialized': True, - 'models_available': 10, - 'all_models_healthy': True, - 'usage_tracking_working': True - }, - orchestrator_compatibility={ - 'orchestrator_available': True, - 'pipeline_loadable': True, - 'execution_successful': True - } - ) - - score = validator._calculate_integration_score(perfect_result) - assert score == 100.0 - - # Test with issues - result_with_issues = perfect_result - result_with_issues.issues = ["Issue 1", "Issue 2", "Issue 3"] - - score_with_penalty = validator._calculate_integration_score(result_with_issues) - assert score_with_penalty < 100.0 - assert score_with_penalty >= 90.0 # Max 10 point penalty - - def test_pipeline_integration_validator_recommendations(self): - """Test recommendation generation.""" - - validator = PipelineIntegrationValidator() - - # Create result with various issues - problematic_result = PipelineIntegrationResult( - pipeline_name="problematic", - validation_passed=False, - execution_successful=False, - integration_score=50.0, - test_model_performance={'model_available': False}, - provider_integration_status={'models_available': 0}, - orchestrator_compatibility={'execution_successful': False, 'error': 'Test error'} - ) - - recommendations = validator._generate_recommendations(problematic_result) - - assert len(recommendations) > 0 - assert any("validation issues" in rec.lower() for rec in recommendations) - assert any("model" in rec.lower() for rec in recommendations) - assert any("test error" in rec.lower() for rec in recommendations) - - def test_pipeline_integration_validator_summary_generation(self): - """Test integration summary generation.""" - - validator = PipelineIntegrationValidator() - - # Add mock results - validator.integration_results = [ - PipelineIntegrationResult( - pipeline_name="pipeline1", - validation_passed=True, - execution_successful=True, - integration_score=95.0, - execution_time=1.5 - ), - PipelineIntegrationResult( - pipeline_name="pipeline2", - validation_passed=False, - execution_successful=False, - integration_score=40.0, - execution_time=0.8, - issues=["Validation failed", "Execution error"] - ) - ] - - summary = validator.get_integration_summary() - - assert summary['total_validations'] == 2 - assert summary['successful_validations'] == 1 - assert summary['successful_executions'] == 1 - assert summary['validation_success_rate'] == 50.0 - assert summary['execution_success_rate'] == 50.0 - assert summary['average_integration_score'] == 67.5 - assert summary['quality_distribution']['high_quality'] == 1 - assert summary['quality_distribution']['needs_work'] == 1 - - -class TestUtilityFunctions: - """Test suite for utility functions.""" - - def test_create_pipeline_test_orchestrator(self): - """Test orchestrator creation utility.""" - - orchestrator = create_pipeline_test_orchestrator() - - assert orchestrator is not None - assert orchestrator.model_registry is not None - assert orchestrator.control_system is not None - - # Verify test provider is registered - providers = orchestrator.model_registry.get_registered_providers() - assert any(provider.name == "pipeline-test-provider" for provider in providers) - - def test_create_pipeline_integration_validator(self): - """Test validator creation utility.""" - - with tempfile.TemporaryDirectory() as temp_dir: - examples_dir = Path(temp_dir) - - validator = create_pipeline_integration_validator(examples_dir) - - assert isinstance(validator, PipelineIntegrationValidator) - assert validator.examples_dir == examples_dir - - -class TestIntegrationScenarios: - """Integration test scenarios demonstrating end-to-end functionality.""" - - @pytest.mark.asyncio - async def test_complete_pipeline_integration_workflow(self): - """Test complete integration workflow from model creation to validation.""" - - with tempfile.TemporaryDirectory() as temp_dir: - examples_dir = Path(temp_dir) - - # Create test pipeline - pipeline_content = """ -name: Integration Test Pipeline -description: Test pipeline for integration validation -version: "1.0.0" - -parameters: - input_text: - type: string - default: "Hello, world!" - description: Input text for processing - -steps: - - id: process_text - tool: text-processing - action: analyze - parameters: - text: "{{ input_text }}" - analysis_type: "sentiment" - - - id: save_result - tool: filesystem - action: write - parameters: - path: "outputs/result.txt" - content: "Analysis: {{ process_text.result }}" - dependencies: - - process_text -""" - - test_pipeline = examples_dir / "integration_test.yaml" - test_pipeline.write_text(pipeline_content) - - # Create validator - validator = PipelineIntegrationValidator(examples_dir=examples_dir) - - # Run complete validation - result = await validator.validate_pipeline_integration("integration_test", test_pipeline) - - # Verify comprehensive validation - assert result.pipeline_name == "integration_test" - assert isinstance(result.integration_score, float) - assert result.execution_time > 0.0 - - # Verify all validation components ran - assert len(result.test_model_performance) > 0 - assert len(result.provider_integration_status) > 0 - assert len(result.orchestrator_compatibility) > 0 - - # Verify recommendations generated - assert isinstance(result.recommendations, list) - - # Get integration summary - summary = validator.get_integration_summary() - assert summary['total_validations'] == 1 - - @pytest.mark.asyncio - async def test_multiple_pipeline_validation(self): - """Test validation of multiple pipelines systematically.""" - - with tempfile.TemporaryDirectory() as temp_dir: - examples_dir = Path(temp_dir) - - # Create multiple test pipelines - pipelines = { - "simple.yaml": """ -name: Simple Pipeline -steps: - - id: step1 - tool: test - action: run -""", - "complex.yaml": """ -name: Complex Pipeline -description: More complex pipeline with dependencies - -parameters: - threshold: - type: number - default: 0.5 - -steps: - - id: analyze - tool: analysis - action: compute - parameters: - threshold: "{{ threshold }}" - - - id: process - tool: processing - action: transform - parameters: - data: "{{ analyze.result }}" - dependencies: - - analyze - - - id: output - tool: filesystem - action: write - parameters: - path: "output.json" - content: "{{ process.transformed }}" - dependencies: - - process -""" - } - - for filename, content in pipelines.items(): - (examples_dir / filename).write_text(content) - - # Run validation on all pipelines - validator = PipelineIntegrationValidator(examples_dir=examples_dir) - results = await validator.validate_all_examples() - - # Verify results - assert len(results) == 2 - assert "simple" in results - assert "complex" in results - - # Verify each result - for pipeline_name, result in results.items(): - assert isinstance(result, PipelineIntegrationResult) - assert result.pipeline_name == pipeline_name - assert result.integration_score >= 0.0 - - # Get comprehensive summary - summary = validator.get_integration_summary() - assert summary['total_validations'] == 2 - assert summary['average_integration_score'] >= 0.0 - - -if __name__ == "__main__": - # Run tests with pytest - pytest.main([__file__, "-v"]) \ No newline at end of file diff --git a/tests/test_infrastructure.py b/tests/test_infrastructure.py index e567e80b..e35d65f0 100644 --- a/tests/test_infrastructure.py +++ b/tests/test_infrastructure.py @@ -35,7 +35,17 @@ def __init__( """Initialize test model with minimal defaults.""" if capabilities is None: capabilities = ModelCapabilities( - supported_tasks=["text-generation", "analysis"], + # "generate" is the task name the control system actually asks + # for and that real providers advertise (asserted in + # tests/test_provider_abstractions.py). Advertising only + # "text-generation" made this model ineligible for every + # `action: generate` step -- selection failed with + # NoEligibleModelsError, so no pipeline could use it. The + # older names are kept so existing callers keep matching. + supported_tasks=[ + "generate", "analyze", "transform", "summarize", + "text-generation", "analysis", + ], context_window=8192, supports_function_calling=True, supports_structured_output=True @@ -161,16 +171,28 @@ async def initialize(self) -> None: def create_test_orchestrator(): - """Create orchestrator with test model for testing.""" + """Create orchestrator with test model for testing. + + Uses `models.model_registry.ModelRegistry`, not `models.registry`. The two + classes share a name and are not interchangeable: only `models.registry` + has `register_provider`, and only `models.model_registry` has + `can_provide_models`, which `Orchestrator.__init__` calls. This helper + previously built the first and every call raised + + AttributeError: 'ModelRegistry' object has no attribute + 'can_provide_models' + + so it could not construct an orchestrator at all. Registering the model + directly avoids needing `register_provider` and works with the class the + orchestrator expects. (The duplicate-registry split itself is #429.) + """ from orchestrator.orchestrator import Orchestrator - from orchestrator.models.registry import ModelRegistry + from orchestrator.models.model_registry import ModelRegistry from orchestrator.control_systems.hybrid_control_system import HybridControlSystem - - # Create registry and add test provider + registry = ModelRegistry() - test_provider = MockTestProvider() - registry.register_provider(test_provider) - + registry.register_model(MockTestModel()) + # Create control system with populated registry control_system = HybridControlSystem(model_registry=registry) @@ -320,7 +342,7 @@ def test_create_test_orchestrator(): import os import subprocess import re -from typing import Dict, List, Tuple +from typing import Tuple # Dict/List already imported at the top of the file from dataclasses import dataclass diff --git a/tests/test_pipeline_model_contracts.py b/tests/test_pipeline_model_contracts.py new file mode 100644 index 00000000..4de8120f --- /dev/null +++ b/tests/test_pipeline_model_contracts.py @@ -0,0 +1,255 @@ +"""Contract tests for pipelines that use a model. + +The golden pipelines (tests/test_golden_pipelines.py) cover the tool-only +path: no model, no provider. This module covers the other half -- a pipeline +whose steps call a model -- using the deterministic model and provider from +tests/test_infrastructure.py, so it stays hermetic: no network, no API key, no +provider SDK. + +These replace src/orchestrator/testing/pipeline_integration_infrastructure.py, +retired in #435. That module built a parallel testing architecture -- its own +model, provider, validator, scoring system and orchestration facade -- inside +the shipped package, was written against constructor arguments that do not +exist, and consequently never ran. What was worth keeping was the *intent*: +prove that a model-using pipeline compiles, executes, fails honestly, and can +produce structured output. That intent is these tests, against the canonical +YAMLCompiler and Orchestrator and nothing else. + +Test-only models and providers stay under tests/. They are not product code. +""" + +import asyncio +import json +from collections.abc import Mapping + +import jsonschema +import pytest + +from orchestrator.compiler.yaml_compiler import YAMLCompiler +from orchestrator.core.exceptions import YAMLCompilerError +from orchestrator.models.model_registry import ModelRegistry +from tests.test_infrastructure import MockTestModel, create_test_orchestrator + +pytestmark = [pytest.mark.contract, pytest.mark.e2e] + +# Jinja delimiters that must never survive into a result. A step that emits +# "{{ topic }}" has silently shipped its own source instead of a value (#153). +UNRESOLVED_TEMPLATE_MARKERS = ("{{", "}}", "{%", "%}") + + +def assert_no_unresolved_templates(text: str) -> None: + for marker in UNRESOLVED_TEMPLATE_MARKERS: + assert marker not in text, ( + f"unresolved template marker {marker!r} survived into the result: {text!r}" + ) + + +TOOL_PIPELINE = """ +id: canonical_tool_pipeline +name: Canonical Tool Pipeline +description: A deterministic local-tool step +steps: + - id: write_note + tool: filesystem + action: write + parameters: + path: "out/note.txt" + content: "hello" +""" + +# `{{ topic }}` is deliberate: it makes the "no unresolved templates" assertion +# meaningful. A pipeline with no templates would pass that check vacuously. +MODEL_TOPIC = "the water cycle" + +MODEL_PIPELINE = """ +id: canonical_model_pipeline +name: Canonical Model Pipeline +description: A single generate step against the deterministic test model +parameters: + topic: + type: string + default: "unsubstituted-default" +steps: + - id: think + action: generate + parameters: + prompt: "Summarise the following: {{ topic }}" +""" + +# `steps` must be a list of steps. A mapping where a list belongs is malformed. +MALFORMED_PIPELINE = """ +id: broken_pipeline +name: Broken Pipeline +steps: + not_a_list: true +""" + +# One source of truth: the pipeline declares this schema and the test validates +# against this same object, so the two cannot drift apart. YAML is a superset of +# JSON, so the dumped schema drops straight into the document. +STRUCTURED_SCHEMA = { + "type": "object", + "properties": {"test_output": {"type": "string"}}, + "required": ["test_output"], + "additionalProperties": False, +} + + +def structured_pipeline(action: str) -> str: + return f""" +id: canonical_structured_pipeline +name: Canonical Structured Pipeline +steps: + - id: extract + action: {action} + parameters: + prompt: "Extract the fields" + schema: {json.dumps(STRUCTURED_SCHEMA)} +""" + +FAILING_PIPELINE = """ +id: failing_pipeline +name: Failing Pipeline +steps: + - id: read_missing + action: file + parameters: + action: read + path: "/nonexistent/definitely/not/here.txt" +""" + + +@pytest.fixture +def compiler(): + """A real YAMLCompiler backed by the deterministic test model.""" + registry = ModelRegistry() + registry.register_model(MockTestModel()) + return YAMLCompiler(model_registry=registry) + + +# --------------------------------------------------------------------------- +# compile +# --------------------------------------------------------------------------- + +def test_a_valid_pipeline_compiles(compiler): + """The canonical compiler accepts a well-formed pipeline.""" + pipeline = asyncio.run( + compiler.compile(TOOL_PIPELINE, {}, resolve_ambiguities=False) + ) + + assert pipeline.id == "canonical_tool_pipeline" + assert "write_note" in pipeline.tasks + assert pipeline.tasks["write_note"].action == "write" + + +def test_a_malformed_pipeline_is_refused(compiler): + """Validation failure must raise rather than compile to something broken.""" + with pytest.raises(YAMLCompilerError): + asyncio.run( + compiler.compile(MALFORMED_PIPELINE, {}, resolve_ambiguities=False) + ) + + +@pytest.mark.xfail( + strict=True, + reason=( + "#241 / #104: a model step runs but does not validate. Strict " + "validation resolves `action: generate` as a TOOL name and reports " + "\"Tool 'generate' not found in registry\", while the executor runs " + "the same pipeline happily -- see test_a_model_pipeline_executes " + "below. strict=True so this flips to a visible failure the moment " + "the two agree, rather than sitting here forgotten." + ), +) +def test_a_model_pipeline_compiles(compiler): + """`validate` and `run` must accept the same pipelines. Today they do not.""" + pipeline = asyncio.run( + compiler.compile(MODEL_PIPELINE, {}, resolve_ambiguities=False) + ) + + assert "think" in pipeline.tasks + + +# --------------------------------------------------------------------------- +# execute +# --------------------------------------------------------------------------- + +def test_a_model_pipeline_executes(): + """The orchestrator selects the deterministic model and runs the step. + + Asserting only that the step key exists would pass for a step that failed, + returned nothing, or shipped its own unrendered template. Each of those has + happened here, so each is pinned. + """ + orchestrator = create_test_orchestrator() + result = asyncio.run( + orchestrator.execute_yaml( + yaml_content=MODEL_PIPELINE, context={"topic": MODEL_TOPIC} + ) + ) + + assert "think" in result, f"step output missing from {sorted(result)}" + step = result["think"] + assert isinstance(step, Mapping), f"expected a result envelope, got {type(step)}" + + # the envelope + assert step["success"] is True, f"step did not succeed: {step}" + assert step.get("error") in (None, ""), f"a successful step must carry no error: {step}" + assert step["action"] == "generate_text" + + # the model that was actually selected, by identity + assert step["model_used"] == MockTestModel().name + + # the response + response = step["result"] + assert isinstance(response, str), f"expected text, got {type(response)}" + assert response.strip(), "a successful generate step must return a non-empty response" + + # the template was rendered before the model saw it + assert MODEL_TOPIC in response, ( + f"the rendered parameter never reached the model: {response!r}" + ) + assert "unsubstituted-default" not in response + assert_no_unresolved_templates(response) + + +@pytest.mark.parametrize("action", ["generate-structured", "generate_structured"]) +def test_structured_output_pipeline_returns_an_object(action): + """A structured step returns a schema-conforming object, not a sentence. + + Both spellings are pinned. Every other action in the project uses + underscores, so `generate_structured` reads as the canonical name -- but + only `generate-structured` used to dispatch. The underscore spelling fell + through to the natural-language branch, which turns an unrecognised action + into a prompt: the step returned a *string* and still reported success. + """ + orchestrator = create_test_orchestrator() + result = asyncio.run( + orchestrator.execute_yaml(yaml_content=structured_pipeline(action), context={}) + ) + + assert "extract" in result, f"step output missing from {sorted(result)}" + payload = result["extract"] + + assert isinstance(payload, Mapping), ( + f"a structured step must return an object, got {type(payload)}: {payload!r}" + ) + # Conformance, not merely dict-ness. + jsonschema.validate(instance=payload, schema=STRUCTURED_SCHEMA) + assert_no_unresolved_templates(json.dumps(payload)) + + +def test_execution_failure_is_reported_not_swallowed(): + """A step that cannot succeed must surface as a failure. + + A pipeline reporting success having done nothing is the failure mode this + project has repeatedly had to fix, so it is pinned here. + """ + orchestrator = create_test_orchestrator() + result = asyncio.run( + orchestrator.execute_yaml(yaml_content=FAILING_PIPELINE, context={}) + ) + + step = result["read_missing"] + assert step["success"] is False, "reading a missing file must not report success" + assert step["error"], "a failed step must say why"