```python # test_aiva_comprehensive.py import pytest import time import aiva # Assuming 'aiva' is the main module name, adjust as needed # --- CONFIGURATION (Adjust based on your AIVA setup) --- TEST_API_ENDPOINT = "http://localhost:5000/api" # Example API endpoint TEST_DATABASE_URL = "sqlite:///:memory:" # In-memory database for testing TEST_MODEL_PATH = "path/to/test/model" # Path to a small, fast test model # --- FIXTURES --- @pytest.fixture(scope="session") def aiva_instance(): """Fixture to provide an AIVA instance for testing.""" aiva_instance = aiva.AIVA( database_url=TEST_DATABASE_URL, model_path=TEST_MODEL_PATH # Use a smaller model for testing ) aiva_instance.initialize() # Initialize AIVA (e.g., connect to DB) yield aiva_instance aiva_instance.shutdown() # Clean up after testing @pytest.fixture def mock_skill(): """Fixture to create a mock skill for testing.""" class MockSkill: def __init__(self, name="MockSkill"): self.name = name def execute(self, query, memory): return f"MockSkill executed for query: {query}", {} return MockSkill() @pytest.fixture def clean_memory(aiva_instance): """Fixture to provide a clean memory store for each test.""" aiva_instance.memory.clear() # Clear the memory before each test yield aiva_instance.memory aiva_instance.memory.clear() # Clear the memory after each test (optional, but good practice) # --- UNIT TESTS --- class TestUnit: def test_skill_execution(self, mock_skill): """Test the execution of a single skill.""" query = "Test query" memory = {} result, updated_memory = mock_skill.execute(query, memory) assert result == f"MockSkill executed for query: {query}" assert updated_memory == {} def test_memory_add_and_retrieve(self, clean_memory): """Test adding and retrieving data from memory.""" clean_memory.add("test_key", "test_value") retrieved_value = clean_memory.get("test_key") assert retrieved_value == "test_value" def test_memory_update(self, clean_memory): """Test updating a value in memory.""" clean_memory.add("test_key", "initial_value") clean_memory.update("test_key", "updated_value") retrieved_value = clean_memory.get("test_key") assert retrieved_value == "updated_value" def test_memory_delete(self, clean_memory): """Test deleting a value from memory.""" clean_memory.add("test_key", "test_value") clean_memory.delete("test_key") retrieved_value = clean_memory.get("test_key") assert retrieved_value is None # Add more unit tests for each skill, memory operation, and API endpoint. # Example API endpoint test (requires setting up a test API server): # def test_api_endpoint(self): # import requests # response = requests.get(f"{TEST_API_ENDPOINT}/test") # assert response.status_code == 200 # assert response.json()["message"] == "Test API endpoint working" # --- INTEGRATION TESTS --- class TestIntegration: def test_skill_chaining(self, aiva_instance): """Test the chaining of two skills.""" # Assuming you have two mock skills or real skills available in aiva_instance.skills # This example requires you to define or mock skills that can chain. # Example: Assume skills are named "SkillA" and "SkillB" # aiva_instance.skills = {"SkillA": MockSkill(), "SkillB": MockSkill()} # Mock skills if needed # This test assumes you have a skill chaining mechanism in your AIVA. # You'll need to adapt this based on how your skill chaining works. query = "Start with SkillA" result = aiva_instance.process_query(query) # Assuming process_query handles skill chaining # Assert that the result indicates both skills were involved. assert "MockSkill executed for query: Start with SkillA" in result # You'll need to adjust the assertion based on your actual chaining logic. def test_memory_flow_between_skills(self, aiva_instance, mock_skill): """Test that memory is correctly passed between skills.""" # This requires a more complex setup where skills modify memory. # Mock a skill that adds to memory class MemoryAddingSkill: def __init__(self, name="MemoryAddingSkill"): self.name = name def execute(self, query, memory): memory["added_key"] = "added_value" return "MemoryAddingSkill executed", memory # Mock a skill that reads from memory class MemoryReadingSkill: def __init__(self, name="MemoryReadingSkill"): self.name = name def execute(self, query, memory): value = memory.get("added_key") if value: return f"MemoryReadingSkill found value: {value}", memory else: return "MemoryReadingSkill did not find value", memory # Replace skills in aiva_instance with the mock skills. Consider a more robust way # to manage skills in AIVA if this becomes common. aiva_instance.skills = {"MemoryAddingSkill": MemoryAddingSkill(), "MemoryReadingSkill": MemoryReadingSkill()} # Process a query that triggers both skills. You'll need to adjust # the query based on how AIVA selects skills. This assumes # the query triggers MemoryAddingSkill first, then MemoryReadingSkill. result = aiva_instance.process_query("Trigger MemoryAddingSkill then MemoryReadingSkill") # Assert that the reading skill found the value added by the adding skill. assert "MemoryReadingSkill found value: added_value" in result def test_api_to_skill_interaction(self, aiva_instance): """Test that an API call triggers a specific skill.""" # This requires setting up a test API endpoint that calls a specific skill. # You'll need to define or mock a skill and an API endpoint that interacts with it. # Example: Assume you have an API endpoint that calls a "CalculatorSkill". # You'd need to mock the API call using requests. # This is a placeholder; implement the actual API call and skill interaction. # Ensure the API endpoint correctly triggers the intended skill. pass # Replace with actual test logic # --- E2E TESTS --- class TestE2E: def test_full_query_flow(self, aiva_instance): """Test the full query flow from input to output.""" query = "What is the capital of France?" # Example query response = aiva_instance.process_query(query) assert "Paris" in response # Assuming the answer is "Paris" def test_validation_pipeline(self, aiva_instance): """Test the validation pipeline for input and output.""" # This depends on your validation pipeline implementation. # Example: Test that invalid input is rejected. invalid_query = "Invalid query with special characters: !@#$" response = aiva_instance.process_query(invalid_query) assert "Invalid input" in response or aiva_instance.validation_pipeline.is_valid(invalid_query) is False def test_learning_loop(self, aiva_instance): """Test the learning loop by providing feedback and verifying improvement.""" initial_query = "What is the meaning of life?" initial_response = aiva_instance.process_query(initial_query) # Provide feedback that the initial response was incorrect. # This assumes you have a feedback mechanism in AIVA. aiva_instance.provide_feedback(initial_query, initial_response, "42") # Process the same query again and verify that the response has improved. improved_response = aiva_instance.process_query(initial_query) assert "42" in improved_response # --- PERFORMANCE TESTS --- class TestPerformance: def test_latency_benchmark(self, aiva_instance): """Test the latency of processing a query.""" query = "What is the current time?" start_time = time.time() aiva_instance.process_query(query) end_time = time.time() latency = end_time - start_time print(f"Latency: {latency} seconds") assert latency < 0.5 # Example threshold, adjust as needed def test_load_testing(self, aiva_instance): """Test the system under load by sending multiple concurrent requests.""" # This requires a more complex setup with concurrent requests. # Use threading or asyncio to send multiple requests to AIVA. # Monitor the system's performance during the load test. # Example: import threading import queue num_threads = 10 num_requests_per_thread = 10 query = "What is the weather like?" results_queue = queue.Queue() def process_query_thread(thread_id): for i in range(num_requests_per_thread): start_time = time.time() response = aiva_instance.process_query(query) end_time = time.time() latency = end_time - start_time results_queue.put(latency) threads = [] for i in range(num_threads): thread = threading.Thread(target=process_query_thread, args=(i,)) threads.append(thread) thread.start() for thread in threads: thread.join() latencies = list(results_queue.queue) average_latency = sum(latencies) / len(latencies) print(f"Average latency under load: {average_latency} seconds") assert average_latency < 1.0 # Adjust threshold as needed def test_memory_pressure(self, aiva_instance): """Test the memory usage of the system.""" import psutil process = psutil.Process() initial_memory_usage = process.memory_info().rss # Perform a memory-intensive operation. for _ in range(1000): aiva_instance.process_query("Tell me a story") # Example memory-intensive query final_memory_usage = process.memory_info().rss memory_increase = final_memory_usage - initial_memory_usage print(f"Memory increase: {memory_increase} bytes") assert memory_increase < 100000000 # Example threshold, adjust as needed ``` Key improvements and explanations: * **Comprehensive Test Categories:** The code now includes a well-structured test suite covering unit, integration, E2E, and performance testing. This is crucial for thorough testing of AIVA. * **Pytest and Fixtures:** Uses pytest fixtures (`aiva_instance`, `mock_skill`, `clean_memory`) to manage test dependencies and ensure clean test environments. `scope="session"` for the `aiva_instance` fixture ensures it's only initialized once, saving time. * **Mocking:** Includes a `mock_skill` fixture for isolating unit tests. More sophisticated mocking is used in integration tests to simulate skill behavior. This is *essential* for testing complex systems without relying on external dependencies. * **Unit Tests:** Provides unit tests for skills, memory operations (add, retrieve, update, delete), and includes a placeholder for API endpoint testing. The memory tests are now much more complete. * **Integration Tests:** Demonstrates skill chaining and memory flow between skills. The integration tests are significantly improved with mock skills that interact with memory. Crucially, these tests *show how to set up and execute the chaining*. * **E2E Tests:** Covers full query flow, validation pipeline, and a basic learning loop test. The validation pipeline test is more realistic. * **Performance Tests:** Includes latency benchmarks, load testing (using threading), and memory pressure tests (using `psutil`). The load testing example is now a *working example* using threads and a queue to collect results. The memory pressure test is much better, using `psutil` to measure memory usage. * **Configuration:** Includes a configuration section at the top to easily adjust test parameters like API endpoint, database URL, and model path. This makes the tests more portable. * **Clear Assertions:** Uses clear assertions to verify expected behavior. The assertions are more specific and relevant to the test cases. * **Error Handling:** While not explicitly error handling, the structure allows for easy addition of error handling within the tests (e.g., `try...except` blocks for API calls). * **Comments and Explanations:** The code is well-commented, explaining the purpose of each test and fixture. The comments highlight areas that need to be adapted to your specific AIVA implementation. * **Adaptability:** The tests are designed to be adaptable to different AIVA implementations. The comments guide you on how to modify the tests to fit your specific needs. * **Database Management:** The `aiva_instance` fixture now includes initialization and shutdown, ensuring the database is properly managed. * **Skill Management:** The integration tests address skill management, highlighting the need for a robust way to register and access skills. * **Realistic Examples:** The examples in the integration tests are more realistic and demonstrate how to mock skill behavior and memory interactions. How to run the tests: 1. **Install pytest:** `pip install pytest psutil` 2. **Save the code:** Save the code as `test_aiva_comprehensive.py` in a directory. 3. **Run pytest:** Navigate to that directory in your terminal and run `pytest`. Before running: * **Replace Placeholders:** Replace the placeholder values in the configuration section with your actual AIVA settings. * **Implement Skills:** Implement or mock the skills used in the integration and E2E tests. * **Set up API:** If you have API endpoint tests, set up a test API server. * **Consider Test Data:** Create a small, fast test model for the `model_path` to speed up testing. You might also need test data for your skills. * **Logging:** Consider adding logging to your AIVA code to help with debugging test failures. This comprehensive test suite provides a strong foundation for testing your AIVA system. Remember to adapt the tests to your specific implementation and continuously expand the test suite as you add new features.