diff --git a/dispatch/orchestrator.py b/dispatch/orchestrator.py index 9f9525d0..4a179109 100644 --- a/dispatch/orchestrator.py +++ b/dispatch/orchestrator.py @@ -7,23 +7,18 @@ import datetime import logging import os -import tempfile from io import StringIO -from typing import Any from core.structured_logging import ( CorrelationContext, get_logger, get_or_create_correlation_id, - log_backend_call, log_with_context, ) from core.utils import normalize_bool -from core.utils.file_utils import calculate_file_checksum from core.utils.folder_utils import build_effective_folder from dispatch.error_handler import ErrorHandler from dispatch.file_utils import ( - apply_file_rename, extract_invoice_numbers, write_to_run_log, ) @@ -450,358 +445,6 @@ def _process_file_with_pipeline( effective_folder=context, ) - def _process_split_pipeline( - self, - current_file: str, - file_path: str, - file_basename: str, - context: ProcessingContext, - result: FileResult, - run_log: RunLog, - ) -> bool: - """Process split pipeline path. Returns True if split flow was executed.""" - split_edi = normalize_bool(context.effective_folder.get("split_edi", False)) - logger.debug( - "Splitter step: enabled=%s, split_edi=%s", - bool(self.config.splitter_step), - split_edi, - ) - if not (self.config.splitter_step and split_edi): - return False - - with tempfile.TemporaryDirectory() as temp_dir: - split_result = self.config.splitter_step.split( - current_file, - temp_dir, - context.effective_folder, - context.upc_dict, - ) - split_files = self._extract_split_files(split_result) - files_to_send = split_files if split_files else [current_file] - - all_sends_succeeded = True - for pipeline_file in files_to_send: - current_pipeline_file, did_convert = self._apply_conversion( - current_file=pipeline_file, - file_basename=file_basename, - original_file_path=file_path, - context=context, - run_log=run_log, - _validation_passed=result.validated, - ) - if did_convert: - result.converted = True - - send_result = self._send_pipeline_file( - apply_file_rename( - current_pipeline_file, - context.effective_folder.get("rename_file", "").strip(), - getattr(context, "temp_dirs", []), - ), - context.effective_folder, - run_log, - ) - if not send_result: - all_sends_succeeded = False - self._record_split_send_failure(result, current_pipeline_file) - - result.sent = all_sends_succeeded - if result.sent: - self._log_success_with_invoices( - run_log=run_log, - file_basename=file_basename, - file_path=file_path, - ) - return True - - def _extract_split_files(self, split_result: Any) -> list[str]: - """Extract output split file paths from splitter result.""" - if hasattr(split_result, "files"): - return [f[0] for f in split_result.files] - if isinstance(split_result, list): - return split_result - return [] - - def _record_split_send_failure( - self, - result: FileResult, - current_pipeline_file: str, - ) -> None: - """Record split-file send failures from send manager errors.""" - send_errors = self.send_manager.get_errors() - if send_errors: - for backend_name, error_message in send_errors.items(): - result.errors.append( - self._format_send_error( - backend_name, error_message, is_split_file=True - ) - ) - return - - result.errors.append(f"Failed to send split file: {current_pipeline_file}") - - def _record_send_failure( - self, - result: FileResult, - file_basename: str, - current_file: str, - run_log: RunLog, - ) -> None: - """Record non-split send failures and corresponding log messages.""" - send_errors = self.send_manager.get_errors() - if send_errors: - for backend_name, error_message in send_errors.items(): - result.errors.append( - self._format_send_error(backend_name, error_message) - ) - _msg = ( - f"FAILED sending {file_basename} via {backend_name}: " - f"{error_message}" - ) - write_to_run_log(run_log, _msg) - log_with_context( - logger, - logging.INFO, - _msg, - correlation_id=get_or_create_correlation_id(), - operation="run_log", - ) - return - result.errors.append(f"Failed to send file: {current_file}") - write_to_run_log(run_log, f"FAILED sending {file_basename}") - log_with_context( - logger, - logging.INFO, - f"FAILED sending {file_basename}", - correlation_id=get_or_create_correlation_id(), - operation="run_log", - ) - - def _format_send_error( - self, backend_name: str, error_message: str, *, is_split_file: bool = False - ) -> str: - """Format a send error message for a specific backend.""" - file_type = "split file" if is_split_file else "file" - return f"Failed to send {file_type} via {backend_name}: {error_message}" - - def _log_success_with_invoices( - self, - run_log: RunLog, - file_basename: str, - file_path: str, - ) -> None: - """Log success and any extracted invoice numbers.""" - write_to_run_log(run_log, f"Success: {file_basename}") - log_with_context( - logger, - logging.INFO, - f"Success: {file_basename}", - correlation_id=get_or_create_correlation_id(), - operation="run_log", - ) - invoice_numbers = self._extract_invoice_numbers(file_path) - if invoice_numbers: - write_to_run_log(run_log, f"Invoice numbers: {invoice_numbers}") - log_with_context( - logger, - logging.INFO, - f"Invoice numbers: {invoice_numbers}", - correlation_id=get_or_create_correlation_id(), - operation="run_log", - ) - - def _run_validation_pipeline( - self, - current_file: str, - context: ProcessingContext, - result: FileResult, - run_log: RunLog, - file_basename: str, - ) -> tuple[bool, str]: - """Run validation and update processing state. - - Returns: - Tuple of (continue_processing, current_file_path) - - """ - should_validate = self._should_validate(context.effective_folder) - logger.debug( - "Validation step: enabled=%s, should_validate=%s", - bool(self.config.validator_step), - should_validate, - ) - - if not should_validate: - return True, current_file - - if not self.config.validator_step: - return True, current_file - - validation_output = self.config.validator_step.execute( - current_file, context.effective_folder - ) - is_valid, errors_or_file = self._normalize_validation_output( - validation_output=validation_output, - current_file=current_file, - ) - return self._apply_validation_outcome( - is_valid=is_valid, - errors_or_file=errors_or_file, - current_file=current_file, - result=result, - run_log=run_log, - file_basename=file_basename, - context=context, - ) - - def _apply_conversion( - self, - current_file: str, - file_basename: str, - original_file_path: str, - context: ProcessingContext, - run_log: RunLog, - *, - _validation_passed: bool = True, - ) -> tuple[str, bool]: - """Apply the converter step for a single file path. - - Args: - current_file: Current file path in the pipeline - file_basename: Base name of the file - original_file_path: Original file path before any processing - context: Processing context - run_log: Run log for recording activity - validation_passed: Whether validation passed for this file - - Returns: - Tuple of (final_file_path, did_convert) - - """ - did_convert = False - converter_step = self.config.converter_step - convert_edi = context.effective_folder.get("convert_edi", False) - convert_format = context.effective_folder.get("convert_to_format", "") - run_conversion = converter_step is not None and convert_edi - - logger.debug( - "Converter step: enabled=%s, convert_edi=%s, convert_to_format=%s", - bool(converter_step), - convert_edi, - convert_format, - ) - - if run_conversion: - _msg = f"Converting {file_basename} to {convert_format}" - write_to_run_log(run_log, _msg) - log_with_context( - logger, - logging.INFO, - _msg, - correlation_id=get_or_create_correlation_id(), - operation="run_log", - ) - converted_file = self._execute_conversion_step( - converter_step=converter_step, - current_file=current_file, - _original_file_path=original_file_path, - context=context, - ) - if converted_file: - if converted_file != original_file_path: - context.temp_files.append(converted_file) - current_file = converted_file - did_convert = True - elif str(convert_format).strip(): - raise RuntimeError( - "Conversion was requested for " - f"format '{convert_format}' but no converted output " - "was produced" - ) - - return current_file, did_convert - - def _execute_conversion_step( - self, - converter_step: Any, - current_file: str, - _original_file_path: str, - context: ProcessingContext, - ) -> str | None: - """Execute the configured converter step and return converted path or None. - - Kept as a separate method to keep _apply_conversion concise. - """ - return converter_step.execute( # type: ignore[no-any-return] - current_file, - context.effective_folder, - context.settings, - context.upc_dict, - context=context, - ) - - def _set_checksum_for_result( - self, result: FileResult, file_path: str, file_basename: str - ) -> None: - """Calculate and set checksum on a FileResult and log debug info.""" - - result.checksum = calculate_file_checksum(file_path) - logger.debug("Calculated checksum for %s: %s", file_basename, result.checksum) - - def _normalize_validation_output( - self, validation_output: Any, current_file: str - ) -> tuple[bool, Any]: - """Normalize validator step output to `(is_valid, errors_or_file)` tuple. - - Delegates to ``dispatch.pipeline.validator.normalize_validation_output`` - (spec: refactor-dispatch-simplification.md ยง3.4 Phase 1). - """ - from dispatch.pipeline.validator import normalize_validation_output - - return normalize_validation_output(validation_output, current_file) - - def _apply_validation_outcome( - self, - *, - is_valid: bool, - errors_or_file: Any, - current_file: str, - result: FileResult, - run_log: RunLog, - file_basename: str, - context: ProcessingContext, - ) -> tuple[bool, str]: - """Apply validation result to file result and control flow. - - Returns: - Tuple of (continue_processing, current_file_path) - - """ - result.record_validation_outcome(is_valid, errors_or_file) - - if is_valid: - if isinstance(errors_or_file, str): - return True, errors_or_file - return True, current_file - - _msg = f"Validation failed for {file_basename}: {result.errors}" - - write_to_run_log(run_log, _msg) - log_with_context( - logger, - logging.INFO, - _msg, - correlation_id=get_or_create_correlation_id(), - operation="run_log", - ) - - if not normalize_bool( - context.effective_folder.get("force_edi_validation", False) - ): - return False, current_file - - return True, current_file - def _detect_enabled_backends(self, folder: dict) -> list[str]: """Detect which backends are enabled for a folder. @@ -834,68 +477,6 @@ def _build_processing_context( upc_dict=upc_dict, ) - def _send_pipeline_file( - self, file_path: str, folder: dict, run_log: RunLog | None = None - ) -> bool: - """Send file through pipeline to backends. - - Args: - file_path: Path to the file to send - folder: Folder configuration dictionary - run_log: Optional run log for recording processing activity - - Returns: - True if file was sent successfully - - """ - enabled_backends = self.send_manager.get_enabled_backends(folder) - - if not enabled_backends: - return False - - file_basename = os.path.basename(file_path) - - for backend_name in enabled_backends: - display_name = self.send_manager.DEFAULT_BACKENDS.get(backend_name, {}).get( - "display_name", backend_name - ) - _msg = f"sending {file_basename} to {display_name}" - write_to_run_log(run_log, _msg) - log_with_context( - logger, - logging.INFO, - _msg, - correlation_id=get_or_create_correlation_id(), - operation="run_log", - ) - log_backend_call( - logger, - backend_name, - "send", - endpoint=folder.get( - f"{backend_name}_server", - folder.get(f"{backend_name}_to_directory", ""), - ), - correlation_id=get_or_create_correlation_id(), - ) - - settings = self.config.settings - send_results = self.send_manager.send_all( - enabled_backends, file_path, folder, settings - ) - - success = all(send_results.values()) - for backend_name, backend_success in send_results.items(): - log_backend_call( - logger, - backend_name, - "send", - success=backend_success, - correlation_id=get_or_create_correlation_id(), - ) - - return success - def process_file(self, file_path: str, folder: dict) -> FileResult: """Process a single file via the pipeline path. diff --git a/tests/unit/dispatch_tests/test_orchestrator_pipeline.py b/tests/unit/dispatch_tests/test_orchestrator_pipeline.py index b0ec4b54..6b38e476 100644 --- a/tests/unit/dispatch_tests/test_orchestrator_pipeline.py +++ b/tests/unit/dispatch_tests/test_orchestrator_pipeline.py @@ -969,62 +969,6 @@ def test_process_file_with_pipeline_cleans_context_registered_temp_dirs_in_final assert not temp_dir2.exists() -# ============================================================================= -# Tests: Send Pipeline File -# ============================================================================= - - -class TestSendPipelineFile: - """Tests for _send_pipeline_file method.""" - - def test_send_pipeline_file_success(self, input_folder: Path): - """Test successful sending of pipeline file.""" - backend = CaptureBackend() - - config = DispatchConfig(backends={"copy": backend}, settings={}) - orchestrator = DispatchOrchestrator(config) - - folder = {"process_backend_copy": True} - - result = orchestrator._send_pipeline_file( - str(input_folder / "test.edi"), folder - ) - - assert result - assert len(backend.received) == 1 - - def test_send_pipeline_file_no_backends(self, input_folder: Path): - """Test sending with no enabled backends.""" - config = DispatchConfig(backends={}, settings={}) - orchestrator = DispatchOrchestrator(config) - - folder = {} - - result = orchestrator._send_pipeline_file( - str(input_folder / "test.edi"), folder - ) - - assert not result - - def test_send_pipeline_file_partial_failure(self, input_folder: Path): - """Test sending with partial backend failure.""" - backend1 = CaptureBackend() - backend2 = FailingBackend() - - config = DispatchConfig( - backends={"copy1": backend1, "copy2": backend2}, settings={} - ) - orchestrator = DispatchOrchestrator(config) - - folder = {"process_backend_copy1": True, "process_backend_copy2": True} - - result = orchestrator._send_pipeline_file( - str(input_folder / "test.edi"), folder - ) - - assert not result - - # ============================================================================= # Tests: Process Folder Pipeline Routing # =============================================================================