From 1d3d4a7f94098315df7e22442897453e416228df Mon Sep 17 00:00:00 2001 From: Jacob Coffee Date: Sat, 10 Oct 2026 19:51:36 -0500 Subject: [PATCH 1/7] fix: cascade-delete posts when deleting a user in advanced-alchemy example DELETE /api/users/{id} raised a 500 IntegrityError for any user with posts, since the default relationship behaviour nulls out the foreign key instead of deleting dependents, and Post.author_id is non-nullable. Found while writing tests for the example app. --- examples/litestar_advanced_alchemy/models.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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): From 985e8af650702f6d28774304e3de3c1346339e1c Mon Sep 17 00:00:00 2001 From: Jacob Coffee Date: Sat, 10 Oct 2026 19:51:42 -0500 Subject: [PATCH 2/7] test: add full test coverage for example apps Add tests/examples/ covering every app under examples/: asgi_basic, litestar_basic, litestar_advanced_alchemy, starlette_basic, fastapi_basic, websocket_panel_example, graphql_panel_example, mcp_server_example, and the standalone panel demo scripts (cache, templates, settings, profiling, memory, flamegraph, async_profiler, alerts). Each app is exercised through its own framework's real test client: documented routes are checked for status/content-type, toolbar HTML injection and static assets, and the per-request panel data the example is built to demonstrate (SQL N+1 detection, GraphQL resolver tracking, websocket connection stats, async profiler stats, and so on). A shared tests/examples/conftest.py fixture (import_example) imports each example fresh per test and evicts it afterwards, since several examples build module-level toolbar/app singletons at import time. Add make test-examples to run just this suite. --- Makefile | 5 +- tests/examples/__init__.py | 1 + tests/examples/conftest.py | 52 ++++ tests/examples/test_alerts_panel_example.py | 184 ++++++++++++ tests/examples/test_asgi_basic.py | 136 +++++++++ tests/examples/test_async_profiler_example.py | 168 +++++++++++ tests/examples/test_cache_panel_demo.py | 104 +++++++ tests/examples/test_fastapi_basic.py | 135 +++++++++ tests/examples/test_flamegraph_example.py | 72 +++++ tests/examples/test_graphql_panel_example.py | 107 +++++++ .../test_litestar_advanced_alchemy.py | 263 ++++++++++++++++++ tests/examples/test_litestar_basic.py | 130 +++++++++ tests/examples/test_mcp_server_example.py | 151 ++++++++++ tests/examples/test_memory_panel_example.py | 92 ++++++ .../examples/test_profiling_panel_example.py | 107 +++++++ tests/examples/test_settings_panel_example.py | 128 +++++++++ tests/examples/test_starlette_basic.py | 113 ++++++++ .../examples/test_templates_panel_example.py | 79 ++++++ .../examples/test_websocket_panel_example.py | 94 +++++++ 19 files changed, 2120 insertions(+), 1 deletion(-) create mode 100644 tests/examples/__init__.py create mode 100644 tests/examples/conftest.py create mode 100644 tests/examples/test_alerts_panel_example.py create mode 100644 tests/examples/test_asgi_basic.py create mode 100644 tests/examples/test_async_profiler_example.py create mode 100644 tests/examples/test_cache_panel_demo.py create mode 100644 tests/examples/test_fastapi_basic.py create mode 100644 tests/examples/test_flamegraph_example.py create mode 100644 tests/examples/test_graphql_panel_example.py create mode 100644 tests/examples/test_litestar_advanced_alchemy.py create mode 100644 tests/examples/test_litestar_basic.py create mode 100644 tests/examples/test_mcp_server_example.py create mode 100644 tests/examples/test_memory_panel_example.py create mode 100644 tests/examples/test_profiling_panel_example.py create mode 100644 tests/examples/test_settings_panel_example.py create mode 100644 tests/examples/test_starlette_basic.py create mode 100644 tests/examples/test_templates_panel_example.py create mode 100644 tests/examples/test_websocket_panel_example.py 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/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..749159f --- /dev/null +++ b/tests/examples/test_alerts_panel_example.py @@ -0,0 +1,184 @@ +"""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_prints_expected_output( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """The script's main() should report real alerts for each scenario.""" + module = import_example("examples.alerts_panel_example") + + await module.main() + + output = capsys.readouterr().out + assert "Total Alerts: 3" in output + assert "Content-Security-Policy" in output + + +@pytest.mark.asyncio +async def test_alerts_panel_missing_security_headers( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 1's missing headers scenario and assert real alerts generated.""" + module = import_example("examples.alerts_panel_example") + + config = module.DebugToolbarConfig(enabled=True) + toolbar = module.DebugToolbar(config=config) + panel = module.AlertsPanel(toolbar) + context = await toolbar.process_request() + + context.metadata["headers"] = { + "content-type": "text/html", + "user-agent": "Mozilla/5.0", + } + context.metadata["response_headers"] = { + "content-type": "text/html; charset=utf-8", + "content-length": "1024", + } + + headers_panel_data = { + "response_headers": { + "security_headers": { + "missing": [ + { + "name": "Content-Security-Policy", + "description": "Prevents XSS and other code injection attacks", + }, + { + "name": "X-Content-Type-Options", + "description": "Prevents MIME-sniffing attacks", + }, + { + "name": "Strict-Transport-Security", + "description": "Forces HTTPS connections", + }, + ] + } + } + } + context.panel_data["HeadersPanel"] = headers_panel_data + + stats = await panel.generate_stats(context) + + assert stats["total_alerts"] == 3 + titles = {alert["title"] for alert in stats["alerts"]} + assert any("Content-Security-Policy" in title for title in titles) + assert all(alert["category"] == "security" for alert in stats["alerts"]) + + +@pytest.mark.asyncio +async def test_alerts_panel_debug_mode_in_production( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 4's debug-in-production scenario and assert a configuration alert fires.""" + module = import_example("examples.alerts_panel_example") + + toolbar4 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) + panel4 = module.AlertsPanel(toolbar4) + context4 = await toolbar4.process_request() + + context4.metadata["headers"] = {} + context4.metadata["response_headers"] = {} + context4.panel_data["SettingsPanel"] = { + "debug": True, + "environment": "production", + } + + stats4 = await panel4.generate_stats(context4) + + configuration_alerts = [alert for alert in stats4["alerts"] if alert["category"] == "configuration"] + assert len(configuration_alerts) >= 1 + + +@pytest.mark.asyncio +async def test_alerts_panel_n_plus_one_detection( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 7's N+1 query scenario and assert a database alert fires.""" + module = import_example("examples.alerts_panel_example") + + toolbar7 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) + panel7 = module.AlertsPanel(toolbar7) + context7 = await toolbar7.process_request() + + context7.metadata["headers"] = {} + context7.metadata["response_headers"] = {} + context7.panel_data["SQLAlchemyPanel"] = { + "queries": [], + "n_plus_one_groups": [ + { + "count": 15, + "origin_display": "app.py:42 in get_users()", + "suggestion": "Use joinedload(User.posts) to fetch related posts in a single query.", + } + ], + } + + stats7 = await panel7.generate_stats(context7) + + n_plus_one_alerts = [alert for alert in stats7["alerts"] if "n+1" in alert["title"].lower()] + assert len(n_plus_one_alerts) == 1 + assert "joinedload" in n_plus_one_alerts[0]["suggestion"] + + +@pytest.mark.asyncio +async def test_alerts_panel_multiple_categories_summary( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 8's combined scenario and assert alerts are split across categories.""" + module = import_example("examples.alerts_panel_example") + + toolbar8 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) + panel8 = module.AlertsPanel(toolbar8) + context8 = await toolbar8.process_request() + + context8.metadata["method"] = "POST" + context8.metadata["headers"] = {"content-type": "text/html"} + context8.metadata["response_headers"] = { + "content-length": str(2 * 1024 * 1024), + "set-cookie": "token=xyz; Path=/", + } + + context8.panel_data["HeadersPanel"] = { + "response_headers": { + "security_headers": { + "missing": [ + { + "name": "X-Frame-Options", + "description": "Prevents clickjacking attacks", + } + ] + } + } + } + + context8.panel_data["SQLAlchemyPanel"] = { + "queries": [ + {"duration_ms": 250.0, "sql": "SELECT * FROM large_table"}, + ], + "n_plus_one_groups": [ + { + "count": 5, + "origin_display": "views.py:123 in list_items()", + "suggestion": "Use selectinload() to optimize the query.", + } + ], + } + + stats8 = await panel8.generate_stats(context8) + + assert stats8["total_alerts"] == len(stats8["alerts"]) + assert sum(stats8["by_category"].values()) == stats8["total_alerts"] + assert stats8["by_category"].get("security", 0) >= 1 + assert stats8["by_category"].get("database", 0) >= 1 diff --git a/tests/examples/test_asgi_basic.py b/tests/examples/test_asgi_basic.py new file mode 100644 index 0000000..c614a04 --- /dev/null +++ b/tests/examples/test_asgi_basic.py @@ -0,0 +1,136 @@ +"""Tests for the raw-ASGI ``examples/asgi_basic`` example. + +This example hand-rolls its own debug-toolbar middleware directly against the +``DebugToolbar`` core API, predating the framework integrations' HTML +injection and static-asset routes. On this branch (ahead of PR #61) it does +not inject the toolbar markup into HTML responses and does not serve +``/_debug_toolbar/static/*`` assets at all -- only the two hand-rolled JSON +endpoints (``/_debug_toolbar/requests`` and ``/_debug_toolbar/request/``) +exist. These tests assert the behavior that is actually present today. +""" + +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_returns_html(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 + + +async def test_home_page_does_not_inject_toolbar_markup(asgi_basic: ModuleType) -> None: + """The hand-rolled middleware never rewrites the response body (PR #61).""" + async with _client(asgi_basic) as client: + response = await client.get("/") + + assert 'id="debug-toolbar"' not in response.text + + +async def test_about_page_returns_html(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 + + +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["items"] == [1, 2, 3] + 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("/nope") + + 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_debug_toolbar_requests_lists_recorded_request(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + await client.get("/") + response = await client.get("/_debug_toolbar/requests") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("application/json") + requests_list = response.json() + assert len(requests_list) == 1 + assert requests_list[0]["method"] == "GET" + assert requests_list[0]["path"] == "/" + + +async def test_debug_toolbar_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/requests") + request_id = requests_response.json()[0]["request_id"] + detail_response = await client.get(f"/_debug_toolbar/request/{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_request_detail_unknown_id_returns_404(asgi_basic: ModuleType) -> None: + async with _client(asgi_basic) as client: + response = await client.get(f"/_debug_toolbar/request/{uuid4()}") + + assert response.status_code == 404 + + +async def test_static_assets_are_not_served_yet(asgi_basic: ModuleType) -> None: + """No ``/_debug_toolbar/static/*`` route exists in this example (PR #61).""" + async with _client(asgi_basic) as client: + css_response = await client.get("/_debug_toolbar/static/toolbar.css") + js_response = await client.get("/_debug_toolbar/static/toolbar.js") + + assert css_response.status_code == 404 + assert js_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..c3f5a5d --- /dev/null +++ b/tests/examples/test_async_profiler_example.py @@ -0,0 +1,168 @@ +"""Tests for the async profiler panel example script.""" + +from __future__ import annotations + +import asyncio +import time +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_prints_expected_output( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """The script's main() should report real task counts and durations.""" + module = import_example("examples.async_profiler_example") + + await module.main() + + output = capsys.readouterr().out + assert "Total tasks created: 5" in output + assert "Completed tasks: 5" in output + assert "Total tasks: 4" in output + assert "Nested task results: [0, 2, 4]" in output + + +@pytest.mark.asyncio +async def test_async_profiler_tracks_task_completion( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 1's setup and assert on the real summary stats.""" + module = import_example("examples.async_profiler_example") + + config = module.DebugToolbarConfig( + enabled=True, + async_profiler_backend="taskfactory", + async_capture_task_stacks=True, + async_max_stack_depth=5, + ) + toolbar = module.DebugToolbar(config=config) + panel = module.AsyncProfilerPanel(toolbar) + context = await toolbar.process_request() + + await panel.process_request(context) + + tasks = [asyncio.create_task(module.fast_task(i), name=f"fast_task_{i}") for i in range(5)] + await asyncio.gather(*tasks) + + await panel.process_response(context) + stats = await panel.generate_stats(context) + + assert stats["backend"] == "taskfactory" + assert stats["summary"]["total_tasks"] == 5 + assert stats["summary"]["completed_tasks"] == 5 + created = [task for task in stats["tasks"] if task["event_type"] == "created"] + assert len(created) == 5 + assert all(task["coro_name"] == "fast_task" for task in created) + + +@pytest.mark.asyncio +async def test_async_profiler_detects_slow_tasks( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 2's mixed fast/slow tasks and assert durations separate them.""" + module = import_example("examples.async_profiler_example") + + config = module.DebugToolbarConfig( + enabled=True, + async_profiler_backend="taskfactory", + async_capture_task_stacks=True, + async_max_stack_depth=5, + ) + toolbar2 = module.DebugToolbar(config=config) + panel2 = module.AsyncProfilerPanel(toolbar2) + context2 = await toolbar2.process_request() + + await panel2.process_request(context2) + + mixed_tasks = [ + asyncio.create_task(module.fast_task(1), name="fast_1"), + asyncio.create_task(module.slow_task(2), name="slow_1"), + asyncio.create_task(module.fast_task(3), name="fast_2"), + asyncio.create_task(module.slow_task(4), name="slow_2"), + ] + await asyncio.gather(*mixed_tasks) + + await panel2.process_response(context2) + stats2 = await panel2.generate_stats(context2) + + assert stats2["summary"]["total_tasks"] == 4 + + completed = [task for task in stats2["tasks"] if task["event_type"] == "completed" and task.get("duration_ms")] + durations_by_coro: dict[str, list[float]] = {} + for task in completed: + durations_by_coro.setdefault(task["coro_name"], []).append(task["duration_ms"]) + + assert all(duration < 50 for duration in durations_by_coro["fast_task"]) + assert all(duration >= 90 for duration in durations_by_coro["slow_task"]) + + +@pytest.mark.asyncio +async def test_async_profiler_event_loop_lag_monitoring( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 5's event loop lag monitoring and assert real lag was recorded.""" + module = import_example("examples.async_profiler_example") + + config_lag = module.DebugToolbarConfig( + enabled=True, + async_profiler_backend="taskfactory", + async_enable_event_loop_monitoring=True, + async_event_loop_lag_threshold_ms=5.0, + ) + toolbar4 = module.DebugToolbar(config=config_lag) + panel4 = module.AsyncProfilerPanel(toolbar4) + context4 = await toolbar4.process_request() + + await panel4.process_request(context4) + + for _ in range(10): + await asyncio.sleep(0.01) + time.sleep(0.02) # noqa: ASYNC251 + + await panel4.process_response(context4) + stats4 = await panel4.generate_stats(context4) + + assert stats4["summary"]["max_lag_ms"] > 0 + assert stats4["event_loop_lag"] + + +@pytest.mark.asyncio +async def test_async_profiler_nested_task_parent_tracking( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 6's nested task tracking and assert parent links are recorded.""" + module = import_example("examples.async_profiler_example") + + config = module.DebugToolbarConfig( + enabled=True, + async_profiler_backend="taskfactory", + async_capture_task_stacks=True, + async_max_stack_depth=5, + ) + toolbar5 = module.DebugToolbar(config=config) + panel5 = module.AsyncProfilerPanel(toolbar5) + context5 = await toolbar5.process_request() + + await panel5.process_request(context5) + + result = await asyncio.create_task(module.nested_tasks(), name="parent_task") + assert result == [0, 2, 4] + + await panel5.process_response(context5) + stats5 = await panel5.generate_stats(context5) + + created_tasks = [task for task in stats5["tasks"] if task["event_type"] == "created"] + parent_ids = [task["parent_task_id"] for task in created_tasks if task["parent_task_id"] is not None] + counts: dict[str, int] = {} + for parent_id in parent_ids: + counts[parent_id] = counts.get(parent_id, 0) + 1 + + assert 3 in counts.values() diff --git a/tests/examples/test_cache_panel_demo.py b/tests/examples/test_cache_panel_demo.py new file mode 100644 index 0000000..c483685 --- /dev/null +++ b/tests/examples/test_cache_panel_demo.py @@ -0,0 +1,104 @@ +"""Tests for the cache panel demo script.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from types import ModuleType + + +def test_demo_context_manager_tracking_records_operations( + import_example: Callable[[str], ModuleType], +) -> None: + """Demo should track a mix of hits, misses, and backends.""" + module = import_example("examples.cache_panel_demo") + + module.demo_context_manager_tracking() + + +def test_demo_context_manager_tracking_builds_real_operations( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate the demo's tracker usage directly to assert on real fields.""" + module = import_example("examples.cache_panel_demo") + + tracker = module.CacheTracker() + + with tracker.track_operation("GET", "user:123", "redis") as extra: + extra["hit"] = True + with tracker.track_operation("GET", "user:456", "redis") as extra: + extra["hit"] = False + with tracker.track_operation("SET", "user:456", "redis") as extra: + extra["ttl"] = 3600 + with tracker.track_operation("GET", "session:abc123", "memcached") as extra: + extra["hit"] = True + + assert len(tracker.operations) == 4 + + first, second, third, fourth = tracker.operations + assert first.operation == "GET" + assert first.key == "user:123" + assert first.hit is True + assert first.backend == "redis" + + assert second.hit is False + + assert third.operation == "SET" + assert third.extra["ttl"] == 3600 + + assert fourth.backend == "memcached" + assert fourth.hit is True + + +def test_demo_multi_key_operation_tracks_list_key( + import_example: Callable[[str], ModuleType], +) -> None: + """Multi-key operations should preserve the list of keys and extras.""" + module = import_example("examples.cache_panel_demo") + + module.demo_multi_key_operation() + + tracker = module.CacheTracker() + with tracker.track_operation("MGET", ["key1", "key2", "key3"], "redis") as extra: + extra["hit"] = True + extra["keys_found"] = 2 + + op = tracker.operations[0] + assert op.operation == "MGET" + assert op.key == ["key1", "key2", "key3"] + assert op.hit is True + assert op.extra["keys_found"] == 2 + + +def test_demo_statistics_computes_hit_rate( + import_example: Callable[[str], ModuleType], +) -> None: + """Statistics demo should produce computable totals, hits, misses, and hit rate.""" + module = import_example("examples.cache_panel_demo") + + module.demo_statistics() + + tracker = module.CacheTracker() + with tracker.track_operation("GET", "key1", "redis") as extra: + extra["hit"] = True + with tracker.track_operation("GET", "key2", "redis") as extra: + extra["hit"] = True + with tracker.track_operation("GET", "key3", "redis") as extra: + extra["hit"] = False + with tracker.track_operation("SET", "key4", "redis"): + pass + with tracker.track_operation("GET", "key5", "memcached") as extra: + extra["hit"] = True + + total = len(tracker.operations) + hits = sum(1 for op in tracker.operations if op.hit is True) + misses = sum(1 for op in tracker.operations if op.hit is False) + + assert total == 5 + assert hits == 3 + assert misses == 1 + + hit_rate = (hits / (hits + misses) * 100) if (hits + misses) > 0 else 0.0 + assert hit_rate == 75.0 diff --git a/tests/examples/test_fastapi_basic.py b/tests/examples/test_fastapi_basic.py new file mode 100644 index 0000000..01f4f2c --- /dev/null +++ b/tests/examples/test_fastapi_basic.py @@ -0,0 +1,135 @@ +"""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_is_present(fastapi_basic: ModuleType) -> None: + """Only asserts the panel key exists; its stats are covered by an in-flight fix.""" + 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}") + + assert "DependencyInjectionPanel" in detail_response.json()["panel_data"] diff --git a/tests/examples/test_flamegraph_example.py b/tests/examples/test_flamegraph_example.py new file mode 100644 index 0000000..dfaf4fb --- /dev/null +++ b/tests/examples/test_flamegraph_example.py @@ -0,0 +1,72 @@ +"""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_writes_valid_speedscope_json( + import_example: Callable[[str], ModuleType], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """The script's main() 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 + + 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 + + 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"]) + + +def test_flamegraph_generator_produces_fibonacci_frames( + import_example: Callable[[str], ModuleType], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Replicate the example's profiling setup directly and inspect the frame names.""" + monkeypatch.chdir(tmp_path) + module = import_example("examples.flamegraph_example") + + import cProfile + + from debug_toolbar.core.panels.flamegraph import FlameGraphGenerator + + profiler = cProfile.Profile() + profiler.enable() + module.compute_intensive_task() + module.data_processing() + profiler.disable() + + generator = FlameGraphGenerator(profiler) + speedscope_json = generator.generate() + + frame_names = [frame["name"] for frame in speedscope_json["shared"]["frames"]] + assert any("fibonacci" in name for name in frame_names) + assert any("data_processing" in name for name in frame_names) 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..a71c2bb --- /dev/null +++ b/tests/examples/test_memory_panel_example.py @@ -0,0 +1,92 @@ +"""Tests for the memory 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_prints_expected_output( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """The script's main() should report real memory tracking across 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 + + +@pytest.mark.asyncio +async def test_memory_panel_tracks_allocation( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 1's basic tracemalloc tracking and assert on real stats.""" + module = import_example("examples.memory_panel_example") + + config = module.DebugToolbarConfig(enabled=True) + toolbar = module.DebugToolbar(config=config) + toolbar.config.memory_backend = "tracemalloc" + + panel = module.MemoryPanel(toolbar) + context = await toolbar.process_request() + + await panel.process_request(context) + + data = module.allocate_memory(5) + nested = module.create_nested_structures() + + await panel.process_response(context) + + stats = await panel.generate_stats(context) + + assert stats["backend"] == "tracemalloc" + assert stats["memory_before"] >= 0 + assert stats["memory_after"] >= 0 + assert stats["memory_delta"] > 0 + assert stats["peak_memory"] > 0 + assert len(stats["top_allocations"]) > 0 + + top_alloc = stats["top_allocations"][0] + assert top_alloc["size"] > 0 + assert top_alloc["count"] > 0 + + del data + del nested + + +@pytest.mark.asyncio +async def test_memory_panel_auto_backend_selection( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 2's auto backend selection and assert a real backend is chosen.""" + module = import_example("examples.memory_panel_example") + + toolbar2 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) + toolbar2.config.memory_backend = "auto" + + panel2 = module.MemoryPanel(toolbar2) + context2 = await toolbar2.process_request() + + await panel2.process_request(context2) + + temp_data = [i**3 for i in range(100000)] + + await panel2.process_response(context2) + stats2 = await panel2.generate_stats(context2) + + assert stats2["backend"] in {"tracemalloc", "memray"} + assert stats2["memory_delta"] >= 0 + assert stats2["peak_memory"] >= 0 + + del temp_data diff --git a/tests/examples/test_profiling_panel_example.py b/tests/examples/test_profiling_panel_example.py new file mode 100644 index 0000000..efcd451 --- /dev/null +++ b/tests/examples/test_profiling_panel_example.py @@ -0,0 +1,107 @@ +"""Tests for the profiling panel example script.""" + +from __future__ import annotations + +import json +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_prints_expected_output( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """The script's main() should report cProfile results with real function counts.""" + monkeypatch.chdir(tmp_path) + module = import_example("examples.profiling_panel_example") + + await module.main() + + output = capsys.readouterr().out + assert "Backend: cprofile" 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 + + +@pytest.mark.asyncio +async def test_profiling_panel_tracks_function_calls( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 1's setup to assert on real profiling stats.""" + module = import_example("examples.profiling_panel_example") + + config = module.DebugToolbarConfig(enabled=True) + toolbar = module.DebugToolbar(config=config) + toolbar.config.profiler_backend = "cprofile" + toolbar.config.profiler_top_functions = 10 + toolbar.config.profiler_sort_by = "cumulative" + + panel = module.ProfilingPanel(toolbar) + context = await toolbar.process_request() + + await panel.process_request(context) + + module.compute_intensive_task() + module.io_simulation() + + await panel.process_response(context) + + stats = await panel.generate_stats(context) + + assert stats["backend"] == "cprofile" + assert stats["total_time"] > 0 + assert stats["function_calls"] > 0 + assert stats["primitive_calls"] > 0 + assert len(stats["top_functions"]) > 0 + + top_func = stats["top_functions"][0] + assert top_func["calls"] > 0 + assert top_func["cumulative_time"] >= 0 + + +@pytest.mark.asyncio +async def test_profiling_panel_flamegraph_data( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 7's flame graph generation and assert on the speedscope JSON.""" + module = import_example("examples.profiling_panel_example") + + toolbar3 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) + toolbar3.config.profiler_backend = "cprofile" + toolbar3.config.enable_flamegraph = True + + panel3 = module.ProfilingPanel(toolbar3) + context3 = await toolbar3.process_request() + + await panel3.process_request(context3) + + module.compute_intensive_task() + + await panel3.process_response(context3) + stats3 = await panel3.generate_stats(context3) + + assert stats3["flamegraph_available"] is True + flamegraph = stats3["flamegraph_data"] + assert flamegraph["$schema"] + assert len(flamegraph["shared"]["frames"]) > 0 + assert len(flamegraph["profiles"]) > 0 + + profile_data = flamegraph["profiles"][0] + assert profile_data["type"] + assert profile_data["endValue"] >= 0 + assert len(profile_data["samples"]) > 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..838c396 --- /dev/null +++ b/tests/examples/test_settings_panel_example.py @@ -0,0 +1,128 @@ +"""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_prints_expected_output( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """The script's main() should report real toolbar, custom, and redacted settings.""" + module = import_example("examples.settings_panel_example") + + await module.main() + + output = capsys.readouterr().out + assert "Toolbar Config: True" in output + assert "App Name: My Litestar App" in output + assert "Database Host: localhost" in output + assert "Database Password: " in output + assert "secret123" not in output + + +@pytest.mark.asyncio +async def test_settings_panel_basic_stats( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 1's basic panel setup and assert on real stats fields.""" + module = import_example("examples.settings_panel_example") + + config = module.DebugToolbarConfig( + enabled=True, + intercept_redirects=False, + max_request_history=100, + ) + toolbar = module.DebugToolbar(config=config) + + basic_panel = module.SettingsPanel(toolbar) + context = await toolbar.process_request() + stats = await basic_panel.generate_stats(context) + + assert stats["toolbar_config"]["enabled"] is True + assert "debug" in stats["python_settings"] + assert "variables" in stats["environment"] + assert isinstance(stats["environment"]["redacted_count"], int) + + +@pytest.mark.asyncio +async def test_settings_panel_redacts_sensitive_custom_settings( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 2's custom settings and assert the password/api key are redacted.""" + module = import_example("examples.settings_panel_example") + + toolbar = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) + + custom_settings = { + "app_name": "My Litestar App", + "version": "1.0.0", + "database": { + "host": "localhost", + "port": 5432, + "password": "secret123", + }, + "api_key": "abc123xyz", + "debug_mode": True, + } + + custom_panel = module.SettingsPanel(toolbar, custom_settings=custom_settings) + context2 = await toolbar.process_request() + stats2 = await custom_panel.generate_stats(context2) + + assert stats2["custom_settings"]["app_name"] == "My Litestar App" + assert stats2["custom_settings"]["database"]["host"] == "localhost" + assert stats2["custom_settings"]["database"]["password"] == module.SettingsPanel.REDACTED_VALUE + assert stats2["custom_settings"]["api_key"] == module.SettingsPanel.REDACTED_VALUE + + +@pytest.mark.asyncio +async def test_settings_panel_without_environment_section( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 3's show_env=False behavior.""" + module = import_example("examples.settings_panel_example") + + toolbar = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) + + no_env_panel = module.SettingsPanel(toolbar, show_env=False) + context3 = await toolbar.process_request() + stats3 = await no_env_panel.generate_stats(context3) + + assert stats3["environment"] is None + + +@pytest.mark.asyncio +async def test_settings_panel_custom_sensitive_patterns( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate example 4's custom sensitive key patterns.""" + module = import_example("examples.settings_panel_example") + + toolbar = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) + + sensitive_panel = module.SettingsPanel( + toolbar, + custom_settings={ + "app_name": "My App", + "internal_id": "secret-internal-123", + "confidential_data": "top-secret", + "public_url": "https://example.com", + }, + sensitive_keys=["INTERNAL", "CONFIDENTIAL"], + ) + context4 = await toolbar.process_request() + stats4 = await sensitive_panel.generate_stats(context4) + + assert stats4["custom_settings"]["app_name"] == "My App" + assert stats4["custom_settings"]["internal_id"] == module.SettingsPanel.REDACTED_VALUE + assert stats4["custom_settings"]["confidential_data"] == module.SettingsPanel.REDACTED_VALUE + assert stats4["custom_settings"]["public_url"] == "https://example.com" 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..b97af46 --- /dev/null +++ b/tests/examples/test_templates_panel_example.py @@ -0,0 +1,79 @@ +"""Tests for the templates panel example script.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING + +import pytest + +from debug_toolbar.core.context import RequestContext, set_request_context + +if TYPE_CHECKING: + from types import ModuleType + + +@pytest.mark.asyncio +async def test_main_prints_expected_summary( + import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], +) -> None: + """The script's main() should report three renders across the jinja2 engine.""" + module = import_example("examples.templates_panel_example") + + await module.main() + + output = capsys.readouterr().out + assert "Total renders: 3" in output + assert "Engines used: jinja2" in output + assert "Template 1 result: Hello World!" in output + + +@pytest.mark.asyncio +async def test_templates_panel_generates_real_stats( + import_example: Callable[[str], ModuleType], +) -> None: + """Replicate the example's panel setup directly to assert on the real stats dict.""" + module = import_example("examples.templates_panel_example") + + config = module.DebugToolbarConfig( + enabled=True, + panels=[ + "debug_toolbar.core.panels.timer.TimerPanel", + "debug_toolbar.core.panels.templates.TemplatesPanel", + ], + ) + toolbar = module.DebugToolbar(config) + templates_panel = toolbar.get_panel("TemplatesPanel") + assert templates_panel is not None + + context = RequestContext() + set_request_context(context) + + await templates_panel.process_request(context) + + from jinja2 import Environment, Template + + template1 = Template("Hello {{ name }}!") + result1 = template1.render(name="World") + assert result1 == "Hello World!" + + env = Environment() + template2 = Template("
{{ title }}
") + result2 = template2.render(title="My List") + assert result2 == "
My List
" + + template3 = env.from_string("{{ greeting }}, {{ name }}!") + result3 = template3.render(greeting="Hi", name="Jinja") + assert result3 == "Hi, Jinja!" + + await templates_panel.process_response(context) + + stats = await templates_panel.generate_stats(context) + + assert stats["total_renders"] == 3 + assert stats["total_time"] >= 0 + assert stats["engines_used"] == ["jinja2"] + assert stats["renders"][0]["template_name"] + assert stats["renders"][1]["template_name"] + assert stats["renders"][2]["template_name"] 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 From 61dd602ba7aeaadbc6f8165b72e31b835f840314 Mon Sep 17 00:00:00 2001 From: Jacob Coffee Date: Sat, 10 Oct 2026 19:51:47 -0500 Subject: [PATCH 3/7] ci: add weekly workflow to test examples against latest dependencies Runs tests/examples against uv sync --all-extras --upgrade (ignoring the lockfile pins) on the oldest and newest supported Python versions, on a weekly schedule, workflow_dispatch, and PRs touching examples/ or tests/examples/. Prints resolved versions of the key frameworks for debugging. Not part of the required CI Success aggregator. Confirmed this actually catches regressions: with mcp upgraded past 2.0, debug_toolbar.mcp.server fails to import because mcp 2.x renamed FastMCP to MCPServer, which this workflow surfaces as a test failure in tests/examples/test_mcp_server_example.py. --- .github/workflows/examples-latest.yml | 60 +++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 .github/workflows/examples-latest.yml diff --git a/.github/workflows/examples-latest.yml b/.github/workflows/examples-latest.yml new file mode 100644 index 0000000..edb1093 --- /dev/null +++ b/.github/workflows/examples-latest.yml @@ -0,0 +1,60 @@ +name: Examples (latest dependencies) + +on: + schedule: + - cron: '0 6 * * 1' + workflow_dispatch: + pull_request: + branches: [main] + paths: + - 'examples/**' + - 'tests/examples/**' + - '.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 From 89018e282704751209efa8b549af87f176149aca Mon Sep 17 00:00:00 2001 From: Jacob Coffee Date: Sat, 10 Oct 2026 19:58:49 -0500 Subject: [PATCH 4/7] test: exercise the real example scripts instead of replicating panel logic The standalone panel demo tests previously had a second test per script that reconstructed the same DebugToolbar/Panel/Config objects the script's main() builds, independent of the script itself. That let a broken script (e.g. a renamed import, wrong panel class, wrong config key) pass the test suite as long as the underlying panel class still worked when built directly. Replace those with a single test per script that imports the module and calls its real main() (or free functions, for cache_panel_demo), capturing real printed output and asserting on specific computed values (counts, ratios, byte totals, timings, redacted fields) that would change if the panel actually broke. Also widen examples-latest.yml's pull_request trigger to include src/** and pyproject.toml, since a library change (not just an examples/tests change) can break examples against upgraded dependencies. --- .github/workflows/examples-latest.yml | 2 + tests/examples/test_alerts_panel_example.py | 177 +++--------------- tests/examples/test_async_profiler_example.py | 172 ++++------------- tests/examples/test_cache_panel_demo.py | 98 +++------- tests/examples/test_flamegraph_example.py | 36 +--- tests/examples/test_memory_panel_example.py | 75 ++------ .../examples/test_profiling_panel_example.py | 87 ++------- tests/examples/test_settings_panel_example.py | 125 +++---------- .../examples/test_templates_panel_example.py | 63 +------ 9 files changed, 155 insertions(+), 680 deletions(-) diff --git a/.github/workflows/examples-latest.yml b/.github/workflows/examples-latest.yml index edb1093..a625f13 100644 --- a/.github/workflows/examples-latest.yml +++ b/.github/workflows/examples-latest.yml @@ -9,6 +9,8 @@ on: paths: - 'examples/**' - 'tests/examples/**' + - 'src/**' + - 'pyproject.toml' - '.github/workflows/examples-latest.yml' concurrency: diff --git a/tests/examples/test_alerts_panel_example.py b/tests/examples/test_alerts_panel_example.py index 749159f..2cf4824 100644 --- a/tests/examples/test_alerts_panel_example.py +++ b/tests/examples/test_alerts_panel_example.py @@ -12,173 +12,46 @@ @pytest.mark.asyncio -async def test_main_prints_expected_output( +async def test_main_reports_real_alerts_for_every_scenario( import_example: Callable[[str], ModuleType], capsys: pytest.CaptureFixture[str], ) -> None: - """The script's main() should report real alerts for each scenario.""" + """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 -@pytest.mark.asyncio -async def test_alerts_panel_missing_security_headers( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 1's missing headers scenario and assert real alerts generated.""" - module = import_example("examples.alerts_panel_example") - - config = module.DebugToolbarConfig(enabled=True) - toolbar = module.DebugToolbar(config=config) - panel = module.AlertsPanel(toolbar) - context = await toolbar.process_request() - - context.metadata["headers"] = { - "content-type": "text/html", - "user-agent": "Mozilla/5.0", - } - context.metadata["response_headers"] = { - "content-type": "text/html; charset=utf-8", - "content-length": "1024", - } - - headers_panel_data = { - "response_headers": { - "security_headers": { - "missing": [ - { - "name": "Content-Security-Policy", - "description": "Prevents XSS and other code injection attacks", - }, - { - "name": "X-Content-Type-Options", - "description": "Prevents MIME-sniffing attacks", - }, - { - "name": "Strict-Transport-Security", - "description": "Forces HTTPS connections", - }, - ] - } - } - } - context.panel_data["HeadersPanel"] = headers_panel_data - - stats = await panel.generate_stats(context) - - assert stats["total_alerts"] == 3 - titles = {alert["title"] for alert in stats["alerts"]} - assert any("Content-Security-Policy" in title for title in titles) - assert all(alert["category"] == "security" for alert in stats["alerts"]) - + # Example 3: missing CSRF protection on a POST request. + assert "CSRF Alerts:" in output -@pytest.mark.asyncio -async def test_alerts_panel_debug_mode_in_production( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 4's debug-in-production scenario and assert a configuration alert fires.""" - module = import_example("examples.alerts_panel_example") + # Example 4: debug mode enabled in production really triggers a configuration alert. + assert "Configuration Alerts:" in output - toolbar4 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) - panel4 = module.AlertsPanel(toolbar4) - context4 = await toolbar4.process_request() + # Example 5: oversized response body. + assert "Performance Alerts:" in output - context4.metadata["headers"] = {} - context4.metadata["response_headers"] = {} - context4.panel_data["SettingsPanel"] = { - "debug": True, - "environment": "production", - } + # Example 6: slow database queries. + assert "Database Performance Alerts:" in output - stats4 = await panel4.generate_stats(context4) + # 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 - configuration_alerts = [alert for alert in stats4["alerts"] if alert["category"] == "configuration"] - assert len(configuration_alerts) >= 1 - - -@pytest.mark.asyncio -async def test_alerts_panel_n_plus_one_detection( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 7's N+1 query scenario and assert a database alert fires.""" - module = import_example("examples.alerts_panel_example") - - toolbar7 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) - panel7 = module.AlertsPanel(toolbar7) - context7 = await toolbar7.process_request() - - context7.metadata["headers"] = {} - context7.metadata["response_headers"] = {} - context7.panel_data["SQLAlchemyPanel"] = { - "queries": [], - "n_plus_one_groups": [ - { - "count": 15, - "origin_display": "app.py:42 in get_users()", - "suggestion": "Use joinedload(User.posts) to fetch related posts in a single query.", - } - ], - } - - stats7 = await panel7.generate_stats(context7) - - n_plus_one_alerts = [alert for alert in stats7["alerts"] if "n+1" in alert["title"].lower()] - assert len(n_plus_one_alerts) == 1 - assert "joinedload" in n_plus_one_alerts[0]["suggestion"] - - -@pytest.mark.asyncio -async def test_alerts_panel_multiple_categories_summary( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 8's combined scenario and assert alerts are split across categories.""" - module = import_example("examples.alerts_panel_example") + # 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 - toolbar8 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) - panel8 = module.AlertsPanel(toolbar8) - context8 = await toolbar8.process_request() - - context8.metadata["method"] = "POST" - context8.metadata["headers"] = {"content-type": "text/html"} - context8.metadata["response_headers"] = { - "content-length": str(2 * 1024 * 1024), - "set-cookie": "token=xyz; Path=/", - } - - context8.panel_data["HeadersPanel"] = { - "response_headers": { - "security_headers": { - "missing": [ - { - "name": "X-Frame-Options", - "description": "Prevents clickjacking attacks", - } - ] - } - } - } - - context8.panel_data["SQLAlchemyPanel"] = { - "queries": [ - {"duration_ms": 250.0, "sql": "SELECT * FROM large_table"}, - ], - "n_plus_one_groups": [ - { - "count": 5, - "origin_display": "views.py:123 in list_items()", - "suggestion": "Use selectinload() to optimize the query.", - } - ], - } - - stats8 = await panel8.generate_stats(context8) - - assert stats8["total_alerts"] == len(stats8["alerts"]) - assert sum(stats8["by_category"].values()) == stats8["total_alerts"] - assert stats8["by_category"].get("security", 0) >= 1 - assert stats8["by_category"].get("database", 0) >= 1 + assert "Examples completed successfully!" in output diff --git a/tests/examples/test_async_profiler_example.py b/tests/examples/test_async_profiler_example.py index c3f5a5d..7b0783f 100644 --- a/tests/examples/test_async_profiler_example.py +++ b/tests/examples/test_async_profiler_example.py @@ -2,8 +2,7 @@ from __future__ import annotations -import asyncio -import time +import re from collections.abc import Callable from typing import TYPE_CHECKING @@ -14,155 +13,56 @@ @pytest.mark.asyncio -async def test_main_prints_expected_output( +async def test_main_reports_real_task_tracking( import_example: Callable[[str], ModuleType], capsys: pytest.CaptureFixture[str], ) -> None: - """The script's main() should report real task counts and durations.""" + """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 "Total tasks created: 5" in output - assert "Completed tasks: 5" in output - assert "Total tasks: 4" in output - assert "Nested task results: [0, 2, 4]" in output - - -@pytest.mark.asyncio -async def test_async_profiler_tracks_task_completion( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 1's setup and assert on the real summary stats.""" - module = import_example("examples.async_profiler_example") - - config = module.DebugToolbarConfig( - enabled=True, - async_profiler_backend="taskfactory", - async_capture_task_stacks=True, - async_max_stack_depth=5, - ) - toolbar = module.DebugToolbar(config=config) - panel = module.AsyncProfilerPanel(toolbar) - context = await toolbar.process_request() - - await panel.process_request(context) - - tasks = [asyncio.create_task(module.fast_task(i), name=f"fast_task_{i}") for i in range(5)] - await asyncio.gather(*tasks) - - await panel.process_response(context) - stats = await panel.generate_stats(context) - - assert stats["backend"] == "taskfactory" - assert stats["summary"]["total_tasks"] == 5 - assert stats["summary"]["completed_tasks"] == 5 - created = [task for task in stats["tasks"] if task["event_type"] == "created"] - assert len(created) == 5 - assert all(task["coro_name"] == "fast_task" for task in created) - - -@pytest.mark.asyncio -async def test_async_profiler_detects_slow_tasks( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 2's mixed fast/slow tasks and assert durations separate them.""" - module = import_example("examples.async_profiler_example") - - config = module.DebugToolbarConfig( - enabled=True, - async_profiler_backend="taskfactory", - async_capture_task_stacks=True, - async_max_stack_depth=5, - ) - toolbar2 = module.DebugToolbar(config=config) - panel2 = module.AsyncProfilerPanel(toolbar2) - context2 = await toolbar2.process_request() - - await panel2.process_request(context2) + assert "Backend: taskfactory" in output - mixed_tasks = [ - asyncio.create_task(module.fast_task(1), name="fast_1"), - asyncio.create_task(module.slow_task(2), name="slow_1"), - asyncio.create_task(module.fast_task(3), name="fast_2"), - asyncio.create_task(module.slow_task(4), name="slow_2"), - ] - await asyncio.gather(*mixed_tasks) + total_match = re.search(r"Total tasks created: (\d+)", output) + assert total_match is not None + assert int(total_match.group(1)) == 5 - await panel2.process_response(context2) - stats2 = await panel2.generate_stats(context2) + completed_match = re.search(r"Completed tasks: (\d+)", output) + assert completed_match is not None + assert int(completed_match.group(1)) == 5 - assert stats2["summary"]["total_tasks"] == 4 + assert "Tasks tracked:" in output + assert "(fast_task)" in output - completed = [task for task in stats2["tasks"] if task["event_type"] == "completed" and task.get("duration_ms")] - durations_by_coro: dict[str, list[float]] = {} - for task in completed: - durations_by_coro.setdefault(task["coro_name"], []).append(task["duration_ms"]) - - assert all(duration < 50 for duration in durations_by_coro["fast_task"]) - assert all(duration >= 90 for duration in durations_by_coro["slow_task"]) - - -@pytest.mark.asyncio -async def test_async_profiler_event_loop_lag_monitoring( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 5's event loop lag monitoring and assert real lag was recorded.""" - module = import_example("examples.async_profiler_example") - - config_lag = module.DebugToolbarConfig( - enabled=True, - async_profiler_backend="taskfactory", - async_enable_event_loop_monitoring=True, - async_event_loop_lag_threshold_ms=5.0, - ) - toolbar4 = module.DebugToolbar(config=config_lag) - panel4 = module.AsyncProfilerPanel(toolbar4) - context4 = await toolbar4.process_request() - - await panel4.process_request(context4) - - for _ in range(10): - await asyncio.sleep(0.01) - time.sleep(0.02) # noqa: ASYNC251 - - await panel4.process_response(context4) - stats4 = await panel4.generate_stats(context4) - - assert stats4["summary"]["max_lag_ms"] > 0 - assert stats4["event_loop_lag"] - - -@pytest.mark.asyncio -async def test_async_profiler_nested_task_parent_tracking( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 6's nested task tracking and assert parent links are recorded.""" - module = import_example("examples.async_profiler_example") + 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"]) - config = module.DebugToolbarConfig( - enabled=True, - async_profiler_backend="taskfactory", - async_capture_task_stacks=True, - async_max_stack_depth=5, - ) - toolbar5 = module.DebugToolbar(config=config) - panel5 = module.AsyncProfilerPanel(toolbar5) - context5 = await toolbar5.process_request() + timeline_match = re.search(r"Timeline duration: ([\d.]+)ms", output) + assert timeline_match is not None + assert float(timeline_match.group(1)) > 0 - await panel5.process_request(context5) + 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 - result = await asyncio.create_task(module.nested_tasks(), name="parent_task") - assert result == [0, 2, 4] + 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 - await panel5.process_response(context5) - stats5 = await panel5.generate_stats(context5) + assert "Nested task results: [0, 2, 4]" in output - created_tasks = [task for task in stats5["tasks"] if task["event_type"] == "created"] - parent_ids = [task["parent_task_id"] for task in created_tasks if task["parent_task_id"] is not None] - counts: dict[str, int] = {} - for parent_id in parent_ids: - counts[parent_id] = counts.get(parent_id, 0) + 1 + parent_lines = [line for line in output.splitlines() if "(parent:" in line] + assert len(parent_lines) == 4 + parents = [line.split("(parent: ")[1] for line in parent_lines] + assert len(set(parents[1:])) == 1 + assert parents[0] != parents[1] - assert 3 in counts.values() + 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 index c483685..9fa5027 100644 --- a/tests/examples/test_cache_panel_demo.py +++ b/tests/examples/test_cache_panel_demo.py @@ -5,100 +5,56 @@ 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_records_operations( +def test_demo_context_manager_tracking_reports_real_operations( import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], ) -> None: - """Demo should track a mix of hits, misses, and backends.""" + """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_context_manager_tracking_builds_real_operations( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate the demo's tracker usage directly to assert on real fields.""" - module = import_example("examples.cache_panel_demo") - - tracker = module.CacheTracker() - - with tracker.track_operation("GET", "user:123", "redis") as extra: - extra["hit"] = True - with tracker.track_operation("GET", "user:456", "redis") as extra: - extra["hit"] = False - with tracker.track_operation("SET", "user:456", "redis") as extra: - extra["ttl"] = 3600 - with tracker.track_operation("GET", "session:abc123", "memcached") as extra: - extra["hit"] = True - - assert len(tracker.operations) == 4 - first, second, third, fourth = tracker.operations - assert first.operation == "GET" - assert first.key == "user:123" - assert first.hit is True - assert first.backend == "redis" - - assert second.hit is False - - assert third.operation == "SET" - assert third.extra["ttl"] == 3600 - - assert fourth.backend == "memcached" - assert fourth.hit is True - - -def test_demo_multi_key_operation_tracks_list_key( +def test_demo_multi_key_operation_reports_real_keys( import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], ) -> None: - """Multi-key operations should preserve the list of keys and extras.""" + """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() - tracker = module.CacheTracker() - with tracker.track_operation("MGET", ["key1", "key2", "key3"], "redis") as extra: - extra["hit"] = True - extra["keys_found"] = 2 - - op = tracker.operations[0] - assert op.operation == "MGET" - assert op.key == ["key1", "key2", "key3"] - assert op.hit is True - assert op.extra["keys_found"] == 2 + 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_computes_hit_rate( +def test_demo_statistics_reports_real_hit_rate( import_example: Callable[[str], ModuleType], + capsys: pytest.CaptureFixture[str], ) -> None: - """Statistics demo should produce computable totals, hits, misses, and hit rate.""" + """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() - tracker = module.CacheTracker() - with tracker.track_operation("GET", "key1", "redis") as extra: - extra["hit"] = True - with tracker.track_operation("GET", "key2", "redis") as extra: - extra["hit"] = True - with tracker.track_operation("GET", "key3", "redis") as extra: - extra["hit"] = False - with tracker.track_operation("SET", "key4", "redis"): - pass - with tracker.track_operation("GET", "key5", "memcached") as extra: - extra["hit"] = True - - total = len(tracker.operations) - hits = sum(1 for op in tracker.operations if op.hit is True) - misses = sum(1 for op in tracker.operations if op.hit is False) - - assert total == 5 - assert hits == 3 - assert misses == 1 - - hit_rate = (hits / (hits + misses) * 100) if (hits + misses) > 0 else 0.0 - assert hit_rate == 75.0 + 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_flamegraph_example.py b/tests/examples/test_flamegraph_example.py index dfaf4fb..0dc242b 100644 --- a/tests/examples/test_flamegraph_example.py +++ b/tests/examples/test_flamegraph_example.py @@ -13,13 +13,13 @@ import pytest -def test_main_writes_valid_speedscope_json( +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: - """The script's main() should profile real work and write a valid speedscope file.""" + """Running the real script should profile real work and write a valid speedscope file.""" monkeypatch.chdir(tmp_path) module = import_example("examples.flamegraph_example") @@ -28,6 +28,8 @@ def test_main_writes_valid_speedscope_json( 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() @@ -36,6 +38,9 @@ def test_main_writes_valid_speedscope_json( 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"] @@ -43,30 +48,3 @@ def test_main_writes_valid_speedscope_json( assert profile["endValue"] >= 0 assert len(profile["samples"]) > 0 assert len(profile["samples"]) == len(profile["weights"]) - - -def test_flamegraph_generator_produces_fibonacci_frames( - import_example: Callable[[str], ModuleType], - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - """Replicate the example's profiling setup directly and inspect the frame names.""" - monkeypatch.chdir(tmp_path) - module = import_example("examples.flamegraph_example") - - import cProfile - - from debug_toolbar.core.panels.flamegraph import FlameGraphGenerator - - profiler = cProfile.Profile() - profiler.enable() - module.compute_intensive_task() - module.data_processing() - profiler.disable() - - generator = FlameGraphGenerator(profiler) - speedscope_json = generator.generate() - - frame_names = [frame["name"] for frame in speedscope_json["shared"]["frames"]] - assert any("fibonacci" in name for name in frame_names) - assert any("data_processing" in name for name in frame_names) diff --git a/tests/examples/test_memory_panel_example.py b/tests/examples/test_memory_panel_example.py index a71c2bb..3253a57 100644 --- a/tests/examples/test_memory_panel_example.py +++ b/tests/examples/test_memory_panel_example.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from collections.abc import Callable from typing import TYPE_CHECKING @@ -12,11 +13,11 @@ @pytest.mark.asyncio -async def test_main_prints_expected_output( +async def test_main_reports_real_memory_allocations( import_example: Callable[[str], ModuleType], capsys: pytest.CaptureFixture[str], ) -> None: - """The script's main() should report real memory tracking across backends.""" + """Running the real script should report real, non-zero memory deltas for both backends.""" module = import_example("examples.memory_panel_example") await module.main() @@ -26,67 +27,13 @@ async def test_main_prints_expected_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) -@pytest.mark.asyncio -async def test_memory_panel_tracks_allocation( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 1's basic tracemalloc tracking and assert on real stats.""" - module = import_example("examples.memory_panel_example") - - config = module.DebugToolbarConfig(enabled=True) - toolbar = module.DebugToolbar(config=config) - toolbar.config.memory_backend = "tracemalloc" - - panel = module.MemoryPanel(toolbar) - context = await toolbar.process_request() - - await panel.process_request(context) - - data = module.allocate_memory(5) - nested = module.create_nested_structures() - - await panel.process_response(context) - - stats = await panel.generate_stats(context) - - assert stats["backend"] == "tracemalloc" - assert stats["memory_before"] >= 0 - assert stats["memory_after"] >= 0 - assert stats["memory_delta"] > 0 - assert stats["peak_memory"] > 0 - assert len(stats["top_allocations"]) > 0 - - top_alloc = stats["top_allocations"][0] - assert top_alloc["size"] > 0 - assert top_alloc["count"] > 0 - - del data - del nested - - -@pytest.mark.asyncio -async def test_memory_panel_auto_backend_selection( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 2's auto backend selection and assert a real backend is chosen.""" - module = import_example("examples.memory_panel_example") - - toolbar2 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) - toolbar2.config.memory_backend = "auto" - - panel2 = module.MemoryPanel(toolbar2) - context2 = await toolbar2.process_request() - - await panel2.process_request(context2) - - temp_data = [i**3 for i in range(100000)] - - await panel2.process_response(context2) - stats2 = await panel2.generate_stats(context2) - - assert stats2["backend"] in {"tracemalloc", "memray"} - assert stats2["memory_delta"] >= 0 - assert stats2["peak_memory"] >= 0 + 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) - del temp_data + 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 index efcd451..c03283e 100644 --- a/tests/examples/test_profiling_panel_example.py +++ b/tests/examples/test_profiling_panel_example.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import re from collections.abc import Callable from typing import TYPE_CHECKING @@ -14,13 +15,13 @@ @pytest.mark.asyncio -async def test_main_prints_expected_output( +async def test_main_reports_real_profiling_results( import_example: Callable[[str], ModuleType], capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: - """The script's main() should report cProfile results with real function counts.""" + """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") @@ -28,6 +29,18 @@ async def test_main_prints_expected_output( 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 @@ -36,72 +49,4 @@ async def test_main_prints_expected_output( saved_data = json.loads(saved_path.read_text()) assert saved_data["$schema"] assert len(saved_data["profiles"]) > 0 - - -@pytest.mark.asyncio -async def test_profiling_panel_tracks_function_calls( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 1's setup to assert on real profiling stats.""" - module = import_example("examples.profiling_panel_example") - - config = module.DebugToolbarConfig(enabled=True) - toolbar = module.DebugToolbar(config=config) - toolbar.config.profiler_backend = "cprofile" - toolbar.config.profiler_top_functions = 10 - toolbar.config.profiler_sort_by = "cumulative" - - panel = module.ProfilingPanel(toolbar) - context = await toolbar.process_request() - - await panel.process_request(context) - - module.compute_intensive_task() - module.io_simulation() - - await panel.process_response(context) - - stats = await panel.generate_stats(context) - - assert stats["backend"] == "cprofile" - assert stats["total_time"] > 0 - assert stats["function_calls"] > 0 - assert stats["primitive_calls"] > 0 - assert len(stats["top_functions"]) > 0 - - top_func = stats["top_functions"][0] - assert top_func["calls"] > 0 - assert top_func["cumulative_time"] >= 0 - - -@pytest.mark.asyncio -async def test_profiling_panel_flamegraph_data( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 7's flame graph generation and assert on the speedscope JSON.""" - module = import_example("examples.profiling_panel_example") - - toolbar3 = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) - toolbar3.config.profiler_backend = "cprofile" - toolbar3.config.enable_flamegraph = True - - panel3 = module.ProfilingPanel(toolbar3) - context3 = await toolbar3.process_request() - - await panel3.process_request(context3) - - module.compute_intensive_task() - - await panel3.process_response(context3) - stats3 = await panel3.generate_stats(context3) - - assert stats3["flamegraph_available"] is True - flamegraph = stats3["flamegraph_data"] - assert flamegraph["$schema"] - assert len(flamegraph["shared"]["frames"]) > 0 - assert len(flamegraph["profiles"]) > 0 - - profile_data = flamegraph["profiles"][0] - assert profile_data["type"] - assert profile_data["endValue"] >= 0 - assert len(profile_data["samples"]) > 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 index 838c396..86a37e3 100644 --- a/tests/examples/test_settings_panel_example.py +++ b/tests/examples/test_settings_panel_example.py @@ -12,117 +12,38 @@ @pytest.mark.asyncio -async def test_main_prints_expected_output( +async def test_main_reports_real_settings_and_redactions( import_example: Callable[[str], ModuleType], capsys: pytest.CaptureFixture[str], ) -> None: - """The script's main() should report real toolbar, custom, and redacted settings.""" + """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 "Database Password: **********" in output + assert "API Key: **********" in output assert "secret123" not in output - - -@pytest.mark.asyncio -async def test_settings_panel_basic_stats( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 1's basic panel setup and assert on real stats fields.""" - module = import_example("examples.settings_panel_example") - - config = module.DebugToolbarConfig( - enabled=True, - intercept_redirects=False, - max_request_history=100, - ) - toolbar = module.DebugToolbar(config=config) - - basic_panel = module.SettingsPanel(toolbar) - context = await toolbar.process_request() - stats = await basic_panel.generate_stats(context) - - assert stats["toolbar_config"]["enabled"] is True - assert "debug" in stats["python_settings"] - assert "variables" in stats["environment"] - assert isinstance(stats["environment"]["redacted_count"], int) - - -@pytest.mark.asyncio -async def test_settings_panel_redacts_sensitive_custom_settings( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 2's custom settings and assert the password/api key are redacted.""" - module = import_example("examples.settings_panel_example") - - toolbar = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) - - custom_settings = { - "app_name": "My Litestar App", - "version": "1.0.0", - "database": { - "host": "localhost", - "port": 5432, - "password": "secret123", - }, - "api_key": "abc123xyz", - "debug_mode": True, - } - - custom_panel = module.SettingsPanel(toolbar, custom_settings=custom_settings) - context2 = await toolbar.process_request() - stats2 = await custom_panel.generate_stats(context2) - - assert stats2["custom_settings"]["app_name"] == "My Litestar App" - assert stats2["custom_settings"]["database"]["host"] == "localhost" - assert stats2["custom_settings"]["database"]["password"] == module.SettingsPanel.REDACTED_VALUE - assert stats2["custom_settings"]["api_key"] == module.SettingsPanel.REDACTED_VALUE - - -@pytest.mark.asyncio -async def test_settings_panel_without_environment_section( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 3's show_env=False behavior.""" - module = import_example("examples.settings_panel_example") - - toolbar = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) - - no_env_panel = module.SettingsPanel(toolbar, show_env=False) - context3 = await toolbar.process_request() - stats3 = await no_env_panel.generate_stats(context3) - - assert stats3["environment"] is None - - -@pytest.mark.asyncio -async def test_settings_panel_custom_sensitive_patterns( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate example 4's custom sensitive key patterns.""" - module = import_example("examples.settings_panel_example") - - toolbar = module.DebugToolbar(config=module.DebugToolbarConfig(enabled=True)) - - sensitive_panel = module.SettingsPanel( - toolbar, - custom_settings={ - "app_name": "My App", - "internal_id": "secret-internal-123", - "confidential_data": "top-secret", - "public_url": "https://example.com", - }, - sensitive_keys=["INTERNAL", "CONFIDENTIAL"], - ) - context4 = await toolbar.process_request() - stats4 = await sensitive_panel.generate_stats(context4) - - assert stats4["custom_settings"]["app_name"] == "My App" - assert stats4["custom_settings"]["internal_id"] == module.SettingsPanel.REDACTED_VALUE - assert stats4["custom_settings"]["confidential_data"] == module.SettingsPanel.REDACTED_VALUE - assert stats4["custom_settings"]["public_url"] == "https://example.com" + 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_templates_panel_example.py b/tests/examples/test_templates_panel_example.py index b97af46..7bb7169 100644 --- a/tests/examples/test_templates_panel_example.py +++ b/tests/examples/test_templates_panel_example.py @@ -7,73 +7,26 @@ import pytest -from debug_toolbar.core.context import RequestContext, set_request_context - if TYPE_CHECKING: from types import ModuleType @pytest.mark.asyncio -async def test_main_prints_expected_summary( +async def test_main_reports_real_jinja2_render_statistics( import_example: Callable[[str], ModuleType], capsys: pytest.CaptureFixture[str], ) -> None: - """The script's main() should report three renders across the jinja2 engine.""" + """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 "Template 1 result: Hello World!" in output - - -@pytest.mark.asyncio -async def test_templates_panel_generates_real_stats( - import_example: Callable[[str], ModuleType], -) -> None: - """Replicate the example's panel setup directly to assert on the real stats dict.""" - module = import_example("examples.templates_panel_example") - - config = module.DebugToolbarConfig( - enabled=True, - panels=[ - "debug_toolbar.core.panels.timer.TimerPanel", - "debug_toolbar.core.panels.templates.TemplatesPanel", - ], - ) - toolbar = module.DebugToolbar(config) - templates_panel = toolbar.get_panel("TemplatesPanel") - assert templates_panel is not None - - context = RequestContext() - set_request_context(context) - - await templates_panel.process_request(context) - - from jinja2 import Environment, Template - - template1 = Template("Hello {{ name }}!") - result1 = template1.render(name="World") - assert result1 == "Hello World!" - - env = Environment() - template2 = Template("
{{ title }}
") - result2 = template2.render(title="My List") - assert result2 == "
My List
" - - template3 = env.from_string("{{ greeting }}, {{ name }}!") - result3 = template3.render(greeting="Hi", name="Jinja") - assert result3 == "Hi, Jinja!" - - await templates_panel.process_response(context) - - stats = await templates_panel.generate_stats(context) - - assert stats["total_renders"] == 3 - assert stats["total_time"] >= 0 - assert stats["engines_used"] == ["jinja2"] - assert stats["renders"][0]["template_name"] - assert stats["renders"][1]["template_name"] - assert stats["renders"][2]["template_name"] + assert output.count("Template:") == 3 + assert output.count("Engine: jinja2") == 3 + assert "Context keys: name" in output From e739c043c1927e01f8b577b864adebe9c4491f7d Mon Sep 17 00:00:00 2001 From: Jacob Coffee Date: Sat, 10 Oct 2026 20:21:00 -0500 Subject: [PATCH 5/7] test: rebase asgi_basic tests onto the merged PR #61 fix PR #61 made the asgi_basic example inject the toolbar markup into HTML responses, serve /_debug_toolbar/static/{toolbar.css,toolbar.js}, and move its JSON endpoints to /_debug_toolbar/api/requests(/). Flip the tests that previously asserted the pre-fix absence of these behaviors to assert the real, now-fixed behavior instead. --- tests/examples/test_asgi_basic.py | 75 ++++++++++++++++--------------- 1 file changed, 38 insertions(+), 37 deletions(-) diff --git a/tests/examples/test_asgi_basic.py b/tests/examples/test_asgi_basic.py index c614a04..9b231da 100644 --- a/tests/examples/test_asgi_basic.py +++ b/tests/examples/test_asgi_basic.py @@ -1,12 +1,9 @@ """Tests for the raw-ASGI ``examples/asgi_basic`` example. This example hand-rolls its own debug-toolbar middleware directly against the -``DebugToolbar`` core API, predating the framework integrations' HTML -injection and static-asset routes. On this branch (ahead of PR #61) it does -not inject the toolbar markup into HTML responses and does not serve -``/_debug_toolbar/static/*`` assets at all -- only the two hand-rolled JSON -endpoints (``/_debug_toolbar/requests`` and ``/_debug_toolbar/request/``) -exist. These tests assert the behavior that is actually present today. +``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 @@ -32,30 +29,26 @@ def _client(asgi_basic: ModuleType) -> httpx.AsyncClient: return httpx.AsyncClient(transport=transport, base_url="http://testserver") -async def test_home_page_returns_html(asgi_basic: ModuleType) -> None: +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_home_page_does_not_inject_toolbar_markup(asgi_basic: ModuleType) -> None: - """The hand-rolled middleware never rewrites the response body (PR #61).""" - async with _client(asgi_basic) as client: - response = await client.get("/") - - assert 'id="debug-toolbar"' not in response.text - - -async def test_about_page_returns_html(asgi_basic: ModuleType) -> None: +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: @@ -65,13 +58,13 @@ async def test_api_data_returns_json_without_toolbar_markup(asgi_basic: ModuleTy assert response.status_code == 200 assert response.headers["content-type"].startswith("application/json") data = response.json() - assert data["items"] == [1, 2, 3] + 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("/nope") + response = await client.get("/no-such-page") assert response.status_code == 404 @@ -83,25 +76,43 @@ async def test_server_timing_header_present_after_request(asgi_basic: ModuleType assert "server-timing" in response.headers -async def test_debug_toolbar_requests_lists_recorded_request(asgi_basic: ModuleType) -> None: +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/requests") + 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 requests_list[0]["method"] == "GET" + assert len(requests_list) >= 1 + assert "request_id" in requests_list[0] assert requests_list[0]["path"] == "/" -async def test_debug_toolbar_request_detail_contains_panel_data(asgi_basic: ModuleType) -> None: +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/requests") + 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/request/{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") @@ -119,18 +130,8 @@ async def test_debug_toolbar_request_detail_contains_panel_data(asgi_basic: Modu assert request_stats["path"] == "/" -async def test_debug_toolbar_request_detail_unknown_id_returns_404(asgi_basic: ModuleType) -> None: +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/request/{uuid4()}") + response = await client.get(f"/_debug_toolbar/api/requests/{uuid4()}") assert response.status_code == 404 - - -async def test_static_assets_are_not_served_yet(asgi_basic: ModuleType) -> None: - """No ``/_debug_toolbar/static/*`` route exists in this example (PR #61).""" - async with _client(asgi_basic) as client: - css_response = await client.get("/_debug_toolbar/static/toolbar.css") - js_response = await client.get("/_debug_toolbar/static/toolbar.js") - - assert css_response.status_code == 404 - assert js_response.status_code == 404 From 0d9d47afeb38031cc4afba7122ab4c3498deae72 Mon Sep 17 00:00:00 2001 From: Jacob Coffee Date: Sat, 10 Oct 2026 20:47:56 -0500 Subject: [PATCH 6/7] test: assert real behavior now fixed by PR #64 and #67 PR #64 made the FastAPI dependency injection panel actually populate resolved_dependencies for nested Depends() trees; assert on the real dependency names (get_db, get_current_user, get_token) and counts instead of only checking the panel key exists. PR #67 made the async profiler's taskfactory backend report the real caller-supplied task name (previously always 'Task-N') and made blocking call detection actually fire (previously silent, since asyncio logs slow callbacks via the asyncio logger rather than call_exception_handler). Assert on the real task names and a real detected blocking call instead of working around both bugs. --- tests/examples/test_async_profiler_example.py | 12 +++++++++++- tests/examples/test_fastapi_basic.py | 15 ++++++++++++--- 2 files changed, 23 insertions(+), 4 deletions(-) diff --git a/tests/examples/test_async_profiler_example.py b/tests/examples/test_async_profiler_example.py index 7b0783f..6d15f49 100644 --- a/tests/examples/test_async_profiler_example.py +++ b/tests/examples/test_async_profiler_example.py @@ -34,7 +34,8 @@ async def test_main_reports_real_task_tracking( assert int(completed_match.group(1)) == 5 assert "Tasks tracked:" in output - assert "(fast_task)" 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) @@ -50,6 +51,15 @@ async def test_main_reports_real_task_tracking( 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 diff --git a/tests/examples/test_fastapi_basic.py b/tests/examples/test_fastapi_basic.py index 01f4f2c..1e10ef8 100644 --- a/tests/examples/test_fastapi_basic.py +++ b/tests/examples/test_fastapi_basic.py @@ -124,12 +124,21 @@ def test_request_detail_contains_multiple_panels(fastapi_basic: ModuleType) -> N assert request_stats["path"] == "/" -def test_dependency_injection_panel_is_present(fastapi_basic: ModuleType) -> None: - """Only asserts the panel key exists; its stats are covered by an in-flight fix.""" +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}") - assert "DependencyInjectionPanel" in detail_response.json()["panel_data"] + 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"] From 304886bc59bec06aa07c8ca9aff90f392913a668 Mon Sep 17 00:00:00 2001 From: Jacob Coffee Date: Sat, 10 Oct 2026 20:50:47 -0500 Subject: [PATCH 7/7] test: drop comparison of shortened task ids in async profiler example test --- tests/examples/test_async_profiler_example.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/tests/examples/test_async_profiler_example.py b/tests/examples/test_async_profiler_example.py index 6d15f49..e046e1a 100644 --- a/tests/examples/test_async_profiler_example.py +++ b/tests/examples/test_async_profiler_example.py @@ -69,9 +69,6 @@ async def test_main_reports_real_task_tracking( parent_lines = [line for line in output.splitlines() if "(parent:" in line] assert len(parent_lines) == 4 - parents = [line.split("(parent: ")[1] for line in parent_lines] - assert len(set(parents[1:])) == 1 - assert parents[0] != parents[1] assert "Panel nav subtitle:" in output assert "Panel title: Async" in output