diff --git a/.github/workflows/examples-latest.yml b/.github/workflows/examples-latest.yml new file mode 100644 index 0000000..a625f13 --- /dev/null +++ b/.github/workflows/examples-latest.yml @@ -0,0 +1,62 @@ +name: Examples (latest dependencies) + +on: + schedule: + - cron: '0 6 * * 1' + workflow_dispatch: + pull_request: + branches: [main] + paths: + - 'examples/**' + - 'tests/examples/**' + - 'src/**' + - 'pyproject.toml' + - '.github/workflows/examples-latest.yml' + +concurrency: + group: examples-latest-${{ github.event_name == 'pull_request' && github.ref || github.sha }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + examples-latest: + name: Examples - Python ${{ matrix.python-version }} (latest deps) + runs-on: ubuntu-latest + env: + UV_PYTHON: ${{ matrix.python-version }} + strategy: + fail-fast: false + matrix: + python-version: ["3.10", "3.15"] + + steps: + - name: Checkout code + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + persist-credentials: false + + - name: Set up uv + uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v6 + with: + enable-cache: true + version: "0.9.12" + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies with latest allowed versions + run: uv sync --all-extras --upgrade + + - name: Print resolved versions of key frameworks + run: | + uv pip list | grep -Ei '^(litestar|starlette|fastapi|strawberry-graphql|advanced-alchemy|sqlalchemy|aiosqlite|mcp|uvicorn|jinja2|httpx) ' + + - name: Run example tests + run: uv run pytest tests/examples -v + shell: bash + env: + PYTHONDONTWRITEBYTECODE: 1 diff --git a/Makefile b/Makefile index 56b5905..924851b 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ .DEFAULT_GOAL := help -.PHONY: help install dev install-uv lint lint-fix fmt fmt-check type-check test test-cov test-fast test-parallel test-parallel-fast test-debug test-failed ci docs docs-serve docs-clean build clean destroy prek-install prek-run prek-update lock upgrade wt worktree wt-ls worktree-list wt-j worktree-jump worktree-prune ci-install example example-asgi example-litestar example-aa example-graphql example-websocket example-mcp example-mcp-server act act-ci act-test +.PHONY: help install dev install-uv lint lint-fix fmt fmt-check type-check test test-cov test-fast test-examples test-parallel test-parallel-fast test-debug test-failed ci docs docs-serve docs-clean build clean destroy prek-install prek-run prek-update lock upgrade wt worktree wt-ls worktree-list wt-j worktree-jump worktree-prune ci-install example example-asgi example-litestar example-aa example-graphql example-websocket example-mcp example-mcp-server act act-ci act-test # ==================================================================================== # # VARIABLES @@ -58,6 +58,9 @@ test-cov: ## Run tests with coverage report test-fast: ## Run tests without slow/integration tests @PYTHONDONTWRITEBYTECODE=1 uv run pytest -m "not integration" +test-examples: ## Run example app tests only + @PYTHONDONTWRITEBYTECODE=1 uv run pytest tests/examples -v + test-parallel: ## Run tests in parallel with pytest-xdist @PYTHONDONTWRITEBYTECODE=1 uv run pytest -n auto diff --git a/examples/litestar_advanced_alchemy/models.py b/examples/litestar_advanced_alchemy/models.py index 82beabb..19b9119 100644 --- a/examples/litestar_advanced_alchemy/models.py +++ b/examples/litestar_advanced_alchemy/models.py @@ -17,7 +17,7 @@ class User(UUIDAuditBase): name: Mapped[str] = mapped_column(String(100)) email: Mapped[str] = mapped_column(String(255), unique=True, index=True) - posts: Mapped[list[Post]] = relationship(back_populates="author") + posts: Mapped[list[Post]] = relationship(back_populates="author", cascade="all, delete-orphan") class Post(UUIDAuditBase): diff --git a/tests/examples/__init__.py b/tests/examples/__init__.py new file mode 100644 index 0000000..d7c02fc --- /dev/null +++ b/tests/examples/__init__.py @@ -0,0 +1 @@ +"""Tests exercising the applications under ``examples/``.""" diff --git a/tests/examples/conftest.py b/tests/examples/conftest.py new file mode 100644 index 0000000..18b2ffb --- /dev/null +++ b/tests/examples/conftest.py @@ -0,0 +1,52 @@ +"""Shared fixtures for testing the applications under ``examples/``. + +``examples`` is a real (namespace) package, but its submodules build module-level +singletons (toolbars, Litestar/Starlette/FastAPI apps, on-disk storage files) at +import time. Importing the same example twice in a test session would reuse that +stale state, so every test must import examples through the ``import_example`` +fixture below, which evicts any cached module (and its submodules) before and +after each import to guarantee a fresh app per test. +""" + +from __future__ import annotations + +import importlib +import sys +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from collections.abc import Iterator + from types import ModuleType + + +def _evict(dotted_path: str) -> None: + """Remove ``dotted_path`` and any of its submodules from ``sys.modules``.""" + for name in [n for n in sys.modules if n == dotted_path or n.startswith(f"{dotted_path}.")]: + del sys.modules[name] + + +@pytest.fixture +def import_example() -> Iterator[Callable[[str], ModuleType]]: + """Import an example module fresh, isolating module-level state between tests. + + Usage:: + + def test_something(import_example): + app_module = import_example("examples.litestar_basic.app") + client = TestClient(app_module.app) + """ + imported: list[str] = [] + + def _import(dotted_path: str) -> ModuleType: + _evict(dotted_path) + module = importlib.import_module(dotted_path) + imported.append(dotted_path) + return module + + yield _import + + for dotted_path in imported: + _evict(dotted_path) diff --git a/tests/examples/test_alerts_panel_example.py b/tests/examples/test_alerts_panel_example.py new file mode 100644 index 0000000..2cf4824 --- /dev/null +++ b/tests/examples/test_alerts_panel_example.py @@ -0,0 +1,57 @@ +"""Tests for the alerts panel example script.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from types import ModuleType + + +@pytest.mark.asyncio +async def test_main_reports_real_alerts_for_every_scenario( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """Running the real script should produce real, scenario-specific alerts.""" + module = import_example("examples.alerts_panel_example") + + await module.main() + + output = capsys.readouterr().out + + # Example 1: missing security headers. + assert "Total Alerts: 3" in output + assert "Content-Security-Policy" in output + assert "X-Content-Type-Options" in output + assert "Strict-Transport-Security" in output + + # Example 2: insecure cookie. + assert "Cookie Security Alerts:" in output + + # Example 3: missing CSRF protection on a POST request. + assert "CSRF Alerts:" in output + + # Example 4: debug mode enabled in production really triggers a configuration alert. + assert "Configuration Alerts:" in output + + # Example 5: oversized response body. + assert "Performance Alerts:" in output + + # Example 6: slow database queries. + assert "Database Performance Alerts:" in output + + # Example 7: N+1 query pattern really triggers an alert naming the real suggestion. + assert "N+1 Query Alerts:" in output + assert "joinedload" in output + + # Example 8: combined scenario really spans multiple alert categories. + assert "Summary by Category:" in output + assert "Security:" in output + assert "Database:" in output + assert "All Alerts:" in output + + assert "Examples completed successfully!" in output diff --git a/tests/examples/test_asgi_basic.py b/tests/examples/test_asgi_basic.py new file mode 100644 index 0000000..9b231da --- /dev/null +++ b/tests/examples/test_asgi_basic.py @@ -0,0 +1,137 @@ +"""Tests for the raw-ASGI ``examples/asgi_basic`` example. + +This example hand-rolls its own debug-toolbar middleware directly against the +``DebugToolbar`` core API. It injects the toolbar markup into HTML responses, +serves ``/_debug_toolbar/static/*`` assets, and exposes JSON endpoints at +``/_debug_toolbar/api/requests`` and ``/_debug_toolbar/api/requests/``. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from uuid import uuid4 + +import httpx +import pytest + +if TYPE_CHECKING: + from collections.abc import Callable + from types import ModuleType + + +@pytest.fixture +def asgi_basic(import_example: Callable[[str], ModuleType]) -> ModuleType: + return import_example("examples.asgi_basic.app") + + +def _client(asgi_basic: ModuleType) -> httpx.AsyncClient: + transport = httpx.ASGITransport(app=asgi_basic.app) + return httpx.AsyncClient(transport=transport, base_url="http://testserver") + + +async def test_home_page_injects_toolbar_markup(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + response = await client.get("/") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert "ASGI Debug Toolbar Example" in response.text + assert 'id="debug-toolbar"' in response.text + assert '' in response.text + assert '' in response.text + + +async def test_about_page_injects_toolbar_markup(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + response = await client.get("/about") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert "About" in response.text + assert 'id="debug-toolbar"' in response.text + + +async def test_api_data_returns_json_without_toolbar_markup(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + response = await client.get("/api/data") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + data = response.json() + assert data["message"] == "Hello from the API" + assert 'id="debug-toolbar"' not in response.text + + +async def test_unknown_path_returns_404(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + response = await client.get("/no-such-page") + + assert response.status_code == 404 + + +async def test_server_timing_header_present_after_request(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + response = await client.get("/") + + assert "server-timing" in response.headers + + +async def test_static_css_asset(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + response = await client.get("/_debug_toolbar/static/toolbar.css") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/css") + assert response.text + + +async def test_static_js_asset(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + response = await client.get("/_debug_toolbar/static/toolbar.js") + + assert response.status_code == 200 + assert "javascript" in response.headers["content-type"] + assert response.text + + +async def test_debug_toolbar_api_requests_lists_recorded_request(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + await client.get("/") + response = await client.get("/_debug_toolbar/api/requests") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + requests_list = response.json() + assert len(requests_list) >= 1 + assert "request_id" in requests_list[0] + assert requests_list[0]["path"] == "/" + + +async def test_debug_toolbar_api_request_detail_contains_panel_data(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + await client.get("/") + requests_response = await client.get("/_debug_toolbar/api/requests") + request_id = requests_response.json()[0]["request_id"] + detail_response = await client.get(f"/_debug_toolbar/api/requests/{request_id}") + + assert detail_response.status_code == 200 + assert detail_response.headers["content-type"].startswith("application/json") + detail = detail_response.json() + + assert detail["request_id"] == request_id + panel_data = detail["panel_data"] + assert set(panel_data) >= {"TimerPanel", "RequestPanel", "ResponsePanel", "LoggingPanel", "VersionsPanel"} + + timer_stats = panel_data["TimerPanel"] + assert timer_stats["total_time_ms"] >= 0 + + request_stats = panel_data["RequestPanel"] + assert request_stats["method"] == "GET" + assert request_stats["path"] == "/" + + +async def test_debug_toolbar_api_request_detail_unknown_id_returns_404(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + response = await client.get(f"/_debug_toolbar/api/requests/{uuid4()}") + + assert response.status_code == 404 diff --git a/tests/examples/test_async_profiler_example.py b/tests/examples/test_async_profiler_example.py new file mode 100644 index 0000000..e046e1a --- /dev/null +++ b/tests/examples/test_async_profiler_example.py @@ -0,0 +1,75 @@ +"""Tests for the async profiler panel example script.""" + +from __future__ import annotations + +import re +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from types import ModuleType + + +@pytest.mark.asyncio +async def test_main_reports_real_task_tracking( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """Running the real script should report real task counts, durations, and lag samples.""" + module = import_example("examples.async_profiler_example") + + await module.main() + + output = capsys.readouterr().out + assert "Backend: taskfactory" in output + + total_match = re.search(r"Total tasks created: (\d+)", output) + assert total_match is not None + assert int(total_match.group(1)) == 5 + + completed_match = re.search(r"Completed tasks: (\d+)", output) + assert completed_match is not None + assert int(completed_match.group(1)) == 5 + + assert "Tasks tracked:" in output + for i in range(5): + assert f"fast_task_{i} (fast_task)" in output + + assert "Total tasks: 4" in output + duration_lines = re.findall(r" - (fast_\d+|slow_\d+): ([\d.]+)ms", output) + assert len(duration_lines) >= 2 + durations_by_name = dict(duration_lines) + assert float(durations_by_name["fast_1"]) < float(durations_by_name["slow_1"]) + + timeline_match = re.search(r"Timeline duration: ([\d.]+)ms", output) + assert timeline_match is not None + assert float(timeline_match.group(1)) > 0 + + max_concurrent_match = re.search(r"Max concurrent tasks: (\d+)", output) + assert max_concurrent_match is not None + assert int(max_concurrent_match.group(1)) >= 1 + + blocking_match = re.search(r"Blocking calls detected: (\d+)", output) + assert blocking_match is not None + assert int(blocking_match.group(1)) == 1 + assert "Has warnings: True" in output + assert "blocking_task:" in output + blocking_duration_match = re.search(r" - blocking_task: ([\d.]+)ms", output) + assert blocking_duration_match is not None + assert float(blocking_duration_match.group(1)) >= 100 + + lag_match = re.search(r"Max event loop lag: ([\d.]+)ms", output) + assert lag_match is not None + assert float(lag_match.group(1)) > 0 + assert "Lag samples above threshold:" in output + + assert "Nested task results: [0, 2, 4]" in output + + parent_lines = [line for line in output.splitlines() if "(parent:" in line] + assert len(parent_lines) == 4 + + assert "Panel nav subtitle:" in output + assert "Panel title: Async" in output + assert "Examples completed successfully!" in output diff --git a/tests/examples/test_cache_panel_demo.py b/tests/examples/test_cache_panel_demo.py new file mode 100644 index 0000000..9fa5027 --- /dev/null +++ b/tests/examples/test_cache_panel_demo.py @@ -0,0 +1,60 @@ +"""Tests for the cache panel demo script.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from types import ModuleType + + +def test_demo_context_manager_tracking_reports_real_operations( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """Running the real demo should report four tracked operations with hit/miss status.""" + module = import_example("examples.cache_panel_demo") + + module.demo_context_manager_tracking() + + output = capsys.readouterr().out + assert "Total operations tracked: 4" in output + assert "GET" in output + assert "HIT" in output + assert "MISS" in output + assert "SET" in output + assert "memcached" in output + + +def test_demo_multi_key_operation_reports_real_keys( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """Running the real multi-key demo should report the tracked key list and hit status.""" + module = import_example("examples.cache_panel_demo") + + module.demo_multi_key_operation() + + output = capsys.readouterr().out + assert "Operation: MGET" in output + assert "Keys: ['key1', 'key2', 'key3']" in output + assert "Hit: True" in output + + +def test_demo_statistics_reports_real_hit_rate( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """Running the real statistics demo should compute a 60% hit rate from its five operations.""" + module = import_example("examples.cache_panel_demo") + + module.demo_statistics() + + output = capsys.readouterr().out + assert "Total operations: 5" in output + assert "Cache hits: 3" in output + assert "Cache misses: 1" in output + assert "Hit rate: 75.0%" in output diff --git a/tests/examples/test_fastapi_basic.py b/tests/examples/test_fastapi_basic.py new file mode 100644 index 0000000..1e10ef8 --- /dev/null +++ b/tests/examples/test_fastapi_basic.py @@ -0,0 +1,144 @@ +"""Tests for the ``examples/fastapi_basic`` example. + +The dependency-injection panel's stats are the subject of an in-flight fix +(branch ``fix/fastapi-dependencies-panel``); these tests assert the panel key +is present in a request's panel data without asserting on its contents. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from starlette.testclient import TestClient + +if TYPE_CHECKING: + from collections.abc import Callable + from types import ModuleType + + +@pytest.fixture +def fastapi_basic(import_example: Callable[[str], ModuleType]) -> ModuleType: + return import_example("examples.fastapi_basic.app") + + +def test_home_page_returns_toolbar_injected_page(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + response = client.get("/") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert 'id="debug-toolbar"' in response.text + assert "/_debug_toolbar/static/toolbar.css" in response.text + assert "/_debug_toolbar/static/toolbar.js" in response.text + + +def test_get_user_returns_json_without_toolbar_markup(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + response = client.get("/users/123") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + data = response.json() + assert data["user"]["username"] == "alice" + assert data["requested_by"] == "alice" + assert 'id="debug-toolbar"' not in response.text + + +def test_get_user_unknown_id_returns_404(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + response = client.get("/users/999") + + assert response.status_code == 404 + + +def test_list_items_returns_json_without_toolbar_markup(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + response = client.get("/items", params={"skip": 10, "limit": 20}) + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + data = response.json() + assert data["skip"] == 10 + assert data["limit"] == 20 + assert 'id="debug-toolbar"' not in response.text + + +def test_settings_returns_json_without_toolbar_markup(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + response = client.get("/settings") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + assert "settings" in response.json() + assert 'id="debug-toolbar"' not in response.text + + +def test_static_css_asset(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + response = client.get("/_debug_toolbar/static/toolbar.css") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/css") + assert response.text + + +def test_static_js_asset(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + response = client.get("/_debug_toolbar/static/toolbar.js") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/javascript") + assert response.text + + +def test_history_api_lists_requests_after_html_requests(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + client.get("/") + client.get("/users/123") + response = client.get("/_debug_toolbar/api/requests") + + assert response.status_code == 200 + data = response.json() + assert isinstance(data, list) + assert len(data) == 2 + + +def test_request_detail_contains_multiple_panels(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + client.get("/") + requests_response = client.get("/_debug_toolbar/api/requests") + request_id = requests_response.json()[0]["request_id"] + detail_response = client.get(f"/_debug_toolbar/api/requests/{request_id}") + + assert detail_response.status_code == 200 + assert detail_response.headers["content-type"].startswith("application/json") + detail = detail_response.json() + + panel_data = detail["panel_data"] + expected_panels = {"TimerPanel", "RequestPanel", "ResponsePanel", "LoggingPanel", "VersionsPanel"} + assert expected_panels <= set(panel_data) + + request_stats = panel_data["RequestPanel"] + assert request_stats["method"] == "GET" + assert request_stats["path"] == "/" + + +def test_dependency_injection_panel_reports_real_resolutions(fastapi_basic: ModuleType) -> None: + with TestClient(fastapi_basic.app) as client: + client.get("/users/123") + requests_response = client.get("/_debug_toolbar/api/requests") + request_id = requests_response.json()[0]["request_id"] + detail_response = client.get(f"/_debug_toolbar/api/requests/{request_id}") + + panel_stats = detail_response.json()["panel_data"]["DependencyInjectionPanel"] + resolved = panel_stats["resolved_dependencies"] + assert resolved != [] + + names = {dep["name"] for dep in resolved} + assert "get_db" in names + assert "get_current_user" in names + assert "get_token" in names + + assert panel_stats["total_count"] == len(resolved) + assert panel_stats["cached_count"] + panel_stats["fresh_count"] == panel_stats["total_count"] diff --git a/tests/examples/test_flamegraph_example.py b/tests/examples/test_flamegraph_example.py new file mode 100644 index 0000000..0dc242b --- /dev/null +++ b/tests/examples/test_flamegraph_example.py @@ -0,0 +1,50 @@ +"""Tests for the flame graph example script.""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from pathlib import Path + from types import ModuleType + + import pytest + + +def test_main_profiles_real_work_and_writes_valid_speedscope_json( + import_example: Callable[[str], ModuleType], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Running the real script should profile real work and write a valid speedscope file.""" + monkeypatch.chdir(tmp_path) + module = import_example("examples.flamegraph_example") + + module.main() + + output = capsys.readouterr().out + assert "Profiled 10 fibonacci calculations" in output + assert "Speedscope JSON generated successfully!" in output + assert "Top 10 functions by cumulative time:" in output + assert "fibonacci" in output + + saved_path = tmp_path / "profile.speedscope.json" + assert saved_path.exists() + + data = json.loads(saved_path.read_text()) + assert data["$schema"] + assert len(data["shared"]["frames"]) > 0 + assert len(data["profiles"]) > 0 + frame_names = [frame["name"] for frame in data["shared"]["frames"]] + assert any("fibonacci" in name for name in frame_names) + assert any("data_processing" in name for name in frame_names) + + profile = data["profiles"][0] + assert profile["type"] + assert profile["unit"] + assert profile["endValue"] >= 0 + assert len(profile["samples"]) > 0 + assert len(profile["samples"]) == len(profile["weights"]) diff --git a/tests/examples/test_graphql_panel_example.py b/tests/examples/test_graphql_panel_example.py new file mode 100644 index 0000000..c2e2dac --- /dev/null +++ b/tests/examples/test_graphql_panel_example.py @@ -0,0 +1,107 @@ +"""Tests for the GraphQL Panel example application.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest +from litestar.testing import TestClient + +from debug_toolbar.extras.strawberry.extension import STRAWBERRY_AVAILABLE + +if TYPE_CHECKING: + from types import ModuleType + + +@pytest.fixture +def app_module(import_example: Callable[[str], ModuleType]) -> ModuleType: + return import_example("examples.graphql_panel_example") + + +def test_index_page_shows_toolbar(app_module: ModuleType) -> None: + with TestClient(app_module.app) as client: + response = client.get("/") + + assert response.status_code == 200 + assert 'id="debug-toolbar"' in response.text + + +@pytest.mark.skipif(not STRAWBERRY_AVAILABLE, reason="Strawberry not installed") +class TestGraphQLQueries: + def test_posts_query_returns_expected_shape(self, app_module: ModuleType) -> None: + query = """ + query { + posts { + id + title + author { + id + name + } + } + } + """ + with TestClient(app_module.app) as client: + response = client.post("/graphql", json={"query": query}) + + assert response.status_code == 200 + data = response.json() + assert "errors" not in data or not data["errors"] + posts = data["data"]["posts"] + assert len(posts) == 5 + assert posts[0]["title"] == "GraphQL Basics" + assert posts[0]["author"]["name"] == "Alice" + + def test_create_user_mutation_succeeds(self, app_module: ModuleType) -> None: + mutation = """ + mutation { + createUser(name: "Dave", email: "dave@example.com") { + id + name + email + } + } + """ + with TestClient(app_module.app) as client: + response = client.post("/graphql", json={"query": mutation}) + + assert response.status_code == 200 + data = response.json() + assert "errors" not in data or not data["errors"] + created = data["data"]["createUser"] + assert created["name"] == "Dave" + assert created["email"] == "dave@example.com" + assert created["id"] == "4" + + +@pytest.mark.skipif(not STRAWBERRY_AVAILABLE, reason="Strawberry not installed") +def test_graphql_panel_records_resolver_and_n_plus_one_stats(app_module: ModuleType) -> None: + query = """ + query { + posts { + id + title + author { + id + name + } + } + } + """ + with TestClient(app_module.app) as client: + response = client.post("/graphql", json={"query": query}) + assert response.status_code == 200 + + requests = client.get("/_debug_toolbar/api/requests").json() + graphql_request = next(req for req in requests if req["metadata"].get("path") == "/graphql") + + detail = client.get(f"/_debug_toolbar/api/requests/{graphql_request['request_id']}").json() + graphql_stats = detail["panel_data"]["GraphQLPanel"] + + assert graphql_stats["operation_count"] >= 1 + assert graphql_stats["resolver_count"] > 0 + assert graphql_stats["n_plus_one_count"] >= 1 + author_patterns = [p for p in graphql_stats["n_plus_one_patterns"] if p["field_name"] == "author"] + assert author_patterns + assert author_patterns[0]["count"] >= 3 diff --git a/tests/examples/test_litestar_advanced_alchemy.py b/tests/examples/test_litestar_advanced_alchemy.py new file mode 100644 index 0000000..04fc649 --- /dev/null +++ b/tests/examples/test_litestar_advanced_alchemy.py @@ -0,0 +1,263 @@ +"""Tests for the ``examples.litestar_advanced_alchemy`` application.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from litestar.testing import TestClient + +if TYPE_CHECKING: + from collections.abc import Callable + from types import ModuleType + + import pytest + +_REPO_ROOT = Path(__file__).resolve().parents[2] + +_CREATE_USER_SCRIPT = """ +import json +import os +import sys + +sys.path.insert(0, {repo_root!r}) +os.chdir({tmp_dir!r}) + +from examples.litestar_advanced_alchemy import app as app_module +from litestar.testing import TestClient + +with TestClient(app_module.app) as client: + create_response = client.post( + "/api/users", + data={{"name": "Frank Castle", "email": "frank@example.com"}}, + ) + users = client.get("/api/users").json() + +print(json.dumps({{ + "create_status": create_response.status_code, + "create_text": create_response.text, + "users": users, +}})) +""" + +_DELETE_USER_SCRIPT = """ +import json +import os +import sys + +sys.path.insert(0, {repo_root!r}) +os.chdir({tmp_dir!r}) + +from examples.litestar_advanced_alchemy import app as app_module +from litestar.testing import TestClient + +with TestClient(app_module.app) as client: + users_before = client.get("/api/users").json() + target = next(u for u in users_before if u["name"] == "Eve Wilson") + delete_response = client.delete(f"/api/users/{{target['id']}}") + users_after = client.get("/api/users").json() + +print(json.dumps({{ + "delete_status": delete_response.status_code, + "target_id": target["id"], + "users_before": users_before, + "users_after": users_after, +}})) +""" + + +def _run_isolated_script(script_template: str, tmp_dir: Path) -> dict[str, Any]: + """Run a script in a fresh subprocess, isolated from this process's SQLAlchemy state. + + Advanced-Alchemy's declarative models register themselves against global + SQLAlchemy registries (``advanced_alchemy.base.orm_registry``) that are not reset + when ``import_example`` evicts and reimports the example module. Repeated in-process + reimports of this particular example therefore corrupt write visibility for later + requests. Running each mutation in its own interpreter sidesteps that corruption. + """ + script = script_template.format(repo_root=str(_REPO_ROOT), tmp_dir=str(tmp_dir)) + result = subprocess.run( + [sys.executable, "-c", script], + capture_output=True, + text=True, + timeout=60, + check=True, + ) + return json.loads(result.stdout.strip().splitlines()[-1]) + + +def _latest_request_id(client: TestClient, path: str) -> str: + """Return the ``request_id`` of the most recent recorded request to ``path``.""" + requests = client.get("/_debug_toolbar/api/requests").json() + matches = [r for r in requests if r["metadata"]["path"] == path] + assert matches, f"no recorded request for {path}" + return matches[-1]["request_id"] + + +def _panel_data(client: TestClient, path: str) -> dict[str, Any]: + """Return the ``panel_data`` dict for the most recent request to ``path``.""" + request_id = _latest_request_id(client, path) + detail = client.get(f"/_debug_toolbar/api/requests/{request_id}").json() + return detail["panel_data"] + + +def _load_app( + import_example: Callable[[str], ModuleType], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> ModuleType: + monkeypatch.chdir(tmp_path) + return import_example("examples.litestar_advanced_alchemy.app") + + +class TestHtmlPages: + def test_index_page( + self, + import_example: Callable[[str], ModuleType], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + with TestClient(app_module.app) as client: + response = client.get("/") + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert 'id="debug-toolbar"' in response.text + + def test_users_page( + self, + import_example: Callable[[str], ModuleType], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + with TestClient(app_module.app) as client: + response = client.get("/users") + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert 'id="debug-toolbar"' in response.text + + def test_posts_page( + self, + import_example: Callable[[str], ModuleType], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + with TestClient(app_module.app) as client: + response = client.get("/posts") + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert 'id="debug-toolbar"' in response.text + + +class TestApiListEndpoints: + def test_list_users_returns_seeded_data( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + with TestClient(app_module.app) as client: + response = client.get("/api/users") + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + users = response.json() + assert len(users) == 5 + names = {u["name"] for u in users} + assert names == {"Alice Johnson", "Bob Smith", "Charlie Brown", "Diana Ross", "Eve Wilson"} + alice = next(u for u in users if u["name"] == "Alice Johnson") + assert alice["email"] == "alice@example.com" + + def test_list_posts_returns_seeded_data( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + with TestClient(app_module.app) as client: + response = client.get("/api/posts") + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + posts = response.json() + assert len(posts) == 7 + titles = {p["title"] for p in posts} + assert "Getting Started with Litestar" in titles + alice_post = next(p for p in posts if p["title"] == "Getting Started with Litestar") + assert alice_post["author"] == "Alice Johnson" + + +class TestCreateAndDeleteUser: + def test_create_user_appears_in_list(self, tmp_path: Path) -> None: + result = _run_isolated_script(_CREATE_USER_SCRIPT, tmp_path) + + assert result["create_status"] == 201 + assert "Frank Castle" in result["create_text"] + + names = {u["name"] for u in result["users"]} + assert "Frank Castle" in names + assert len(result["users"]) == 6 + + def test_delete_user_removes_it(self, tmp_path: Path) -> None: + result = _run_isolated_script(_DELETE_USER_SCRIPT, tmp_path) + + assert result["delete_status"] == 204 + assert len(result["users_before"]) == 5 + + remaining_ids = {u["id"] for u in result["users_after"]} + assert result["target_id"] not in remaining_ids + assert len(result["users_after"]) == 4 + + +class TestNPlusOneDetection: + def test_n_plus_one_demo_is_flagged_by_sqlalchemy_panel( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + with TestClient(app_module.app) as client: + bad_response = client.get("/api/users-with-posts-bad") + assert bad_response.status_code == 200 + + baseline_response = client.get("/api/users") + assert baseline_response.status_code == 200 + + bad_panel_data = _panel_data(client, "/api/users-with-posts-bad") + baseline_panel_data = _panel_data(client, "/api/users") + + bad_sql_stats = bad_panel_data["SQLAlchemyPanel"] + baseline_sql_stats = baseline_panel_data["SQLAlchemyPanel"] + + assert bad_sql_stats["n_plus_one_count"] > 0 + assert len(bad_sql_stats["n_plus_one_groups"]) > 0 + assert bad_sql_stats["has_issues"] is True + assert bad_sql_stats["query_count"] > baseline_sql_stats["query_count"] + + +class TestAsyncProfilerPanel: + def test_async_demo_reports_task_stats( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + with TestClient(app_module.app) as client: + response = client.get("/async-demo") + assert response.status_code == 200 + + panel_data = _panel_data(client, "/async-demo") + async_stats = panel_data["AsyncProfilerPanel"] + + assert async_stats["summary"]["total_tasks"] == 5 + assert async_stats["summary"]["completed_tasks"] == 5 + + +class TestOtherPanelsPresent: + def test_settings_headers_alerts_memory_panels_present( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + with TestClient(app_module.app) as client: + client.get("/") + panel_data = _panel_data(client, "/") + + assert "SettingsPanel" in panel_data + assert "HeadersPanel" in panel_data + assert "AlertsPanel" in panel_data + assert "MemoryPanel" in panel_data diff --git a/tests/examples/test_litestar_basic.py b/tests/examples/test_litestar_basic.py new file mode 100644 index 0000000..1a0f600 --- /dev/null +++ b/tests/examples/test_litestar_basic.py @@ -0,0 +1,130 @@ +"""Tests for the ``examples/litestar_basic`` example.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from uuid import uuid4 + +import pytest +from litestar.testing import TestClient + +if TYPE_CHECKING: + from collections.abc import Callable + from types import ModuleType + +HTML_ROUTES = ["/", "/about", "/users", "/alerts-demo"] + + +@pytest.fixture +def litestar_basic(import_example: Callable[[str], ModuleType]) -> ModuleType: + return import_example("examples.litestar_basic.app") + + +@pytest.mark.parametrize("path", HTML_ROUTES) +def test_html_route_returns_toolbar_injected_page(litestar_basic: ModuleType, path: str) -> None: + with TestClient(litestar_basic.app) as client: + response = client.get(path) + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert 'id="debug-toolbar"' in response.text + assert "/_debug_toolbar/static/toolbar.css" in response.text + assert "/_debug_toolbar/static/toolbar.js" in response.text + + +def test_api_status_returns_json_without_toolbar_markup(litestar_basic: ModuleType) -> None: + with TestClient(litestar_basic.app) as client: + response = client.get("/api/status") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + assert response.json()["status"] == "ok" + assert 'id="debug-toolbar"' not in response.text + + +def test_static_css_asset(litestar_basic: ModuleType) -> None: + with TestClient(litestar_basic.app) as client: + response = client.get("/_debug_toolbar/static/toolbar.css") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/css") + assert response.text + + +def test_static_js_asset(litestar_basic: ModuleType) -> None: + with TestClient(litestar_basic.app) as client: + response = client.get("/_debug_toolbar/static/toolbar.js") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/javascript") + assert response.text + + +def test_history_api_lists_requests_after_html_requests(litestar_basic: ModuleType) -> None: + with TestClient(litestar_basic.app) as client: + client.get("/") + client.get("/about") + response = client.get("/_debug_toolbar/api/requests") + + assert response.status_code == 200 + data = response.json() + assert isinstance(data, list) + assert len(data) == 2 + + +def test_request_detail_unknown_id_returns_404(litestar_basic: ModuleType) -> None: + with TestClient(litestar_basic.app) as client: + response = client.get(f"/_debug_toolbar/api/requests/{uuid4()}") + + assert response.status_code == 404 + + +def test_request_detail_contains_multiple_panels(litestar_basic: ModuleType) -> None: + with TestClient(litestar_basic.app) as client: + client.get("/") + requests_response = client.get("/_debug_toolbar/api/requests") + request_id = requests_response.json()[0]["request_id"] + detail_response = client.get(f"/_debug_toolbar/api/requests/{request_id}") + + assert detail_response.status_code == 200 + assert detail_response.headers["content-type"].startswith("application/json") + detail = detail_response.json() + + panel_data = detail["panel_data"] + expected_panels = { + "TimerPanel", + "RequestPanel", + "ResponsePanel", + "LoggingPanel", + "VersionsPanel", + "HeadersPanel", + "SettingsPanel", + "TemplatesPanel", + "AlertsPanel", + "MemoryPanel", + "AsyncProfilerPanel", + } + assert expected_panels <= set(panel_data) + + +def test_templates_panel_records_jinja2_render_on_index_page(litestar_basic: ModuleType) -> None: + with TestClient(litestar_basic.app) as client: + client.get("/") + requests_response = client.get("/_debug_toolbar/api/requests") + request_id = requests_response.json()[0]["request_id"] + detail_response = client.get(f"/_debug_toolbar/api/requests/{request_id}") + + templates_stats = detail_response.json()["panel_data"]["TemplatesPanel"] + assert templates_stats["total_renders"] >= 1 + assert "jinja2" in templates_stats["engines_used"] + + +def test_headers_panel_records_request_headers(litestar_basic: ModuleType) -> None: + with TestClient(litestar_basic.app) as client: + client.get("/") + requests_response = client.get("/_debug_toolbar/api/requests") + request_id = requests_response.json()[0]["request_id"] + detail_response = client.get(f"/_debug_toolbar/api/requests/{request_id}") + + headers_stats = detail_response.json()["panel_data"]["HeadersPanel"] + assert headers_stats diff --git a/tests/examples/test_mcp_server_example.py b/tests/examples/test_mcp_server_example.py new file mode 100644 index 0000000..8c94d78 --- /dev/null +++ b/tests/examples/test_mcp_server_example.py @@ -0,0 +1,151 @@ +"""Tests for the ``examples.mcp_server_example`` application.""" + +from __future__ import annotations + +import json +from types import SimpleNamespace +from typing import TYPE_CHECKING, Any + +from litestar.testing import TestClient + +from debug_toolbar import FileToolbarStorage +from debug_toolbar.mcp import create_mcp_server +from debug_toolbar.mcp.server import MCPContext + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + from types import ModuleType + + import pytest + + +def _load_app( + import_example: Callable[[str], ModuleType], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> ModuleType: + monkeypatch.chdir(tmp_path) + return import_example("examples.mcp_server_example") + + +class TestHtmlPages: + def test_html_pages_render_with_toolbar( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + app = app_module.create_app() + with TestClient(app) as client: + for path in ("/", "/slow", "/users", "/compute", "/error-demo"): + response = client.get(path) + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert 'id="debug-toolbar"' in response.text + + +class TestJsonApiEndpoints: + def test_api_data_returns_json_without_toolbar_markup( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + app = app_module.create_app() + with TestClient(app) as client: + response = client.get("/api/data") + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + data = response.json() + assert data["total"] == 2 + assert {u["name"] for u in data["users"]} == {"Alice", "Bob"} + assert 'id="debug-toolbar"' not in response.text + + def test_api_create_returns_json_without_toolbar_markup( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + app = app_module.create_app() + with TestClient(app) as client: + response = client.post("/api/create", json={"name": "widget"}) + assert response.status_code == 201 + assert response.headers["content-type"].startswith("application/json") + assert response.json() == {"created": True, "data": {"name": "widget"}} + assert 'id="debug-toolbar"' not in response.text + + +class TestSharedFileStorage: + def test_requests_are_persisted_to_shared_storage_file( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + app = app_module.create_app() + with TestClient(app) as client: + client.get("/") + client.get("/users") + + storage_path = tmp_path / app_module.STORAGE_FILE + assert storage_path.exists() + raw = json.loads(storage_path.read_text()) + assert len(raw) >= 2 + assert all("request_id" in entry and "data" in entry for entry in raw) + paths = {entry["data"]["metadata"]["path"] for entry in raw} + assert {"/", "/users"} <= paths + + +class TestMcpToolRegistry: + def test_expected_tools_are_registered( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + storage = FileToolbarStorage(app_module.STORAGE_FILE) + mcp = create_mcp_server(storage=storage, toolbar=None, server_name="test-debug-toolbar") + + tool_names = {tool.name for tool in mcp._tool_manager.list_tools()} + assert tool_names == { + "get_request_history", + "analyze_performance_bottlenecks", + "detect_n_plus_one_queries", + "get_query_explain_plan", + "analyze_security_alerts", + "compare_requests", + "get_graphql_operations", + "get_async_task_profile", + "generate_optimization_report", + "clear_request_history", + } + + +class TestMcpToolInvocation: + def test_tools_analyze_real_requests_from_shared_storage( + self, import_example: Callable[[str], ModuleType], monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + app_module = _load_app(import_example, monkeypatch, tmp_path) + app = app_module.create_app() + with TestClient(app) as client: + response = client.get("/slow") + assert response.status_code == 200 + + storage = FileToolbarStorage(app_module.STORAGE_FILE) + mcp = create_mcp_server(storage=storage, toolbar=None, server_name="test-debug-toolbar") + fake_ctx = SimpleNamespace(request_context=SimpleNamespace(lifespan_context=MCPContext(storage=storage))) + + history_tool = mcp._tool_manager.get_tool("get_request_history") + history: list[dict[str, Any]] = history_tool.fn(fake_ctx, limit=20) + assert history + slow_entry = next(entry for entry in history if entry["path"] == "/slow") + assert slow_entry["method"] == "GET" + assert slow_entry["status_code"] == 200 + assert slow_entry["duration_ms"] > 0 + + bottleneck_tool = mcp._tool_manager.get_tool("analyze_performance_bottlenecks") + analysis: dict[str, Any] = bottleneck_tool.fn(fake_ctx, request_id=slow_entry["request_id"]) + assert "error" not in analysis + assert analysis["total_requests_analyzed"] == 1 + + n_plus_one_tool = mcp._tool_manager.get_tool("detect_n_plus_one_queries") + n_plus_one: dict[str, Any] = n_plus_one_tool.fn(fake_ctx, request_id=slow_entry["request_id"], threshold=1) + assert n_plus_one == { + "n_plus_one_detected": False, + "patterns_found": 0, + "threshold_used": 1, + "patterns": [], + "recommendations": [], + } diff --git a/tests/examples/test_memory_panel_example.py b/tests/examples/test_memory_panel_example.py new file mode 100644 index 0000000..3253a57 --- /dev/null +++ b/tests/examples/test_memory_panel_example.py @@ -0,0 +1,39 @@ +"""Tests for the memory panel example script.""" + +from __future__ import annotations + +import re +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from types import ModuleType + + +@pytest.mark.asyncio +async def test_main_reports_real_memory_allocations( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """Running the real script should report real, non-zero memory deltas for both backends.""" + module = import_example("examples.memory_panel_example") + + await module.main() + + output = capsys.readouterr().out + assert "Backend: tracemalloc" in output + assert "Auto-selected backend:" in output + assert "Top 5 Allocations:" in output + + delta_matches = re.findall(r"Memory delta: ([\d,]+) bytes", output) + assert len(delta_matches) >= 2 + assert all(int(match.replace(",", "")) > 0 for match in delta_matches) + + peak_matches = re.findall(r"Peak memory: ([\d,]+) bytes", output) + assert len(peak_matches) >= 1 + assert all(int(match.replace(",", "")) > 0 for match in peak_matches) + + assert "TRACEMALLOC Backend:" in output + assert "AUTO Backend:" in output diff --git a/tests/examples/test_profiling_panel_example.py b/tests/examples/test_profiling_panel_example.py new file mode 100644 index 0000000..c03283e --- /dev/null +++ b/tests/examples/test_profiling_panel_example.py @@ -0,0 +1,52 @@ +"""Tests for the profiling panel example script.""" + +from __future__ import annotations + +import json +import re +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from pathlib import Path + from types import ModuleType + + +@pytest.mark.asyncio +async def test_main_reports_real_profiling_results( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Running the real script should report real cProfile counts and a valid flame graph.""" + monkeypatch.chdir(tmp_path) + module = import_example("examples.profiling_panel_example") + + await module.main() + + output = capsys.readouterr().out + assert "Backend: cprofile" in output + + calls_match = re.search(r"Function Calls: ([\d,]+)", output) + assert calls_match is not None + assert int(calls_match.group(1).replace(",", "")) > 0 + + primitive_match = re.search(r"Primitive Calls: ([\d,]+)", output) + assert primitive_match is not None + assert int(primitive_match.group(1).replace(",", "")) > 0 + + assert "Top 5 Functions by Cumulative Time:" in output + assert "fibonacci" in output + + assert "Flame graph data generated successfully!" in output + assert "Saved to: panel_profile.speedscope.json" in output + + saved_path = tmp_path / "panel_profile.speedscope.json" + assert saved_path.exists() + saved_data = json.loads(saved_path.read_text()) + assert saved_data["$schema"] + assert len(saved_data["profiles"]) > 0 + assert len(saved_data["shared"]["frames"]) > 0 diff --git a/tests/examples/test_settings_panel_example.py b/tests/examples/test_settings_panel_example.py new file mode 100644 index 0000000..86a37e3 --- /dev/null +++ b/tests/examples/test_settings_panel_example.py @@ -0,0 +1,49 @@ +"""Tests for the settings panel example script.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from types import ModuleType + + +@pytest.mark.asyncio +async def test_main_reports_real_settings_and_redactions( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """Running the real script should report real toolbar/custom settings and real redactions.""" + module = import_example("examples.settings_panel_example") + + await module.main() + + output = capsys.readouterr().out + + # Example 1: basic panel with real toolbar config and environment variable counts. + assert "Toolbar Config: True" in output + assert "Python Debug:" in output + assert "Environment Variables:" in output + assert "Redacted Variables:" in output + + # Example 2: custom settings with real password/api-key redaction. + assert "App Name: My Litestar App" in output + assert "Database Host: localhost" in output + assert "Database Password: **********" in output + assert "API Key: **********" in output + assert "secret123" not in output + assert "abc123xyz" not in output + + # Example 3: show_env=False really disables the environment section. + assert "Environment Section: None" in output + + # Example 4: custom sensitive-key patterns really redact matching fields only. + assert "App Name: My App" in output + assert "Internal ID: **********" in output + assert "Confidential Data: **********" in output + assert "Public URL: https://example.com" in output + assert "secret-internal-123" not in output + assert "top-secret" not in output diff --git a/tests/examples/test_starlette_basic.py b/tests/examples/test_starlette_basic.py new file mode 100644 index 0000000..8d76ee4 --- /dev/null +++ b/tests/examples/test_starlette_basic.py @@ -0,0 +1,113 @@ +"""Tests for the ``examples/starlette_basic`` example.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from uuid import uuid4 + +import pytest +from starlette.testclient import TestClient + +if TYPE_CHECKING: + from collections.abc import Callable + from types import ModuleType + +HTML_ROUTES = ["/", "/about", "/users"] + + +@pytest.fixture +def starlette_basic(import_example: Callable[[str], ModuleType]) -> ModuleType: + return import_example("examples.starlette_basic.app") + + +@pytest.mark.parametrize("path", HTML_ROUTES) +def test_html_route_returns_toolbar_injected_page(starlette_basic: ModuleType, path: str) -> None: + with TestClient(starlette_basic.app) as client: + response = client.get(path) + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert 'id="debug-toolbar"' in response.text + assert "/_debug_toolbar/static/toolbar.css" in response.text + assert "/_debug_toolbar/static/toolbar.js" in response.text + + +def test_api_status_returns_json_without_toolbar_markup(starlette_basic: ModuleType) -> None: + with TestClient(starlette_basic.app) as client: + response = client.get("/api/status") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + assert response.json()["status"] == "ok" + assert 'id="debug-toolbar"' not in response.text + + +def test_static_css_asset(starlette_basic: ModuleType) -> None: + with TestClient(starlette_basic.app) as client: + response = client.get("/_debug_toolbar/static/toolbar.css") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/css") + assert response.text + + +def test_static_js_asset(starlette_basic: ModuleType) -> None: + with TestClient(starlette_basic.app) as client: + response = client.get("/_debug_toolbar/static/toolbar.js") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/javascript") + assert response.text + + +def test_history_page_lists_prior_request(starlette_basic: ModuleType) -> None: + with TestClient(starlette_basic.app) as client: + client.get("/") + response = client.get("/_debug_toolbar/") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + assert b"No requests recorded yet" not in response.content + assert b"GET" in response.content + + +def test_history_api_lists_requests_after_html_requests(starlette_basic: ModuleType) -> None: + with TestClient(starlette_basic.app) as client: + client.get("/") + client.get("/about") + response = client.get("/_debug_toolbar/api/requests") + + assert response.status_code == 200 + data = response.json() + assert isinstance(data, list) + assert len(data) == 2 + + +def test_request_detail_unknown_id_returns_404(starlette_basic: ModuleType) -> None: + with TestClient(starlette_basic.app) as client: + response = client.get(f"/_debug_toolbar/api/requests/{uuid4()}") + + assert response.status_code == 404 + + +def test_request_detail_contains_multiple_panels(starlette_basic: ModuleType) -> None: + with TestClient(starlette_basic.app) as client: + client.get("/") + requests_response = client.get("/_debug_toolbar/api/requests") + request_id = requests_response.json()[0]["request_id"] + detail_response = client.get(f"/_debug_toolbar/api/requests/{request_id}") + + assert detail_response.status_code == 200 + assert detail_response.headers["content-type"].startswith("application/json") + detail = detail_response.json() + + panel_data = detail["panel_data"] + expected_panels = {"TimerPanel", "RequestPanel", "ResponsePanel", "LoggingPanel", "VersionsPanel"} + assert expected_panels <= set(panel_data) + + request_stats = panel_data["RequestPanel"] + assert request_stats["method"] == "GET" + assert request_stats["path"] == "/" + + timer_stats = panel_data["TimerPanel"] + assert timer_stats["total_time_ms"] >= 0 diff --git a/tests/examples/test_templates_panel_example.py b/tests/examples/test_templates_panel_example.py new file mode 100644 index 0000000..7bb7169 --- /dev/null +++ b/tests/examples/test_templates_panel_example.py @@ -0,0 +1,32 @@ +"""Tests for the templates panel example script.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from types import ModuleType + + +@pytest.mark.asyncio +async def test_main_reports_real_jinja2_render_statistics( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """Running the real script should report three real Jinja2 template renders.""" + module = import_example("examples.templates_panel_example") + + await module.main() + + output = capsys.readouterr().out + assert "Template 1 result: Hello World!" in output + assert "Template 2 result:" in output + assert "Template 3 result:" in output + assert "Total renders: 3" in output + assert "Engines used: jinja2" in output + assert output.count("Template:") == 3 + assert output.count("Engine: jinja2") == 3 + assert "Context keys: name" in output diff --git a/tests/examples/test_websocket_panel_example.py b/tests/examples/test_websocket_panel_example.py new file mode 100644 index 0000000..07a5c2e --- /dev/null +++ b/tests/examples/test_websocket_panel_example.py @@ -0,0 +1,94 @@ +"""Tests for the WebSocket Panel example application.""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest +from litestar.testing import TestClient + +from debug_toolbar.core.panels.websocket import WebSocketPanel + +if TYPE_CHECKING: + from collections.abc import Iterator + from types import ModuleType + + +@pytest.fixture(autouse=True) +def _clear_websocket_connections() -> Iterator[None]: + """Isolate the module-level WebSocketPanel connection registry between tests.""" + WebSocketPanel._active_connections.clear() + yield + WebSocketPanel._active_connections.clear() + + +@pytest.fixture +def app_module(import_example: Callable[[str], ModuleType]) -> ModuleType: + return import_example("examples.websocket_panel_example") + + +def test_index_page_shows_toolbar(app_module: ModuleType) -> None: + with TestClient(app_module.app) as client: + response = client.get("/") + + assert response.status_code == 200 + assert 'id="debug-toolbar"' in response.text + + +def test_echo_websocket_echoes_message(app_module: ModuleType) -> None: + with TestClient(app_module.app) as client, client.websocket_connect("/ws/echo") as ws: + ws.send_text("hello") + response = ws.receive_text() + assert response == "Echo: hello" + ws.close() + + +def test_chat_websocket_broadcasts_and_tracks_connections(app_module: ModuleType) -> None: + with TestClient(app_module.app) as client: + with client.websocket_connect("/ws/chat?name=Alice") as ws_alice: + alice_join = json.loads(ws_alice.receive_text()) + assert alice_join == {"type": "system", "text": "Alice joined the chat"} + + with client.websocket_connect("/ws/chat?name=Bob") as ws_bob: + alice_sees_bob_join = json.loads(ws_alice.receive_text()) + bob_join = json.loads(ws_bob.receive_text()) + assert alice_sees_bob_join == {"type": "system", "text": "Bob joined the chat"} + assert bob_join == {"type": "system", "text": "Bob joined the chat"} + + connections = client.get("/api/connections").json() + assert connections["active_connections"] == 2 + + ws_alice.send_text("Hello Bob") + broadcast = json.loads(ws_bob.receive_text()) + assert broadcast["type"] == "message" + assert broadcast["user"] == "Alice" + assert broadcast["text"] == "Hello Bob" + + bob_left = json.loads(ws_alice.receive_text()) + assert bob_left == {"type": "system", "text": "Bob left the chat"} + + connections_after = client.get("/api/connections").json() + assert connections_after["active_connections"] == 0 + + +def test_websocket_panel_records_closed_connection_stats(app_module: ModuleType) -> None: + with TestClient(app_module.app) as client: + with client.websocket_connect("/ws/echo") as ws: + ws.send_text("ping") + assert ws.receive_text() == "Echo: ping" + ws.close() + + html_response = client.get("/") + assert html_response.status_code == 200 + + requests = client.get("/_debug_toolbar/api/requests").json() + root_request = next(req for req in requests if req["metadata"].get("path") == "/") + + detail = client.get(f"/_debug_toolbar/api/requests/{root_request['request_id']}").json() + ws_stats = detail["panel_data"]["WebSocketPanel"] + + assert ws_stats["total_closed"] >= 1 + closed_paths = [conn["path"] for conn in ws_stats["closed_connections"]] + assert "/ws/echo" in closed_paths