"""HTTP integration tests for all Pry API routers. Tests that endpoints respond with correct status codes, response shapes, error handling, CORS headers, and auth gating — all via TestClient with mocked dependencies so no real scraping occurs. """ # SPDX-License-Identifier: MIT # Copyright (c) 2026 Rug Munch Media LLC # # Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper # Licensed under MIT. See LICENSE. from __future__ import annotations from collections.abc import Iterator from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi.testclient import TestClient def _has_postgres(host: str = "127.0.0.1", port: int = 5432) -> bool: """Return True if Postgres is reachable on the given host:port. Used to skip integration tests that require a live database when running the suite locally without docker-compose up. """ import socket try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(0.25) return s.connect_ex((host, port)) == 0 except OSError: return False # ── Fixtures ───────────────────────────────────────────────────── @pytest.fixture(autouse=True) def _mock_all_deps() -> Iterator[None]: """Mock all external deps so no real scraping/parsing occurs.""" scraper_mock = MagicMock() scraper_mock.scrape = AsyncMock( return_value={ "status": "ok", "url": "https://example.com", "content": "# Mocked content", "raw_html": "Mocked", "method": "direct", } ) scraper_mock.crawl = AsyncMock(return_value=[{"url": "https://example.com", "status": "ok"}]) scraper_mock.map_urls = AsyncMock( return_value=[{"url": "https://example.com/page1"}, {"url": "https://example.com/page2"}] ) scraper_mock.detect_block = AsyncMock( return_value={ "detected": True, "type": "cloudflare", "confidence": 0.95, } ) extractor_mock = MagicMock() extractor_mock.extract = AsyncMock( return_value={ "success": True, "data": {"product_name": "Widget Pro", "price": "$29.99"}, } ) parser_mock = MagicMock() parser_mock.parse = AsyncMock( return_value={ "success": True, "data": {"title": "Test", "body": "Content"}, } ) cache_mock = MagicMock() cache_mock.get = MagicMock(return_value=None) cache_mock.set = MagicMock() advanced_mock = MagicMock() advanced_mock.extract = AsyncMock( return_value={ "success": True, "data": {"name": "Widget Pro", "price": 29.99}, } ) with patch.multiple( "deps", scraper=scraper_mock, extractor=extractor_mock, parser=parser_mock, cache=cache_mock, advanced=advanced_mock, ): yield @pytest.fixture def client() -> Iterator[TestClient]: """TestClient with x402 middleware bypassed (it would require payment).""" from api import app with ( patch( "x402_middleware.X402Middleware.__call__", lambda self, scope, receive, send: self.app(scope, receive, send), ), TestClient(app) as c, ): yield c @pytest.fixture def client_auth(client: TestClient, monkeypatch) -> TestClient: """TestClient with PRY_API_KEY set (auth required).""" monkeypatch.setattr("api.settings.api_key", "test-api-key") return client # ── Health & Readiness ─────────────────────────────────────────── class TestHealthEndpoints: """Test /health, /live, /ready — should work without auth.""" def test_health_returns_200(self, client: TestClient) -> None: """Health returns 503 when deps unavailable (normal in test).""" resp = client.get("/health") assert resp.status_code in (200, 503) body = resp.json() assert "status" in body def test_live_returns_200(self, client: TestClient) -> None: resp = client.get("/live") assert resp.status_code == 200 assert resp.json() == {"status": "alive"} @pytest.mark.integration @pytest.mark.skipif( not _has_postgres(), reason="requires running Postgres + Redis" ) def test_ready_returns_200(self, client: TestClient) -> None: resp = client.get("/ready") assert resp.status_code == 200 assert resp.json() == {"status": "ready"} # ── OpenAPI / Docs ─────────────────────────────────────────────── class TestDocsEndpoints: """Docs and OpenAPI schema should be accessible without auth.""" def test_openapi_json(self, client: TestClient) -> None: resp = client.get("/openapi.json") assert resp.status_code == 200 schema = resp.json() assert "openapi" in schema assert "paths" in schema assert len(schema["paths"]) > 0 def test_docs_redirects(self, client: TestClient) -> None: resp = client.get("/docs") assert resp.status_code == 200 assert "text/html" in resp.headers["content-type"] # ── Scraping Endpoints ─────────────────────────────────────────── class TestScrapeEndpoint: """POST /v1/scrape — core scraping endpoint.""" def test_scrape_success(self, client: TestClient) -> None: resp = client.post("/v1/scrape", json={"url": "https://example.com"}) assert resp.status_code == 200 body = resp.json() assert body.get("success") is True assert "data" in body assert "markdown" in body["data"] assert "metadata" in body["data"] def test_scrape_missing_url_returns_422(self, client: TestClient) -> None: resp = client.post("/v1/scrape", json={}) assert resp.status_code == 422 body = resp.json() assert "detail" in body def test_scrape_invalid_url_returns_422(self, client: TestClient) -> None: """URL validation happens inside scraper, returns 200 with error.""" resp = client.post("/v1/scrape", json={"url": "not-a-url"}) # Scraper handles validation internally; response may be 200 with error body assert resp.status_code in (200, 422) def test_scrape_with_options(self, client: TestClient) -> None: resp = client.post( "/v1/scrape", json={ "url": "https://example.com", "timeout": 10, "output_format": "markdown", }, ) assert resp.status_code == 200 def test_scrape_response_shape(self, client: TestClient) -> None: resp = client.post("/v1/scrape", json={"url": "https://example.com"}) assert resp.status_code == 200 body = resp.json() # Success response should have data, not error assert "error" not in body or not body.get("error") class TestDetectBlockEndpoint: """POST /v1/detect-block — anti-bot detection.""" def test_detect_block_success(self, client: TestClient) -> None: """detect-block takes raw string body, makes real HTTP calls.""" # This endpoint bypasses the scraper mock and makes real HTTP calls. # In tests, it will either fail or return a result depending on network. # We just verify it doesn't 500. resp = client.post( "/v1/detect-block", data="https://example.com", headers={"Content-Type": "application/json"}, ) assert resp.status_code in (200, 422, 503) class TestCrawlEndpoint: """POST /v1/crawl — multi-page crawl.""" def test_crawl_success(self, client: TestClient) -> None: resp = client.post( "/v1/crawl", json={"url": "https://example.com", "max_pages": 3}, ) assert resp.status_code == 200 body = resp.json() # Crawl returns a dict with job info assert isinstance(body, dict) class TestMapEndpoint: """POST /v1/map — URL discovery.""" def test_map_success(self, client: TestClient) -> None: resp = client.post("/v1/map", json={"url": "https://example.com"}) assert resp.status_code == 200 body = resp.json() assert body.get("success") is True assert "links" in body.get("data", {}) # ── Extraction Endpoints ───────────────────────────────────────── class TestExtractEndpoint: """POST /v1/extract — structured data extraction.""" def test_extract_success(self, client: TestClient) -> None: resp = client.post( "/v1/extract", json={ "url": "https://example.com", "schema": {"product_name": "name of product"}, }, ) assert resp.status_code == 200 body = resp.json() # Structure depends on router, but should have data assert isinstance(body, dict) def test_extract_missing_schema_returns_422(self, client: TestClient) -> None: """Extract with just URL and no schema may still pass validation.""" resp = client.post("/v1/extract", json={"url": "https://example.com"}) # Schema may be optional; just verify it doesn't 500 assert resp.status_code in (200, 422) # ── Config Endpoints ───────────────────────────────────────────── class TestConfigEndpoint: """GET /v1/config — runtime configuration.""" def test_config_success(self, client: TestClient) -> None: resp = client.get("/v1/config") assert resp.status_code == 200 body = resp.json() assert isinstance(body, dict) # ── Stats Endpoints ────────────────────────────────────────────── class TestStatsEndpoint: """GET /v1/stats — usage statistics.""" def test_stats_success(self, client: TestClient) -> None: resp = client.get("/v0/stats") assert resp.status_code == 200 body = resp.json() assert isinstance(body, dict) # ── Templates Endpoints ────────────────────────────────────────── class TestTemplatesEndpoint: """Template listing and retrieval.""" def test_list_templates(self, client: TestClient) -> None: resp = client.get("/v1/templates") # 404 or 200 depending on whether template listing exists assert resp.status_code in (200, 404) # ── Error Handling ─────────────────────────────────────────────── class TestErrorHandling: """API returns consistent error shapes for HTTP errors.""" def test_404_returns_json(self, client: TestClient) -> None: resp = client.get("/nonexistent-route-12345") assert resp.status_code == 404 body = resp.json() assert "detail" in body def test_405_on_wrong_method(self, client: TestClient) -> None: resp = client.get("/v1/scrape") assert resp.status_code == 405 def test_422_on_invalid_json(self, client: TestClient) -> None: resp = client.post( "/v1/scrape", data="not-json", headers={"Content-Type": "application/json"} ) assert resp.status_code == 422 # ── Auth Gating ────────────────────────────────────────────────── class TestAuth: """PRY_API_KEY gating on protected endpoints.""" def test_health_is_public(self, client_auth: TestClient) -> None: """Health endpoints should not require auth.""" resp = client_auth.get("/health") assert resp.status_code in (200, 503) def test_scrape_requires_auth(self, client_auth: TestClient) -> None: """Scraping endpoints require valid Bearer token when api_key is set.""" resp = client_auth.post( "/v1/scrape", json={"url": "https://example.com"}, ) assert resp.status_code == 401 def test_scrape_with_valid_auth(self, client_auth: TestClient) -> None: """Valid Bearer token should pass auth.""" resp = client_auth.post( "/v1/scrape", json={"url": "https://example.com"}, headers={"authorization": "Bearer test-api-key"}, ) assert resp.status_code == 200 def test_scrape_with_wrong_auth(self, client_auth: TestClient) -> None: """Wrong Bearer token should get 401.""" resp = client_auth.post( "/v1/scrape", json={"url": "https://example.com"}, headers={"authorization": "Bearer wrong-key"}, ) assert resp.status_code == 401 # ── CORS Headers ───────────────────────────────────────────────── class TestCORS: """CORS middleware should set broad allow-origin.""" def test_cors_headers_present(self, client: TestClient) -> None: resp = client.options( "/v1/scrape", headers={ "Origin": "https://example.com", "Access-Control-Request-Method": "POST", }, ) assert resp.status_code == 200 headers = resp.headers assert headers.get("access-control-allow-origin") == "*" assert "POST" in headers.get("access-control-allow-methods", "") class TestMetricsEndpoint: """GET /metrics — Prometheus metrics endpoint.""" def test_metrics_returns_200(self, client: TestClient) -> None: resp = client.get("/metrics") assert resp.status_code == 200 assert "text/plain" in resp.headers["content-type"] body = resp.text # Should contain at least some metric names assert "pry_" in body or "# HELP" in body def test_metrics_contains_request_count(self, client: TestClient) -> None: """After one scrape request, request count metric should exist.""" # Trigger a request first client.post("/v1/scrape", json={"url": "https://example.com"}) resp = client.get("/metrics") assert resp.status_code == 200 # The metric may or may not have been recorded depending on the # middleware (it might not use prometheus_client.Counter). # Just verify the endpoint works. assert len(resp.text) > 0