""" Pytest configuration and fixtures for GumYum NPC client tests """ import os import sys import pytest import pytest_asyncio import asyncio import uuid from typing import Dict, Any, Optional # Add parent directory to path for imports sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) # Import both client versions from gumyum_npc_client import GumYumClient as AsyncGumYumClient from gumyum_npc_sync import GumYumClient as SyncGumYumClient @pytest.fixture(scope="session") def api_base_url() -> str: """Base URL for the GumYum NPC API""" return os.getenv("GUMYUM_API_URL", "http://localhost:6543") @pytest.fixture(scope="function") def test_credentials() -> Dict[str, str]: """Test user credentials with unique username""" unique_id = str(uuid.uuid4())[:8] return { "username": os.getenv("TEST_USERNAME", f"pytest_user_{unique_id}"), "password": os.getenv("TEST_PASSWORD", "pytest_password_123"), "email": os.getenv("TEST_EMAIL", f"pytest_{unique_id}@example.com"), } @pytest_asyncio.fixture async def async_client(api_base_url: str) -> AsyncGumYumClient: """Async client fixture with automatic cleanup""" client = AsyncGumYumClient(api_base_url, timeout=60.0) try: yield client finally: await client.close() @pytest.fixture def sync_client(api_base_url: str) -> SyncGumYumClient: """Sync client fixture with automatic cleanup""" client = SyncGumYumClient(api_base_url, timeout=60.0) try: yield client finally: client.close() @pytest_asyncio.fixture async def authenticated_async_client( async_client: AsyncGumYumClient, test_credentials: Dict[str, str] ) -> AsyncGumYumClient: """Async client that's already authenticated""" # Register new user (since we use unique usernames now) await async_client.auth.register( test_credentials["username"], test_credentials["password"], test_credentials["email"], ) return async_client @pytest.fixture def authenticated_sync_client( sync_client: SyncGumYumClient, test_credentials: Dict[str, str] ) -> SyncGumYumClient: """Sync client that's already authenticated""" # Register new user (since we use unique usernames now) sync_client.auth.register( test_credentials["username"], test_credentials["password"], test_credentials["email"], ) return sync_client @pytest.fixture(scope="session") def event_loop(): """Create an event loop for async tests""" loop = asyncio.get_event_loop_policy().new_event_loop() yield loop loop.close() @pytest.fixture def sample_universe_data() -> Dict[str, Any]: """Sample universe data for testing""" return { "theme_info": { "name": "Test Universe", "description": "A test universe for functional testing", "version": "1.0", "author": "pytest", "is_public": False, "builtin": False, }, "character_names": { "human_male": ["TestMale1", "TestMale2"], "human_female": ["TestFemale1", "TestFemale2"], }, "professions": { "tavern": ["Bartender", "Server"], "market": ["Merchant", "Vendor"], }, "world_locations": { "tavern": { "name": "The Test Tavern", "description": "A cozy testing establishment", }, "market": { "name": "Test Market Square", "description": "A bustling marketplace for tests", }, }, "mood_categories": { "neutral": { "moods": ["calm", "focused"], "description": "Balanced emotional state", } }, } @pytest_asyncio.fixture async def test_universe_id( authenticated_async_client: AsyncGumYumClient, sample_universe_data: Dict[str, Any] ) -> str: """Create a test universe and return its ID""" # Since universe creation endpoint has issues, copy a public universe instead universes = await authenticated_async_client.universes.list_public() if not universes: pytest.skip("No public universes available for testing") universe_id = await authenticated_async_client.universes.copy_public( universes[0].id, "Test Universe for Async Tests" ) return universe_id @pytest.fixture def sync_test_universe_id( authenticated_sync_client: SyncGumYumClient, sample_universe_data: Dict[str, Any] ) -> str: """Create a test universe for sync client and return its ID""" # Since universe creation endpoint has issues, copy a public universe instead universes = authenticated_sync_client.universes.list_public() if not universes: pytest.skip("No public universes available for testing") universe_id = authenticated_sync_client.universes.copy_public( universes[0].id, "Test Universe for Sync Tests" ) return universe_id @pytest_asyncio.fixture async def test_npc( authenticated_async_client: AsyncGumYumClient, test_universe_id: str ): """Spawn a test NPC""" npc = await authenticated_async_client.npc.spawn( test_universe_id, seed=12345, npc_id=123456789 ) return npc @pytest.fixture def sync_test_npc( authenticated_sync_client: SyncGumYumClient, sync_test_universe_id: str ): """Spawn a test NPC for sync client""" npc = authenticated_sync_client.npc.spawn( sync_test_universe_id, seed=12345, npc_id=123456789 ) return npc # Skip tests if API is not available def pytest_configure(config): """Configure pytest markers""" config.addinivalue_line( "markers", "integration: mark test as requiring live API endpoint" ) config.addinivalue_line( "markers", "requires_universe: mark test as requiring working universe management endpoints", ) def pytest_collection_modifyitems(config, items): """Skip integration tests if API is not available""" if config.getoption("--skip-integration"): skip_integration = pytest.mark.skip(reason="--skip-integration option given") for item in items: if "integration" in item.keywords: item.add_marker(skip_integration) # Temporarily disable NPC test skipping to debug the actual issue # skip_npc = pytest.mark.skip(reason="NPC spawning/chat tests failing due to server cache issues") # npc_test_names = [ # "test_spawn_npc_deterministic", # "test_spawn_npc_auto_increment", # "test_spawn_with_location_filter", # "test_npc_crud_operations", # "test_simple_chat_completion", # "test_chat_completion_full_response", # "test_streaming_chat_completion", # "test_conversation_with_history", # "test_chat_with_npc_params", # "test_personality_consistency", # "test_concurrent_requests", # "test_concurrent_requests_threaded", # "test_streaming_response_timing", # "test_streaming_vs_non_streaming_consistency" # ] # for item in items: # if item.name in npc_test_names: # item.add_marker(skip_npc) def pytest_addoption(parser): """Add command line options""" parser.addoption( "--skip-integration", action="store_true", default=False, help="Skip integration tests that require live API", ) parser.addoption( "--api-url", action="store", default="http://localhost:6543", help="GumYum API base URL for testing", ) # Health check utility async def check_api_health(base_url: str) -> bool: """Check if the API is available""" try: client = AsyncGumYumClient(base_url, timeout=5.0) health = await client.health_check() await client.close() return health.get("status") == "healthy" except Exception: return False def check_api_health_sync(base_url: str) -> bool: """Check if the API is available (sync version)""" try: client = SyncGumYumClient(base_url, timeout=5.0) health = client.health_check() client.close() return health.get("status") == "healthy" except Exception: return False @pytest_asyncio.fixture(autouse=True) async def skip_if_api_unavailable(request, api_base_url: str): """Skip tests if API is not available""" if "integration" in request.keywords: if not await check_api_health(api_base_url): pytest.skip(f"API not available at {api_base_url}")