Spaces:
Sleeping
Sleeping
| """ | |
| Pytest Configuration and Fixtures for Medium-MCP | |
| This module provides shared fixtures for unit, integration, and E2E tests. | |
| Following pytest-asyncio best practices from official documentation. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| from pathlib import Path | |
| from typing import TYPE_CHECKING, Any, AsyncGenerator, Generator | |
| from unittest.mock import AsyncMock, MagicMock | |
| import pytest | |
| import pytest_asyncio | |
| if TYPE_CHECKING: | |
| from src.service import ScraperService | |
| # Add src to path for imports | |
| sys.path.insert(0, str(Path(__file__).parent.parent)) | |
| # Fixtures directory | |
| FIXTURES_DIR = Path(__file__).parent / "fixtures" | |
| # ============================================================================= | |
| # SAMPLE DATA FIXTURES | |
| # ============================================================================= | |
| def sample_article() -> dict[str, Any]: | |
| """Load sample free article fixture.""" | |
| fixture_path = FIXTURES_DIR / "articles" / "free_article.json" | |
| if fixture_path.exists(): | |
| return json.loads(fixture_path.read_text()) | |
| # Fallback inline fixture | |
| return { | |
| "url": "https://medium.com/@testuser/test-article-abc123def456", | |
| "title": "Test Article Title", | |
| "subtitle": "A comprehensive test subtitle", | |
| "author": { | |
| "name": "Test Author", | |
| "username": "testuser", | |
| "bio": "Test bio", | |
| "avatar_url": "https://example.com/avatar.jpg", | |
| }, | |
| "publication": "Test Publication", | |
| "tags": ["python", "testing", "mcp"], | |
| "reading_time": 5, | |
| "claps": 100, | |
| "is_paywalled": False, | |
| "markdown_content": "# Test Article\n\nThis is test content.", | |
| "html_content": "<h1>Test Article</h1><p>This is test content.</p>", | |
| "word_count": 500, | |
| } | |
| def paywalled_article() -> dict[str, Any]: | |
| """Load paywalled article fixture.""" | |
| fixture_path = FIXTURES_DIR / "articles" / "paywalled.json" | |
| if fixture_path.exists(): | |
| return json.loads(fixture_path.read_text()) | |
| return { | |
| "url": "https://medium.com/@premium/premium-article-xyz789", | |
| "title": "Premium Content Article", | |
| "author": {"name": "Premium Author", "username": "premium"}, | |
| "is_paywalled": True, | |
| "markdown_content": "This content requires a Medium membership...", | |
| } | |
| def sample_html() -> str: | |
| """Load sample Medium HTML page.""" | |
| fixture_path = FIXTURES_DIR / "html" / "medium_page.html" | |
| if fixture_path.exists(): | |
| return fixture_path.read_text() | |
| return """ | |
| <!DOCTYPE html> | |
| <html> | |
| <head><title>Test Article</title></head> | |
| <body> | |
| <article> | |
| <h1>Test Article Title</h1> | |
| <p>Article content here.</p> | |
| </article> | |
| <script id="__APOLLO_STATE__" type="application/json"> | |
| {"ROOT_QUERY": {"post": {"title": "Test"}}} | |
| </script> | |
| </body> | |
| </html> | |
| """ | |
| def graphql_success_response() -> dict[str, Any]: | |
| """GraphQL API success response fixture.""" | |
| return { | |
| "data": { | |
| "post": { | |
| "id": "abc123def456", | |
| "title": "Test Article from GraphQL", | |
| "content": {"bodyModel": {"paragraphs": []}}, | |
| "creator": {"name": "Test Author", "username": "testuser"}, | |
| } | |
| } | |
| } | |
| # ============================================================================= | |
| # MOCK CLIENT FIXTURES | |
| # ============================================================================= | |
| async def mock_httpx_client() -> AsyncGenerator[AsyncMock, None]: | |
| """Mock httpx.AsyncClient with predefined responses.""" | |
| client = AsyncMock() | |
| # Default successful response | |
| mock_response = MagicMock() | |
| mock_response.status_code = 200 | |
| mock_response.text = "<html><body><h1>Test</h1></body></html>" | |
| mock_response.content = b"<html><body><h1>Test</h1></body></html>" | |
| mock_response.headers = {"content-type": "text/html; charset=utf-8"} | |
| mock_response.raise_for_status = MagicMock() | |
| client.get.return_value = mock_response | |
| client.post.return_value = mock_response | |
| client.is_closed = False | |
| client.aclose = AsyncMock() | |
| yield client | |
| def mock_groq_client() -> MagicMock: | |
| """Mock Groq LLM client for report generation.""" | |
| client = MagicMock() | |
| mock_completion = MagicMock() | |
| mock_completion.choices = [ | |
| MagicMock(message=MagicMock(content="Generated summary of the article.")) | |
| ] | |
| client.chat.completions.create.return_value = mock_completion | |
| return client | |
| def mock_elevenlabs_client() -> MagicMock: | |
| """Mock ElevenLabs TTS client for audio generation.""" | |
| client = MagicMock() | |
| # Mock streaming audio response | |
| client.text_to_speech.convert.return_value = iter([ | |
| b"audio_chunk_1", | |
| b"audio_chunk_2", | |
| b"audio_chunk_3", | |
| ]) | |
| return client | |
| # ============================================================================= | |
| # SERVICE FIXTURES | |
| # ============================================================================= | |
| async def mock_scraper_service() -> AsyncGenerator[MagicMock, None]: | |
| """Mock ScraperService for integration tests without real browser.""" | |
| service = MagicMock() | |
| service._initialized = True | |
| service._playwright = AsyncMock() | |
| service._browser = AsyncMock() | |
| service._workers = [] | |
| # Mock the main scraping method | |
| async def mock_scrape(url: str, **kwargs: Any) -> dict[str, Any]: | |
| return { | |
| "url": url, | |
| "title": "Mocked Article", | |
| "markdown_content": "# Mocked Content", | |
| "author": {"name": "Mock Author", "username": "mock"}, | |
| "is_paywalled": False, | |
| } | |
| service.scrape_article = AsyncMock(side_effect=mock_scrape) | |
| service.scrape_search = AsyncMock(return_value=[]) | |
| service.scrape_tag = AsyncMock(return_value=[]) | |
| service.close = AsyncMock() | |
| service.ensure_initialized = AsyncMock() | |
| yield service | |
| # ============================================================================= | |
| # DATABASE FIXTURES | |
| # ============================================================================= | |
| def temp_db_path(tmp_path: Path) -> Path: | |
| """Provide a temporary database path for testing.""" | |
| return tmp_path / "test_articles.db" | |
| # ============================================================================= | |
| # PLAYWRIGHT FIXTURES (for E2E tests) | |
| # ============================================================================= | |
| def gradio_url() -> str: | |
| """Base URL for Gradio app in E2E tests.""" | |
| return "http://localhost:7860" | |
| # ============================================================================= | |
| # HELPER FUNCTIONS | |
| # ============================================================================= | |
| def load_fixture(name: str) -> dict[str, Any] | str: | |
| """Load a fixture file by name.""" | |
| # Try JSON first | |
| json_path = FIXTURES_DIR / f"{name}.json" | |
| if json_path.exists(): | |
| return json.loads(json_path.read_text()) | |
| # Try HTML | |
| html_path = FIXTURES_DIR / f"{name}.html" | |
| if html_path.exists(): | |
| return html_path.read_text() | |
| raise FileNotFoundError(f"Fixture not found: {name}") | |