| """ |
| Pytest configuration and fixtures for SPARKNET tests |
| Following FAANG best practices for test infrastructure |
| """ |
|
|
| import pytest |
| import asyncio |
| import sys |
| from pathlib import Path |
| from typing import Generator, AsyncGenerator |
| from unittest.mock import MagicMock, AsyncMock |
|
|
| |
| sys.path.insert(0, str(Path(__file__).parent.parent / "src")) |
|
|
|
|
| |
| |
| |
|
|
| @pytest.fixture(scope="session") |
| def event_loop(): |
| """Create an event loop for async tests.""" |
| loop = asyncio.new_event_loop() |
| yield loop |
| loop.close() |
|
|
|
|
| |
| |
| |
|
|
| @pytest.fixture |
| def mock_ollama_client(): |
| """Mock Ollama client for unit tests.""" |
| client = MagicMock() |
| client.generate = MagicMock(return_value="Mock LLM response") |
| client.chat = MagicMock(return_value="Mock chat response") |
| client.list_models = MagicMock(return_value=["llama3.2:latest", "qwen2.5:14b"]) |
| return client |
|
|
|
|
| @pytest.fixture |
| def mock_langchain_client(): |
| """Mock LangChain Ollama client for unit tests.""" |
| client = MagicMock() |
|
|
| |
| mock_llm = MagicMock() |
| mock_llm.invoke = MagicMock(return_value=MagicMock(content="Mock response")) |
| mock_llm.ainvoke = AsyncMock(return_value=MagicMock(content="Mock async response")) |
|
|
| client.get_llm = MagicMock(return_value=mock_llm) |
| client.get_embeddings = MagicMock(return_value=MagicMock()) |
|
|
| return client |
|
|
|
|
| |
| |
| |
|
|
| @pytest.fixture |
| def mock_memory_agent(): |
| """Mock memory agent for unit tests.""" |
| agent = MagicMock() |
| agent.retrieve_relevant_context = AsyncMock(return_value=[]) |
| agent.store_episode = AsyncMock(return_value=None) |
| agent.search_stakeholders = AsyncMock(return_value=[]) |
| return agent |
|
|
|
|
| @pytest.fixture |
| def mock_planner_agent(): |
| """Mock planner agent for unit tests.""" |
| from src.agents.base_agent import Task |
|
|
| agent = MagicMock() |
|
|
| mock_task = Task( |
| id="test_task", |
| description="Test task", |
| status="completed", |
| result={ |
| "task_graph": MagicMock( |
| subtasks={}, |
| get_execution_order=MagicMock(return_value=[]) |
| ), |
| "execution_order": [], |
| "total_subtasks": 0, |
| } |
| ) |
| agent.process_task = AsyncMock(return_value=mock_task) |
|
|
| return agent |
|
|
|
|
| @pytest.fixture |
| def mock_critic_agent(): |
| """Mock critic agent for unit tests.""" |
| from src.agents.base_agent import Task |
|
|
| agent = MagicMock() |
|
|
| mock_validation = MagicMock( |
| overall_score=0.9, |
| issues=[], |
| suggestions=[], |
| dimension_scores={"completeness": 0.9, "clarity": 0.9} |
| ) |
|
|
| mock_task = Task( |
| id="test_task", |
| description="Test task", |
| status="completed", |
| result=mock_validation |
| ) |
| agent.process_task = AsyncMock(return_value=mock_task) |
| agent.get_feedback_for_iteration = MagicMock(return_value="Good quality output") |
|
|
| return agent |
|
|
|
|
| |
| |
| |
|
|
| @pytest.fixture |
| def sample_patent_analysis(): |
| """Sample patent analysis result for testing.""" |
| return { |
| "title": "Test Patent: Novel AI System", |
| "abstract": "A system for processing natural language using transformers", |
| "claims": [ |
| "Claim 1: A method for natural language processing", |
| "Claim 2: A system implementing the method of claim 1" |
| ], |
| "trl_level": 4, |
| "innovation_domains": ["Artificial Intelligence", "Natural Language Processing"], |
| "key_innovations": ["Novel attention mechanism", "Efficient inference"], |
| "filing_date": "2023-01-15", |
| "patent_number": "US12345678", |
| } |
|
|
|
|
| @pytest.fixture |
| def sample_market_analysis(): |
| """Sample market analysis result for testing.""" |
| return { |
| "opportunities": [ |
| { |
| "name": "Enterprise NLP Market", |
| "market_size": 12.5, |
| "growth_rate": 0.25, |
| "relevance_score": 0.85, |
| }, |
| { |
| "name": "Conversational AI", |
| "market_size": 8.2, |
| "growth_rate": 0.32, |
| "relevance_score": 0.78, |
| }, |
| ], |
| "competitive_landscape": "Moderate competition with major players", |
| "commercialization_potential": 0.8, |
| } |
|
|
|
|
| @pytest.fixture |
| def sample_stakeholder_match(): |
| """Sample stakeholder match for testing.""" |
| return { |
| "name": "Tech Corp Inc", |
| "type": "company", |
| "domain": "Enterprise Software", |
| "relevance_score": 0.92, |
| "contact_info": { |
| "email": "licensing@techcorp.example", |
| "phone": "+1-555-0123", |
| }, |
| "match_rationale": "Strong alignment with NLP focus areas", |
| } |
|
|
|
|
| |
| |
| |
|
|
| @pytest.fixture |
| def test_config(): |
| """Test configuration dictionary.""" |
| return { |
| "gpu": { |
| "primary": 0, |
| "fallback": [1, 2, 3], |
| "max_memory_per_model": "8GB", |
| }, |
| "ollama": { |
| "host": "localhost", |
| "port": 11434, |
| "default_model": "llama3.2:latest", |
| "timeout": 300, |
| }, |
| "memory": { |
| "vector_store": "chromadb", |
| "embedding_model": "nomic-embed-text:latest", |
| "max_context_length": 4096, |
| "persist_directory": "/tmp/sparknet_test_memory", |
| }, |
| "workflow": { |
| "max_parallel_tasks": 5, |
| "task_timeout": 600, |
| "retry_attempts": 3, |
| }, |
| } |
|
|
|
|
| |
| |
| |
|
|
| @pytest.fixture(autouse=True) |
| def cleanup_test_files(): |
| """Clean up any test files after each test.""" |
| yield |
|
|
| |
| test_output_dir = Path("/tmp/sparknet_test_outputs") |
| if test_output_dir.exists(): |
| import shutil |
| shutil.rmtree(test_output_dir, ignore_errors=True) |
|
|
|
|
| |
| |
| |
|
|
| def pytest_configure(config): |
| """Configure pytest markers.""" |
| config.addinivalue_line( |
| "markers", "slow: mark test as slow (deselect with '-m \"not slow\"')" |
| ) |
| config.addinivalue_line( |
| "markers", "integration: mark test as integration test" |
| ) |
| config.addinivalue_line( |
| "markers", "gpu: mark test as requiring GPU" |
| ) |
| config.addinivalue_line( |
| "markers", "ollama: mark test as requiring Ollama server" |
| ) |
|
|
