import unittest import logging import os import time import json import psycopg2 import redis import requests from typing import Dict, Any, List, Tuple # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Load environment variables (or use defaults) POSTGRES_HOST = os.environ.get("POSTGRES_HOST", "postgresql-genesis-u50607.vm.elestio.app") POSTGRES_PORT = int(os.environ.get("POSTGRES_PORT", "25432")) POSTGRES_USER = os.environ.get("POSTGRES_USER", "genesis") POSTGRES_PASSWORD = os.environ.get("POSTGRES_PASSWORD", "genesis") POSTGRES_DB = os.environ.get("POSTGRES_DB", "genesis") REDIS_HOST = os.environ.get("REDIS_HOST", "redis-genesis-u50607.vm.elestio.app") REDIS_PORT = int(os.environ.get("REDIS_PORT", "26379")) QDRANT_HOST = os.environ.get("QDRANT_HOST", "qdrant-b3knu-u50607.vm.elestio.app") QDRANT_PORT = int(os.environ.get("QDRANT_PORT", "6333")) AIVA_OLLAMA_HOST = os.environ.get("AIVA_OLLAMA_HOST", "localhost") AIVA_OLLAMA_PORT = int(os.environ.get("AIVA_OLLAMA_PORT", "23405")) class GenesisE2ETestSuite(unittest.TestCase): """ End-to-end test suite for Genesis workflows. """ def setUp(self) -> None: """ Set up resources for the tests. Establish connections to the database, Redis, Qdrant, and AIVA Ollama. Log any connection errors. """ self.postgres_conn = None self.redis_conn = None try: self.postgres_conn = psycopg2.connect( host=POSTGRES_HOST, port=POSTGRES_PORT, user=POSTGRES_USER, password=POSTGRES_PASSWORD, database=POSTGRES_DB ) self.postgres_conn.autocommit = True # Set autocommit to True logging.info("Connected to PostgreSQL") except psycopg2.Error as e: logging.error(f"Failed to connect to PostgreSQL: {e}") try: self.redis_conn = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True) self.redis_conn.ping() # Check the connection logging.info("Connected to Redis") except redis.exceptions.ConnectionError as e: logging.error(f"Failed to connect to Redis: {e}") self.qdrant_url = f"http://{QDRANT_HOST}:{QDRANT_PORT}" self.aiva_ollama_url = f"http://{AIVA_OLLAMA_HOST}:{AIVA_OLLAMA_PORT}" logging.info("E2E Test Suite Setup Complete") def tearDown(self) -> None: """ Tear down resources after the tests. Close database and Redis connections. """ if self.postgres_conn: self.postgres_conn.close() logging.info("PostgreSQL connection closed") if self.redis_conn: self.redis_conn.close() logging.info("Redis connection closed") logging.info("E2E Test Suite Teardown Complete") def test_end_to_end_workflow(self) -> None: """ Tests a complete end-to-end workflow, including: - Task submission - Processing - Result retrieval - Data flow validation - Workflow health reporting """ logging.info("Starting end-to-end workflow test...") # 1. Define a sample workflow (replace with a real workflow definition) workflow_definition = { "name": "Sample E2E Workflow", "description": "A simple workflow for testing purposes.", "tasks": [ { "name": "Task 1", "type": "python", "code": "def process(data):\n return data + 1", "input": 10 }, { "name": "Task 2", "type": "python", "code": "def process(data):\n return data * 2", "input": "$Task 1" # Input from Task 1 } ] } # 2. Submit the workflow (replace with actual submission logic) workflow_id = self._submit_workflow(workflow_definition) self.assertIsNotNone(workflow_id, "Workflow submission failed") logging.info(f"Workflow submitted with ID: {workflow_id}") # 3. Monitor the workflow (replace with actual monitoring logic) workflow_status = self._monitor_workflow(workflow_id) self.assertEqual(workflow_status, "completed", "Workflow did not complete successfully") logging.info(f"Workflow status: {workflow_status}") # 4. Retrieve the results (replace with actual result retrieval logic) results = self._retrieve_results(workflow_id) self.assertIsNotNone(results, "Failed to retrieve results") logging.info(f"Workflow results: {results}") # 5. Validate the data flow (replace with actual validation logic) expected_result = 22 # (10 + 1) * 2 self.assertEqual(results["Task 2"], expected_result, "Data flow validation failed") logging.info("Data flow validation successful") # 6. Report workflow health (replace with actual health reporting logic) workflow_health = self._report_workflow_health(workflow_id) self.assertTrue(workflow_health["success"], "Workflow health check failed") logging.info(f"Workflow health: {workflow_health}") logging.info("End-to-end workflow test completed successfully.") def _submit_workflow(self, workflow_definition: Dict[str, Any]) -> str: """ Submits a workflow definition to the system. (Replace with actual submission logic.) """ # Placeholder: In a real system, this would involve sending a request # to a workflow management service. # For this example, we'll just simulate a successful submission. return "workflow_123" def _monitor_workflow(self, workflow_id: str) -> str: """ Monitors the status of a workflow until it completes. (Replace with actual monitoring logic.) """ # Placeholder: In a real system, this would involve polling a workflow # management service for status updates. # For this example, we'll just simulate a successful completion after a delay. time.sleep(5) # Simulate processing time return "completed" def _retrieve_results(self, workflow_id: str) -> Dict[str, Any]: """ Retrieves the results of a completed workflow. (Replace with actual result retrieval logic.) """ # Placeholder: In a real system, this would involve querying a data store # for the workflow results. # For this example, we'll just return a hardcoded result. return {"Task 1": 11, "Task 2": 22} def _report_workflow_health(self, workflow_id: str) -> Dict[str, bool]: """ Reports the health of a workflow. (Replace with actual health reporting logic.) """ # Placeholder: In a real system, this would involve checking various # metrics related to the workflow execution. # For this example, we'll just return a successful health report. return {"success": True} if __name__ == '__main__': unittest.main()