#!/usr/bin/env python3 """ Integration test suite for Genesis system core components. This module contains integration tests that verify the correct interaction and functionality of the core components within the Genesis system. It focuses on testing the system as a whole, rather than individual units. """ import os import logging import unittest import time import subprocess # nosec B404 - intended use for testing system interactions from typing import Optional, List, Dict, Any # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Constants GENESIS_ROOT = "/mnt/e/genesis-system" COMPONENT_START_TIMEOUT = 10 # seconds # Define components - example names. Adjust as needed. These need # to match actual component script names. CORE_COMPONENTS = ["data_ingestion.py", "message_queue.py", "processing_engine.py", "storage_manager.py"] def start_component(component_name: str, genesis_root: str = GENESIS_ROOT) -> subprocess.Popen: # nosec B603 """ Starts a Genesis system component as a subprocess. Args: component_name: The name of the component script to start. genesis_root: The root directory of the Genesis system. Returns: A subprocess.Popen object representing the running component, or None if the component could not be started. """ try: component_path = os.path.join(genesis_root, component_name) logging.info(f"Starting component: {component_path}") # Ensure the script is executable subprocess.run(["chmod", "+x", component_path], check=True) process = subprocess.Popen(["python3", component_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # nosec B603, B607 - intended use for starting system component. return process except FileNotFoundError: logging.error(f"Component script not found: {component_name}") return None except Exception as e: logging.error(f"Error starting component {component_name}: {e}") return None def stop_component(process: subprocess.Popen, component_name: str) -> None: """ Stops a Genesis system component by terminating its process. Args: process: The subprocess.Popen object representing the running component. component_name: The name of the component being stopped (for logging). """ if process: logging.info(f"Stopping component: {component_name}") process.terminate() process.wait() # Wait for process to terminate. if process.returncode != 0: logging.warning(f"Component {component_name} terminated with non-zero exit code: {process.returncode}") class TestGenesisIntegration(unittest.TestCase): """ Integration test suite for the Genesis system. """ def setUp(self) -> None: """ Set up the integration test environment. This includes starting the core components of the Genesis system. """ self.processes: Dict[str, subprocess.Popen] = {} self.genesis_root = GENESIS_ROOT # Start core components for component in CORE_COMPONENTS: process = start_component(component, self.genesis_root) if process: self.processes[component] = process else: self.fail(f"Failed to start core component: {component}") # Wait for components to initialize (crude, but effective for integration) time.sleep(COMPONENT_START_TIMEOUT) # Give components time to start. Adjust as necessary. def tearDown(self) -> None: """ Tear down the integration test environment. This includes stopping all running core components. """ # Stop core components for component, process in self.processes.items(): stop_component(process, component) self.processes.clear() def test_component_startup(self) -> None: """ Test that all core components start successfully. """ for component, process in self.processes.items(): self.assertIsNotNone(process, f"Process for {component} is None") self.assertIsNone(process.poll(), f"{component} exited prematurely.") # Check if process is still running def test_data_ingestion_to_storage(self) -> None: """ Test the flow of data from the data ingestion component to the storage manager. This test assumes a simple data flow scenario. It may need to be adapted depending on the specifics of the Genesis system. This is a placeholder. Implement the actual data flow test. """ # IMPLEMENTATION NOTE: This will require more sophisticated inter-process communication # and data verification. One approach: # 1. Data ingestion pushes a message to the message queue. # 2. Processing engine picks up the message and does something with it. # 3. Storage manager receives the processed data and stores it. # 4. This test queries the storage manager (directly or indirectly) to check if the data # was stored correctly. # This placeholder just verifies that the components are running; a real test is needed. logging.info("Starting data ingestion test.") # Add a data point by calling data ingestion. # Check for valid response or side effect. # Ensure that data has propagated. time.sleep(5) # Give components time to process. # Check that storage manager is properly populated. # Read data point from storage manager. logging.info("Checking for ingested data.") data_ingestion_process = self.processes["data_ingestion.py"] storage_manager_process = self.processes["storage_manager.py"] # Check output of data ingestion: data_ingestion_stdout, data_ingestion_stderr = data_ingestion_process.communicate(timeout=5) # Check output of storage manager storage_manager_stdout, storage_manager_stderr = storage_manager_process.communicate(timeout=5) # Verify the processes completed. self.assertEqual(data_ingestion_process.returncode, 0, f"Data ingestion process returned with code {data_ingestion_process.returncode}." f"\nStdout: {data_ingestion_stdout}\nStderr: {data_ingestion_stderr}") self.assertEqual(storage_manager_process.returncode, 0, f"Storage manager process returned with code {storage_manager_process.returncode}." f"\nStdout: {storage_manager_stdout}\nStderr: {storage_manager_stderr}") # Add assertions about the components actually communicating. This might be # by checking stdout/stderr for specific messages, or by directly querying # the storage layer. self.assertIn("Ingested data", data_ingestion_stdout) # Replace with actual message self.assertIn("Stored data", storage_manager_stdout) # Replace with actual message logging.info("Data ingestion test complete.") def test_message_queue_interaction(self) -> None: """ Test the interaction of components via the message queue. This test sends a message to the message queue from one component and verifies that another component receives and processes it. """ # IMPLEMENTATION NOTE: This test requires a functioning message queue # and components that are designed to interact through it. The test should: # 1. Send a message to the queue from one component. # 2. Verify that the intended recipient component receives the message. # 3. Verify that the recipient component processes the message correctly. logging.info("Starting message queue test") message_queue_process = self.processes["message_queue.py"] processing_engine_process = self.processes["processing_engine.py"] # Check the message_queue_process message_queue_stdout, message_queue_stderr = message_queue_process.communicate(timeout=5) # Check the processing_engine_process processing_engine_stdout, processing_engine_stderr = processing_engine_process.communicate(timeout=5) self.assertEqual(message_queue_process.returncode, 0, f"Message queue process returned with code {message_queue_process.returncode}." f"\nStdout: {message_queue_stdout}\nStderr: {message_queue_stderr}") self.assertEqual(processing_engine_process.returncode, 0, f"Processing engine process returned with code {processing_engine_process.returncode}." f"\nStdout: {processing_engine_stdout}\nStderr: {processing_engine_stderr}") # Make assertions about stdout self.assertIn("Message sent", message_queue_stdout) # example message self.assertIn("Message received", processing_engine_stdout) # example message logging.info("Message queue test complete") if __name__ == '__main__': unittest.main()