Finizi B4B MCP Server

""" Integration tests with real B4B API. These tests require a running B4B API at http://localhost:8000 and use real authentication credentials. Run these tests with: pytest -m integration Skip them with: pytest -m "not integration" """ import pytest from unittest.mock import Mock from src.finizi_b4b_mcp.tools.auth import login, logout, whoami from src.finizi_b4b_mcp.tools.entities import list_entities, get_entity from src.finizi_b4b_mcp.tools.invoices import list_invoices # Real test credentials TEST_PHONE = "+84909495665" TEST_PASSWORD = "Admin123@" async def create_authenticated_context(): """Helper to create and authenticate a context for integration tests.""" ctx = Mock() ctx.session = Mock() ctx.session.metadata = {} # Test login result = await login(TEST_PHONE, TEST_PASSWORD, ctx) # Check if B4B API is available if "success" not in result or not result.get("success"): pytest.skip("B4B API not available or authentication failed") return ctx @pytest.mark.integration @pytest.mark.asyncio async def test_real_login_flow(): """Test complete authentication flow with real API.""" ctx = Mock() ctx.session = Mock() ctx.session.metadata = {} # Test login result = await login(TEST_PHONE, TEST_PASSWORD, ctx) # Should succeed or return clear error if "success" in result: assert result["success"] is True assert "user_token" in ctx.session.metadata assert ctx.session.metadata["user_token"] != "" assert "email" in result print(f"✓ Successfully logged in as {result['email']}") else: pytest.skip("B4B API not available") @pytest.mark.integration @pytest.mark.asyncio async def test_real_entity_operations(): """Test entity operations with authenticated session.""" ctx = await create_authenticated_context() # List entities result = await list_entities(ctx=ctx) # Check response structure assert "items" in result or "error" in result if "items" in result: assert isinstance(result["items"], list) assert "total" in result print(f"✓ Found {result['total']} entities") # If we have entities, test getting one if len(result["items"]) > 0: entity_id = result["items"][0]["id"] entity = await get_entity(entity_id, ctx) assert "id" in entity or "error" in entity if "id" in entity: print(f"✓ Retrieved entity: {entity.get('name', entity_id)}") else: print(f"⚠ Entity list returned error: {result['error']}") @pytest.mark.integration @pytest.mark.asyncio async def test_real_invoice_operations(): """Test invoice operations with authenticated session.""" ctx = await create_authenticated_context() # Get entities first entities = await list_entities(ctx=ctx) if "items" in entities and len(entities["items"]) > 0: entity_id = entities["items"][0]["id"] # List invoices result = await list_invoices(entity_id, ctx=ctx) assert "success" in result or "error" in result if result.get("success"): assert "data" in result print(f"✓ Successfully retrieved invoices for entity {entity_id}") if "total" in result["data"]: print(f" Found {result['data']['total']} invoices") else: print(f"⚠ Invoice list returned error: {result.get('error', 'Unknown error')}") else: print("⚠ No entities available to test invoice operations") @pytest.mark.integration @pytest.mark.asyncio async def test_real_whoami(): """Test whoami with authenticated session.""" ctx = await create_authenticated_context() result = await whoami(ctx) # Should return user info assert "success" in result or "error" in result if result.get("success"): assert "user" in result assert result["user"].get("email") is not None print(f"✓ Whoami returned: {result['user'].get('email')}") else: print(f"⚠ Whoami returned error: {result.get('error', 'Unknown error')}") @pytest.mark.integration @pytest.mark.asyncio async def test_real_logout(): """Test logout clears session.""" ctx = await create_authenticated_context() # Verify token exists assert "user_token" in ctx.session.metadata original_email = ctx.session.metadata.get("user_email", "unknown") # Logout result = await logout(ctx) # Should succeed assert result["success"] is True # Session should be cleared assert len(ctx.session.metadata) == 0 print(f"✓ Successfully logged out {original_email}") @pytest.mark.integration @pytest.mark.asyncio async def test_real_unauthorized_access(): """Test that operations fail without authentication.""" ctx = Mock() ctx.session = Mock() ctx.session.metadata = {} # Try to list entities without authentication result = await list_entities(ctx=ctx) # Should return error assert "error" in result assert ("authentication" in result["error"].lower() or "no authentication session" in result["error"].lower()) print(f"✓ Correctly rejected unauthorized access") @pytest.mark.integration @pytest.mark.asyncio async def test_real_invalid_credentials(): """Test login with invalid credentials.""" ctx = Mock() ctx.session = Mock() ctx.session.metadata = {} # Try to login with wrong password result = await login(TEST_PHONE, "WrongPassword123!", ctx) # Should fail if "success" in result: assert result["success"] is False assert "error" in result print(f"✓ Correctly rejected invalid credentials") else: pytest.skip("B4B API not available")