From dde8dc7418e12c233bc21509a8cbd3a3841fb80c Mon Sep 17 00:00:00 2001 From: BarnabasG Date: Wed, 8 Jul 2026 20:33:25 +0100 Subject: [PATCH 1/5] fix: preexisting correctness bugs from codebase audit Framework adapters: - Flask: record calls with query strings (previously dropped); record the actual matched rule for multi-decorated views (return_rule=True); record followed trailing-slash redirects; exclude static routes by endpoint name (covers custom static_url_path and blueprint statics) - FastAPI: discover plain Starlette routes and mounted Starlette sub-apps (docs routes stay excluded); record post-redirect path so slash-redirected calls count against the real route - Django: convert re_path regexes to matchable templates (previously never matched); count only implemented handler methods for CBVs (was inflating every view to 6 methods incl. TRACE) - Detection: isinstance-based for Flask/FastAPI so app subclasses work; Django fallback no longer substring-matches unrelated modules Plugin / xdist: - Register DeferXdistPlugin only when --api-cov-report is active (was registered on every xdist run and crashed on any worker crash) - Tolerate crashed workers with no workeroutput - Merge discovered endpoints from all workers (was first-worker-wins) - CoverageWrapper: delegate context-manager protocol and repr (with-blocks on wrapped clients no longer TypeError) Report / config: - Path parameters match a single segment ([^/]+) so nested-path calls no longer mark parent routes covered; path converters still match greedily - fail_under now fails when discovery finds nothing, and is vacuous (not a spurious failure) when every endpoint is excluded - Implement the documented group-methods-by-endpoint option (was a no-op) - --api-cov-fail-under=0 no longer discarded as falsy - parse_openapi_spec tolerates empty/non-mapping specs instead of crashing fixture setup; cli show-conftest no longer mangles paths containing .py Packaging / CI / docs: - pytest>=7.0 (pytest.Parser/Config are imported at collection time and do not exist before 7.0); add classifiers, keywords, py.typed - publish workflow: tag only after successful publish, run non-mutating make pipeline-ci (auto-fixing lint could publish code matching no commit), pass token via UV_PUBLISH_TOKEN instead of a temp file - CI: re-enable format/lint as non-mutating checks - README: JSON example uses real method-prefixed keys; PUBLISH.md matches the actual Makefile targets and workflow order --- .github/workflows/ci.yml | 16 +- .github/workflows/publish.yml | 38 ++-- Makefile | 13 +- PUBLISH.md | 29 +-- README.md | 6 +- pyproject.toml | 24 +- src/pytest_api_cov/cli.py | 2 +- src/pytest_api_cov/config.py | 9 +- src/pytest_api_cov/frameworks.py | 160 +++++++++++-- src/pytest_api_cov/openapi.py | 13 +- src/pytest_api_cov/plugin.py | 48 +++- src/pytest_api_cov/py.typed | 0 src/pytest_api_cov/report.py | 58 ++++- tests/integration/test_django_integration.py | 61 +++++ .../test_frameworks_integration.py | 213 +++++++++++++++++- tests/unit/test_cli.py | 10 + tests/unit/test_config.py | 15 ++ tests/unit/test_frameworks.py | 34 +++ tests/unit/test_openapi.py | 28 +++ tests/unit/test_plugin.py | 61 ++++- tests/unit/test_report.py | 66 +++++- uv.lock | 2 +- 22 files changed, 809 insertions(+), 97 deletions(-) create mode 100644 src/pytest_api_cov/py.typed diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index de8f951..7aad4e5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -36,14 +36,14 @@ jobs: run: | uv sync --dev - # - name: Run code formatting check - # run: | - # uv run ruff format --check . - - # - name: Run linting - # run: | - # uv run ruff check . - + - name: Run code formatting check + run: | + uv run ruff format --check src tests example + + - name: Run linting + run: | + uv run ruff check --no-fix src + - name: Run type checking run: | uv run mypy diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 1a9df9a..c62106c 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -63,48 +63,48 @@ jobs: echo "Version $VERSION not found on PyPI (HTTP $HTTP_CODE)" fi - - name: Create version tag - if: steps.check_published.outputs.tag_exists == 'false' - run: | - git config user.name "github-actions[bot]" - git config user.email "github-actions[bot]@users.noreply.github.com" - git tag -a "v${{ steps.version.outputs.version }}" -m "Release version ${{ steps.version.outputs.version }}" - git push origin "v${{ steps.version.outputs.version }}" - - name: Set up build environment if: steps.check_published.outputs.pypi_exists == 'false' run: | uv sync --dev - + - name: Run pipeline (tests, linting, etc.) if: steps.check_published.outputs.pypi_exists == 'false' - run: make pipeline - + run: make pipeline-ci + - name: Build package if: steps.check_published.outputs.pypi_exists == 'false' run: make build - + - name: Publish to PyPI if: steps.check_published.outputs.pypi_exists == 'false' env: - PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }} + UV_PUBLISH_TOKEN: ${{ secrets.PYPI_TOKEN }} run: | - if [ -z "$PYPI_TOKEN" ]; then + if [ -z "$UV_PUBLISH_TOKEN" ]; then echo "❌ PYPI_TOKEN secret is not set. Please configure it in repository settings." echo " Settings → Secrets and variables → Actions → New repository secret" exit 1 fi - echo $PYPI_TOKEN > /tmp/pypi_token.txt - uv publish --token $(cat /tmp/pypi_token.txt) - rm /tmp/pypi_token.txt - + uv publish + - name: Verify publication if: steps.check_published.outputs.pypi_exists == 'false' run: | sleep 10 # Wait for PyPI to index the package uv run --with pytest-api-cov --no-project -- python -c \ "import pytest_api_cov; print(f'✅ Published version: {pytest_api_cov.__version__}')" - + + # Tag only after tests, build, and publish have succeeded, so a failed + # release never leaves a version tag pointing at an unreleased commit. + - name: Create version tag + if: steps.check_published.outputs.tag_exists == 'false' + run: | + git config user.name "github-actions[bot]" + git config user.email "github-actions[bot]@users.noreply.github.com" + git tag -a "v${{ steps.version.outputs.version }}" -m "Release version ${{ steps.version.outputs.version }}" + git push origin "v${{ steps.version.outputs.version }}" + - name: Skip publish - already published if: steps.check_published.outputs.pypi_exists == 'true' run: | diff --git a/Makefile b/Makefile index 7b2f8c7..86efb4a 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ # Makefile -.PHONY: ruff mypy test clean clean-all version +.PHONY: ruff mypy test clean clean-all version check pipeline pipeline-ci version: @uv version @@ -21,6 +21,13 @@ vulture: format: ruff mypy vulture +check: + @echo "Running non-mutating checks (format, lint, types, dead code)..." + @uv run ruff format --check src tests example + @uv run ruff check --no-fix src + @uv run mypy + @uv run vulture + test: @echo "Running plugin tests..." @uv run python -u -m pytest tests/ @@ -60,3 +67,7 @@ build: pipeline: format test cover typeguard test-example test-example-parallel +# CI/publish variant: verifies without rewriting any files, so the built +# artifact always matches the commit being released. +pipeline-ci: check test cover typeguard test-example test-example-parallel + diff --git a/PUBLISH.md b/PUBLISH.md index 16230d4..fcd1a38 100644 --- a/PUBLISH.md +++ b/PUBLISH.md @@ -103,17 +103,17 @@ Main publishing workflow triggered on pushes to master/main. **Key features:** - Reads version from `pyproject.toml` - Checks if already published (skips if duplicate version) -- Creates Git tag automatically -- Runs full test pipeline +- Runs the full non-mutating check pipeline (`make pipeline-ci`) - Builds and publishes to PyPI - Verifies publication +- Creates the Git tag last, only after a successful publish ### `.github/workflows/ci.yml` Continuous integration for pull requests. **Key features:** -- Runs on multiple Python versions (3.10, 3.11, 3.12) +- Runs on multiple Python versions (3.10 - 3.14) - Runs on multiple OS (Ubuntu, Windows, macOS) - Checks code formatting - Runs linting @@ -148,30 +148,25 @@ The workflows use `uv version` (available in uv 0.8+) to extract the version dir If you need to publish manually: ```bash -# Run the full pipeline -make pipeline +# Run the full non-mutating pipeline (what CI runs before publishing) +make pipeline-ci # Build the package make build -# Set your PyPI token -export PYPI_TOKEN="your-token-here" -echo $PYPI_TOKEN > .pypi_token - -# Publish -make publish +# Publish (uv reads the token from UV_PUBLISH_TOKEN) +export UV_PUBLISH_TOKEN="your-token-here" +uv publish ``` ## Publishing to Test PyPI -To publish to Test PyPI instead: +To publish to Test PyPI instead, uncomment the `[[tool.uv.index]]` testpypi +section in `pyproject.toml`, then: ```bash -# Set your Test PyPI token -echo $TEST_PYPI_TOKEN > .test_pypi_token - -# Publish to Test PyPI -make publish-test +export UV_PUBLISH_TOKEN="your-test-pypi-token" +uv publish --index testpypi ``` Or create a separate workflow by copying `.github/workflows/publish.yml` and modifying it to use `TEST_PYPI_TOKEN` and the `--index testpypi` flag. diff --git a/README.md b/README.md index 43afd1a..4b74cbb 100644 --- a/README.md +++ b/README.md @@ -459,15 +459,15 @@ When using `--api-cov-report-path`, the plugin generates a detailed JSON report: "excluded_count": 0, "detail": [ { - "endpoint": "/", + "endpoint": "GET /", "callers": ["test_root_endpoint"] }, { - "endpoint": "/users/{user_id}", + "endpoint": "GET /users/{user_id}", "callers": ["test_get_user"] }, { - "endpoint": "/health", + "endpoint": "GET /health", "callers": [] } ] diff --git a/pyproject.toml b/pyproject.toml index 981bcb7..e8502e6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,12 +6,32 @@ readme = "README.md" authors = [{ name = "Barnaby Gill", email = "barnabasgill@gmail.com" }] license = { text = "Apache-2.0" } requires-python = ">=3.10" +keywords = ["pytest", "plugin", "api", "coverage", "testing", "fastapi", "flask", "django"] +classifiers = [ + "Development Status :: 4 - Beta", + "Framework :: Pytest", + "Framework :: Flask", + "Framework :: Django", + "Framework :: FastAPI", + "Intended Audience :: Developers", + "License :: OSI Approved :: Apache Software License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Topic :: Software Development :: Testing", + "Typing :: Typed", +] dependencies = [ "pydantic>=2.0.0", "rich>=10.0.0", "tomli>=1.2.0", - "pytest>=6.0.0", + # pytest.Parser / pytest.Config are referenced at import time and only exist from pytest 7.0. + "pytest>=7.0.0", "PyYAML>=6.0", "backports.strenum>=1.3.1; python_version < '3.11'", ] @@ -23,6 +43,7 @@ flask = ["flask>=2.0.0"] [project.urls] Source = "https://github.com/BarnabasG/api-coverage" +Issues = "https://github.com/BarnabasG/api-coverage/issues" [dependency-groups] dev = [ @@ -49,7 +70,6 @@ fail_under = 70 show_covered_endpoints = true show_uncovered_endpoints = true show_excluded_endpoints = true -exclusion_patterns = ["xyz"] report_path = "reports/pytest_api_cov.json" [tool.pytest.ini_options] diff --git a/src/pytest_api_cov/cli.py b/src/pytest_api_cov/cli.py index 7484e91..053030c 100644 --- a/src/pytest_api_cov/cli.py +++ b/src/pytest_api_cov/cli.py @@ -5,7 +5,7 @@ def generate_conftest_content(framework: str, file_path: str, app_variable: str) -> str: """Generate example conftest.py content for a given framework.""" - module_path = file_path.replace("/", ".").replace("\\", ".").replace(".py", "") + module_path = file_path.replace("/", ".").replace("\\", ".").removesuffix(".py") if framework == "FastAPI": test_client_import = "from fastapi.testclient import TestClient" diff --git a/src/pytest_api_cov/config.py b/src/pytest_api_cov/config.py index 9ad65e7..e54cb27 100644 --- a/src/pytest_api_cov/config.py +++ b/src/pytest_api_cov/config.py @@ -60,7 +60,10 @@ def read_toml_config(rootdir: Path | None = None) -> dict[str, Any]: "api-cov-openapi-spec": "openapi_spec", } -_UNSET: tuple[Any, ...] = (None, [], False) + +def _is_unset(value: Any) -> bool: + """Detect argparse defaults (None, empty list, False) without equating 0/0.0 to False.""" + return value is None or value is False or (isinstance(value, list) and not value) def read_session_config(session_config: Any) -> dict[str, Any]: @@ -68,7 +71,7 @@ def read_session_config(session_config: Any) -> dict[str, Any]: config: dict[str, Any] = {} for opt, key in _CLI_OPTIONS.items(): value = session_config.getoption(f"--{opt}") - if value not in _UNSET: + if not _is_unset(value): config[key] = value if session_config.getoption("--api-cov-hide-uncovered-endpoints"): @@ -81,7 +84,7 @@ def supports_unicode() -> bool: """Check if the terminal supports Unicode output.""" if not sys.stdout.isatty(): return False - return sys.stdout.encoding.lower() in ("utf-8", "utf8") + return (sys.stdout.encoding or "").lower() in ("utf-8", "utf8") def get_pytest_api_cov_report_config(session_config: Any) -> ApiCoverageReportConfig: diff --git a/src/pytest_api_cov/frameworks.py b/src/pytest_api_cov/frameworks.py index d0ef6dc..1f98d66 100644 --- a/src/pytest_api_cov/frameworks.py +++ b/src/pytest_api_cov/frameworks.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re import sys from abc import ABC, abstractmethod from typing import TYPE_CHECKING, Any @@ -43,13 +44,18 @@ def get_tracked_client(self, recorder: ApiCallRecorder | None, test_name: str) - class FlaskAdapter(BaseAdapter): """Adapter for Flask applications.""" + @staticmethod + def _is_static_rule(rule: Any) -> bool: + """Match app and blueprint static rules by endpoint name, covering custom static_url_path.""" + endpoint = str(getattr(rule, "endpoint", "")) + return endpoint == "static" or endpoint.endswith(".static") + def get_endpoints(self) -> list[str]: """Return list of 'METHOD /path' strings.""" - excluded_rules = ("/static/",) endpoints = [ f"{method} {rule.rule}" for rule in self.app.url_map.iter_rules() - if rule.rule not in excluded_rules + if not self._is_static_rule(rule) for method in rule.methods if method not in ("HEAD", "OPTIONS") ] @@ -58,7 +64,10 @@ def get_endpoints(self) -> list[str]: def get_tracked_client(self, recorder: ApiCallRecorder | None, test_name: str) -> Any: """Return a Flask test client with call tracking.""" + from urllib.parse import urlsplit + from flask.testing import FlaskClient + from werkzeug.routing import RequestRedirect if recorder is None: return self.app.test_client() @@ -67,18 +76,32 @@ def get_tracked_client(self, recorder: ApiCallRecorder | None, test_name: str) - if hasattr(self.app.url_map, "bind"): url_adapter = self.app.url_map.bind("") + def _match_rule(path: str, method: str, follow_redirects: bool) -> Any | None: + try: + rule, _ = url_adapter.match(path, method=method, return_rule=True) # type: ignore[union-attr] + except RequestRedirect as redirect: + # A trailing-slash redirect only reaches the view when redirects are followed. + if not follow_redirects: + return None + try: + rule, _ = url_adapter.match( # type: ignore[union-attr] + urlsplit(redirect.new_url).path, method=method, return_rule=True + ) + except Exception: # noqa: BLE001 + return None + except Exception: # noqa: BLE001 + return None + return rule + class TrackingFlaskClient(FlaskClient): def open(self, *args: Any, **kwargs: Any) -> Any: path = kwargs.get("path") or (args[0] if args else None) method = kwargs.get("method", "GET").upper() - if path and url_adapter is not None: - try: - endpoint_name, _ = url_adapter.match(path, method=method) - endpoint_rule_string = next(self.application.url_map.iter_rules(endpoint_name)).rule - recorder.record_call(endpoint_rule_string, test_name, method) # type: ignore[union-attr] - except Exception: # noqa: BLE001 - pass + if isinstance(path, str) and url_adapter is not None: + rule = _match_rule(path.partition("?")[0], method, bool(kwargs.get("follow_redirects"))) + if rule is not None: + recorder.record_call(rule.rule, test_name, method) # type: ignore[union-attr] return super().open(*args, **kwargs) return TrackingFlaskClient(self.app, self.app.response_class) @@ -96,13 +119,22 @@ def get_endpoints(self) -> list[str]: def _collect_routes(self, routes: list[Any], prefix: str, endpoints: list[str]) -> None: """Recursively collect endpoints from routes, including mounted sub-apps.""" from fastapi.routing import APIRoute - from starlette.routing import Mount + from starlette.routing import Mount, Route for route in routes: if isinstance(route, APIRoute): endpoints.extend( f"{method} {prefix}{route.path}" for method in route.methods if method not in ("HEAD", "OPTIONS") ) + elif isinstance(route, Route) and not isinstance(route, Mount): + # Plain Starlette routes (add_route, mounted Starlette apps). The auto-generated + # docs routes (/docs, /openapi.json, ...) carry include_in_schema=False. + if not getattr(route, "include_in_schema", True): + continue + methods = route.methods or {"GET"} + endpoints.extend( + f"{method} {prefix}{route.path}" for method in methods if method not in ("HEAD", "OPTIONS") + ) elif isinstance(route, Mount): mount_prefix = prefix + route.path if hasattr(route, "routes") and route.routes: @@ -125,15 +157,71 @@ def get_tracked_client(self, recorder: ApiCallRecorder | None, test_name: str) - class TrackingFastAPIClient(TestClient): def send(self, *args: Any, **kwargs: Any) -> Any: request = args[0] + try: + response = super().send(*args, **kwargs) + except BaseException: + if recorder is not None: + recorder.record_call(request.url.path, test_name, request.method.upper()) + raise if recorder is not None: - method = request.method.upper() - path = request.url.path - recorder.record_call(path, test_name, method) - return super().send(*args, **kwargs) + # httpx follows redirects inside send(); response.request points at the + # final request, so followed slash-redirects record the real route path. + final_request = getattr(response, "request", request) + recorder.record_call(final_request.url.path, test_name, final_request.method.upper()) + return response return TrackingFastAPIClient(self.app) +def _django_route_to_template(route: str) -> str: + r"""Convert a Django route string to a matchable template. + + ``path()`` routes pass through unchanged; ``re_path()`` regex groups + (``(?P[0-9]{4})``) become ```` placeholders and escaped + literals (``\.``) are unescaped so recorded request paths can match. + """ + out: list[str] = [] + i = 0 + param_count = 0 + n = len(route) + while i < n: + char = route[i] + if char == "\\" and i + 1 < n: + out.append(route[i + 1]) + i += 2 + elif char == "(": + depth = 0 + j = i + in_class = False + while j < n: + inner = route[j] + if inner == "\\": + j += 2 + continue + if in_class: + in_class = inner != "]" + elif inner == "[": + in_class = True + elif inner == "(": + depth += 1 + elif inner == ")": + depth -= 1 + if depth == 0: + break + j += 1 + named = re.match(r"\(\?P<(\w+)>", route[i : j + 1]) + if named: + out.append(f"<{named.group(1)}>") + else: + param_count += 1 + out.append(f"") + i = j + 1 + else: + out.append(char) + i += 1 + return "".join(out) + + class DjangoAdapter(BaseAdapter): """Adapter for Django applications.""" @@ -147,19 +235,22 @@ def get_endpoints(self) -> list[str]: def _extract_patterns(patterns: list[Any], prefix: str = "") -> None: for pattern in patterns: if isinstance(pattern, URLPattern): - route = str(pattern.pattern).strip("^$") + route = _django_route_to_template(str(pattern.pattern).strip("^$")) full_path = f"/{prefix}{route}".replace("//", "/") view = pattern.callback methods = {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"} - if hasattr(view, "view_class") and hasattr(view.view_class, "http_method_names"): - methods = {m.upper() for m in view.view_class.http_method_names} + view_class = getattr(view, "view_class", None) + if view_class is not None and hasattr(view_class, "http_method_names"): + # Only count methods the class actually implements (mirrors + # View._allowed_methods), not the full http_method_names list. + methods = {m.upper() for m in view_class.http_method_names if hasattr(view_class, m)} endpoints.extend(f"{method} {full_path}" for method in methods if method not in ("HEAD", "OPTIONS")) elif isinstance(pattern, URLResolver): - route = str(pattern.pattern).strip("^$") + route = _django_route_to_template(str(pattern.pattern).strip("^$")) _extract_patterns(pattern.url_patterns, f"{prefix}{route}") _extract_patterns(get_resolver().url_patterns) @@ -195,8 +286,37 @@ def _unwrap_wsgi_app(app: Any) -> Any: return None +def _detect_by_isinstance(app: Any) -> SupportedFramework | None: + """Detect Flask/FastAPI apps (including subclasses) via isinstance checks.""" + try: + from flask import Flask + + if isinstance(app, Flask): + return SupportedFramework.FLASK + except ImportError: + pass + + try: + from fastapi import FastAPI + + if isinstance(app, FastAPI): + return SupportedFramework.FASTAPI + except ImportError: + pass + + return None + + def _detect_framework(app: Any) -> SupportedFramework | None: - """Lightweight check to detect the framework.""" + """Detect the framework, supporting app subclasses via isinstance checks.""" + if app is None: + return None + + framework = _detect_by_isinstance(app) + if framework is not None: + return framework + + # Name-based fallback for Django handlers and duck-typed apps. app_type = type(app).__name__ module_name = getattr(type(app), "__module__", "").split(".")[0] @@ -205,7 +325,7 @@ def _detect_framework(app: Any) -> SupportedFramework | None: return SupportedFramework.FLASK case ("fastapi", "FastAPI"): return SupportedFramework.FASTAPI - case (module, _) if module == "django" or "django" in module: + case ("django", _): return SupportedFramework.DJANGO case _: return None diff --git a/src/pytest_api_cov/openapi.py b/src/pytest_api_cov/openapi.py index e8f20b6..6b766e8 100644 --- a/src/pytest_api_cov/openapi.py +++ b/src/pytest_api_cov/openapi.py @@ -31,8 +31,19 @@ def parse_openapi_spec(path: str) -> list[str]: logger.exception("Failed to parse OpenAPI spec", exc_info=True) return [] + if not isinstance(spec, dict): + logger.error(f"OpenAPI spec is empty or not a mapping: {spec_path}") + return [] + endpoints: list[str] = [] - for path_key, path_item in spec.get("paths", {}).items(): + paths = spec.get("paths", {}) + if not isinstance(paths, dict): + logger.error(f"OpenAPI spec 'paths' section is not a mapping: {spec_path}") + return [] + + for path_key, path_item in paths.items(): + if not isinstance(path_item, dict): + continue endpoints.extend(f"{method.upper()} {path_key}" for method in path_item if method.upper() in HTTP_METHODS) return sorted(endpoints) diff --git a/src/pytest_api_cov/plugin.py b/src/pytest_api_cov/plugin.py index 3099b73..16679c2 100644 --- a/src/pytest_api_cov/plugin.py +++ b/src/pytest_api_cov/plugin.py @@ -98,7 +98,7 @@ def pytest_configure(config: pytest.Config) -> None: logger.setLevel(log_level) logger.info("Initializing API coverage plugin...") - if config.pluginmanager.hasplugin("xdist"): + if config.getoption("--api-cov-report") and config.pluginmanager.hasplugin("xdist"): config.pluginmanager.register(DeferXdistPlugin(), "defer_xdist_plugin") @@ -278,6 +278,30 @@ def tracked(*args: Any, **kwargs: Any) -> Any: object.__setattr__(self, name, tracked) return tracked + # Dunder lookups bypass __getattr__, so the context-manager protocol (used by + # httpx clients to trigger app lifespan) must be delegated explicitly. + def __enter__(self) -> "CoverageWrapper": + """Enter the wrapped client's context, returning the wrapper.""" + self._wrapped.__enter__() + return self + + def __exit__(self, *exc_info: object) -> Any: + """Exit the wrapped client's context.""" + return self._wrapped.__exit__(*exc_info) + + async def __aenter__(self) -> "CoverageWrapper": + """Enter the wrapped client's async context, returning the wrapper.""" + await self._wrapped.__aenter__() + return self + + async def __aexit__(self, *exc_info: object) -> Any: + """Exit the wrapped client's async context.""" + return await self._wrapped.__aexit__(*exc_info) + + def __repr__(self) -> str: + """Show the wrapped client.""" + return f"CoverageWrapper({self._wrapped!r})" + def wrap_client_with_coverage(client: Any, recorder: Any, test_name: str) -> Any: """Wrap an existing test client with coverage tracking.""" @@ -395,8 +419,14 @@ class DeferXdistPlugin: def pytest_testnodedown(self, node: Any) -> None: """Collect API call data from each worker as they finish.""" logger.debug("> Worker node down.") - worker_data = node.workeroutput.get("api_call_recorder", {}) - discovered_endpoints = node.workeroutput.get("discovered_endpoints", []) + workeroutput = getattr(node, "workeroutput", None) + if workeroutput is None: + # A crashed worker never populates workeroutput. + logger.debug("> Worker went down without output; skipping merge.") + return + + worker_data = workeroutput.get("api_call_recorder", {}) + discovered_endpoints = workeroutput.get("discovered_endpoints", []) if worker_data: current = getattr(node.config, "worker_api_call_recorder", {}) @@ -406,6 +436,12 @@ def pytest_testnodedown(self, node: Any) -> None: node.config.worker_api_call_recorder = current - if discovered_endpoints and not getattr(node.config, "worker_discovered_endpoints", []): - node.config.worker_discovered_endpoints = discovered_endpoints - logger.debug(f"> Set discovered endpoints from worker: {discovered_endpoints}") + if discovered_endpoints: + current_endpoints = getattr(node.config, "worker_discovered_endpoints", []) + seen = set(current_endpoints) + for endpoint in discovered_endpoints: + if endpoint not in seen: + seen.add(endpoint) + current_endpoints.append(endpoint) + node.config.worker_discovered_endpoints = current_endpoints + logger.debug(f"> Merged discovered endpoints from worker: {discovered_endpoints}") diff --git a/src/pytest_api_cov/py.typed b/src/pytest_api_cov/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/pytest_api_cov/report.py b/src/pytest_api_cov/report.py index 8b779c1..f7a9d1c 100644 --- a/src/pytest_api_cov/report.py +++ b/src/pytest_api_cov/report.py @@ -17,10 +17,24 @@ @lru_cache(maxsize=512) def endpoint_to_regex(endpoint: str) -> Pattern[str]: - """Create a regex pattern from an endpoint by replacing dynamic segments.""" - placeholder = "___PLACEHOLDER___" - temp_endpoint = re.escape(re.sub(r"<[^>]+>|\{[^}]+\}", placeholder, endpoint)) - return re.compile("^" + temp_endpoint.replace(placeholder, "(.+)") + "$") + """Create a regex pattern from an endpoint by replacing dynamic segments. + + Plain parameters match a single path segment so a call to a nested path + cannot mark a parent route covered; path-converter parameters (Flask + ````, Starlette ``{x:path}``) still match across segments. + """ + segment_placeholder = "___SEGMENT___" + path_placeholder = "___PATH___" + + def _placeholder(match: re.Match[str]) -> str: + inner = match.group(0)[1:-1] + if inner.startswith("path:") or inner.endswith(":path"): + return path_placeholder + return segment_placeholder + + temp_endpoint = re.escape(re.sub(r"<[^>]+>|\{[^}]+\}", _placeholder, endpoint)) + pattern = temp_endpoint.replace(segment_placeholder, "([^/]+)").replace(path_placeholder, "(.+)") + return re.compile("^" + pattern + "$") def contains_escape_characters(endpoint: str) -> bool: @@ -124,6 +138,27 @@ def categorise_endpoints( return covered, uncovered, excluded +def group_endpoints_by_path( + endpoints: list[str], + called_data: dict[str, set[str]], +) -> tuple[list[str], dict[str, set[str]]]: + """Collapse 'METHOD /path' keys to '/path', merging caller sets across methods.""" + grouped_endpoints: list[str] = [] + seen: set[str] = set() + for endpoint in endpoints: + path = endpoint.split(" ", 1)[1] if " " in endpoint else endpoint + if path not in seen: + seen.add(path) + grouped_endpoints.append(path) + + grouped_calls: dict[str, set[str]] = {} + for key, callers in called_data.items(): + path = key.split(" ", 1)[1] if " " in key else key + grouped_calls.setdefault(path, set()).update(callers) + + return grouped_endpoints, grouped_calls + + def print_endpoints( console: Console, label: str, @@ -182,12 +217,21 @@ def generate_pytest_api_cov_report( console = Console() if not discovered_endpoints: + if api_cov_config.fail_under is not None: + console.print( + f"\n[bold red]FAIL: No endpoints discovered but --api-cov-fail-under={api_cov_config.fail_under} " + "is set. Check your app/client fixtures or OpenAPI spec.[/bold red]" + ) + return 1 console.print("\n[bold red]No endpoints discovered. Please check your test setup.[/bold red]") return 0 separator = "=" * 20 console.print(f"\n\n[bold blue]{separator} API Coverage Report {separator}[/bold blue]") + if api_cov_config.group_methods_by_endpoint: + discovered_endpoints, called_data = group_endpoints_by_path(discovered_endpoints, called_data) + covered, uncovered, excluded = categorise_endpoints( discovered_endpoints, called_data, @@ -226,6 +270,12 @@ def generate_pytest_api_cov_report( if api_cov_config.fail_under is None: console.print(f"\n[bold green]Total API Coverage: {coverage}%[/bold green]") + elif not covered and not uncovered: + # Every endpoint was excluded: nothing is measurable, so the gate is vacuous. + console.print( + f"\n[bold yellow]All {len(excluded)} discovered endpoints are excluded; " + "coverage requirement not applied.[/bold yellow]" + ) elif coverage < api_cov_config.fail_under: console.print( f"\n[bold red]FAIL: Required coverage of {api_cov_config.fail_under}% not met. " diff --git a/tests/integration/test_django_integration.py b/tests/integration/test_django_integration.py index 4bb5d37..48bc7b5 100644 --- a/tests/integration/test_django_integration.py +++ b/tests/integration/test_django_integration.py @@ -60,3 +60,64 @@ def test_root(coverage_client): assert "GET /api/root/" in result.stdout.str() assert "Total API Coverage: 20.0%" in result.stdout.str() assert result.ret == 0 + + +def test_django_cbv_methods_and_re_path_templates(pytester): + """CBVs only count implemented methods; re_path routes become matchable templates.""" + pytester.makepyfile( + urls=""" + from django.http import JsonResponse + from django.urls import path, re_path + from django.views import View + + class GetOnlyView(View): + def get(self, request): + return JsonResponse({"ok": True}) + + def year_view(request, year): + return JsonResponse({"year": year}) + + urlpatterns = [ + path("only-get/", GetOnlyView.as_view()), + re_path(r"^articles/(?P[0-9]{4})/$", year_view), + ] + """ + ) + + pytester.makeconftest(""" + import pytest + from django.conf import settings + from django.core.handlers.wsgi import WSGIHandler + + if not settings.configured: + settings.configure( + DEBUG=True, + SECRET_KEY="secret", + ROOT_URLCONF="urls", + ALLOWED_HOSTS=["*"], + INSTALLED_APPS=[], + ) + import django + django.setup() + + @pytest.fixture + def app(): + return WSGIHandler() + """) + + pytester.makepyfile(""" + def test_articles(coverage_client): + response = coverage_client.get("/articles/2024/") + assert response.status_code == 200 + """) + + result = pytester.runpytest("--api-cov-report", "--api-cov-show-covered-endpoints", "-vv") + output = result.stdout.str() + + assert "GET /articles//" in output + assert "GET /only-get/" in output + # CBVs only count implemented handlers; FBVs keep the 5-method default. + assert "POST /only-get/" not in output + assert "POST /articles//" in output + assert "Total API Coverage: 16.67%" in output + assert result.ret == 0 diff --git a/tests/integration/test_frameworks_integration.py b/tests/integration/test_frameworks_integration.py index 5deb869..08ebe98 100644 --- a/tests/integration/test_frameworks_integration.py +++ b/tests/integration/test_frameworks_integration.py @@ -58,7 +58,32 @@ def items(): pytest.skip("Flask not available for integration testing") def test_flask_excluded_endpoints(self): - """Test that static endpoints are excluded.""" + """Framework static routes are excluded by endpoint name; user routes are kept.""" + try: + from flask import Blueprint, Flask + + app = Flask(__name__, static_url_path="/assets") + + @app.route("/api/users") + def api_users(): + return "API Users" + + blueprint = Blueprint("admin", __name__, static_folder="static", url_prefix="/admin") + app.register_blueprint(blueprint) + + adapter = FlaskAdapter(app) + endpoints = adapter.get_endpoints() + paths = [ep.split(" ", 1)[1] if " " in ep else ep for ep in endpoints] + + assert "/assets/" not in paths + assert "/admin/static/" not in paths + assert "GET /api/users" in endpoints + + except ImportError: + pytest.skip("Flask not available for integration testing") + + def test_flask_user_route_shadowing_static_path_is_kept(self): + """A user view routed under /static/ is a real endpoint and must be counted.""" try: from flask import Flask @@ -68,15 +93,10 @@ def test_flask_excluded_endpoints(self): def static_file(filename): return f"Static {filename}" - @app.route("/api/users") - def api_users(): - return "API Users" - adapter = FlaskAdapter(app) endpoints = adapter.get_endpoints() - assert "/static/" not in [ep.split(" ", 1)[1] if " " in ep else ep for ep in endpoints] - assert "GET /api/users" in endpoints + assert "GET /static/" in endpoints except ImportError: pytest.skip("Flask not available for integration testing") @@ -160,3 +180,182 @@ def api_users(): except ImportError: pytest.skip("FastAPI not available for integration testing") + + +class TestFlaskTrackingRegressions: + """Regression tests for Flask tracked-client recording.""" + + @staticmethod + def _make_adapter_and_recorder(): + from flask import Flask + + app = Flask(__name__) + + @app.route("/items") + def items(): + return "Items" + + @app.route("/slash/") + def slash(): + return "Slash" + + @app.route("/a") + @app.route("/b") + def multi(): + return "Multi" + + return FlaskAdapter(app), ApiCallRecorder() + + def test_query_string_calls_are_recorded(self): + """A request with a query string still records the matched rule.""" + try: + adapter, recorder = self._make_adapter_and_recorder() + except ImportError: + pytest.skip("Flask not available for integration testing") + + client = adapter.get_tracked_client(recorder, "test_query") + response = client.get("/items?page=2&size=10") + + assert response.status_code == 200 + assert "GET /items" in recorder + + def test_multi_decorated_view_records_the_called_rule(self): + """With two route decorators on one view, the rule actually called is recorded.""" + try: + adapter, recorder = self._make_adapter_and_recorder() + except ImportError: + pytest.skip("Flask not available for integration testing") + + client = adapter.get_tracked_client(recorder, "test_multi") + client.get("/a") + + assert "GET /a" in recorder + assert "GET /b" not in recorder + + client.get("/b") + assert "GET /b" in recorder + + def test_followed_trailing_slash_redirect_is_recorded(self): + """/slash -> /slash/ with follow_redirects=True reaches the view and is recorded.""" + try: + adapter, recorder = self._make_adapter_and_recorder() + except ImportError: + pytest.skip("Flask not available for integration testing") + + client = adapter.get_tracked_client(recorder, "test_redirect") + response = client.get("/slash", follow_redirects=True) + + assert response.status_code == 200 + assert "GET /slash/" in recorder + + def test_unfollowed_redirect_is_not_recorded(self): + """Without follow_redirects the view never runs, so nothing is recorded.""" + try: + adapter, recorder = self._make_adapter_and_recorder() + except ImportError: + pytest.skip("Flask not available for integration testing") + + client = adapter.get_tracked_client(recorder, "test_no_follow") + response = client.get("/slash") + + assert response.status_code in (301, 308) + assert len(recorder) == 0 + + +class TestFrameworkSubclassDetection: + """Apps subclassing Flask/FastAPI must be detected.""" + + def test_flask_subclass_is_detected(self): + """A Flask subclass resolves to the Flask adapter.""" + try: + from flask import Flask + except ImportError: + pytest.skip("Flask not available for integration testing") + + from pytest_api_cov.frameworks import get_framework_adapter + + class CustomFlask(Flask): + pass + + assert isinstance(get_framework_adapter(CustomFlask(__name__)), FlaskAdapter) + + def test_fastapi_subclass_is_detected(self): + """A FastAPI subclass resolves to the FastAPI adapter.""" + try: + from fastapi import FastAPI + except ImportError: + pytest.skip("FastAPI not available for integration testing") + + from pytest_api_cov.frameworks import get_framework_adapter + + class CustomFastAPI(FastAPI): + pass + + assert isinstance(get_framework_adapter(CustomFastAPI()), FastAPIAdapter) + + +class TestFastAPIRouteDiscoveryRegressions: + """Regression tests for FastAPI/Starlette route discovery and recording.""" + + def test_plain_starlette_routes_are_discovered_but_docs_are_not(self): + """add_route() endpoints appear; auto-generated docs routes do not.""" + try: + from fastapi import FastAPI + from starlette.responses import PlainTextResponse + except ImportError: + pytest.skip("FastAPI not available for integration testing") + + app = FastAPI() + + async def plain(request): + return PlainTextResponse("ok") + + app.add_route("/plain", plain, methods=["GET"]) + + endpoints = FastAPIAdapter(app).get_endpoints() + + assert "GET /plain" in endpoints + assert not any("/docs" in ep or "/openapi.json" in ep or "/redoc" in ep for ep in endpoints) + + def test_mounted_starlette_app_routes_are_discovered(self): + """Routes of a mounted plain Starlette app appear with the mount prefix.""" + try: + from fastapi import FastAPI + from starlette.applications import Starlette + from starlette.responses import PlainTextResponse + from starlette.routing import Route + except ImportError: + pytest.skip("FastAPI not available for integration testing") + + async def sub(request): + return PlainTextResponse("sub") + + subapp = Starlette(routes=[Route("/sub", sub, methods=["GET"])]) + app = FastAPI() + app.mount("/mnt", subapp) + + endpoints = FastAPIAdapter(app).get_endpoints() + + assert "GET /mnt/sub" in endpoints + + def test_followed_slash_redirect_records_final_path(self): + """/items redirected to /items/ records the real route path, not the original.""" + try: + from fastapi import FastAPI + except ImportError: + pytest.skip("FastAPI not available for integration testing") + + app = FastAPI() + + @app.get("/items/") + def items(): + return {"ok": True} + + recorder = ApiCallRecorder() + client = FastAPIAdapter(app).get_tracked_client(recorder, "test_redirect") + + response = client.get("/items", follow_redirects=True) + + assert response.status_code == 200 + assert "GET /items/" in recorder + assert "GET /items" not in recorder diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 34abdd0..5861675 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -108,3 +108,13 @@ def test_main_unknown_command(self): monkeypatch.undo() assert exc_info.value.code == 2 + + +class TestModulePathHandling: + """Regression tests for module path mangling.""" + + def test_module_path_containing_py_substring(self): + """Only a trailing .py is stripped; '.py' inside the module path is preserved.""" + content = generate_conftest_content("FastAPI", "src.my.pyapp.main.py", "app") + + assert "from src.my.pyapp.main import app" in content diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 5404aae..01bf6e5 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -242,3 +242,18 @@ def test_read_toml_config_with_openapi_spec(self, tmp_path): assert config["openapi_spec"] == "openapi.yaml" finally: os.chdir(original_cwd) + + +class TestFalsyCliValues: + """Explicit falsy CLI values must not be treated as unset.""" + + def test_fail_under_zero_is_kept(self): + """--api-cov-fail-under=0 is a deliberate setting, not an argparse default.""" + mock_session_config = Mock() + mock_session_config.getoption.side_effect = lambda name: { + "--api-cov-fail-under": 0.0, + }.get(name) + + config = read_session_config(mock_session_config) + + assert config["fail_under"] == 0.0 diff --git a/tests/unit/test_frameworks.py b/tests/unit/test_frameworks.py index 8b1a07d..e519352 100644 --- a/tests/unit/test_frameworks.py +++ b/tests/unit/test_frameworks.py @@ -219,3 +219,37 @@ def test_get_framework_adapter_with_missing_module(self): with pytest.raises(TypeError, match="Unsupported application type"): get_framework_adapter(mock_app) + + +class TestDjangoRouteToTemplate: + """Tests for converting Django re_path regexes to matchable templates.""" + + def test_named_group(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template("articles/(?P[0-9]{4})/") == "articles//" + + def test_unnamed_groups(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template("files/([0-9]+)/([a-z]+)/") == "files///" + + def test_escaped_literals_are_unescaped(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template(r"feed\.json") == "feed.json" + + def test_character_class_containing_parens(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template(r"tags/(?P[()a-z]+)/") == "tags//" + + def test_nested_groups_consume_whole_group(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template(r"v/(?Pv(1|2))/") == "v//" + + def test_path_route_passes_through(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template("articles//") == "articles//" diff --git a/tests/unit/test_openapi.py b/tests/unit/test_openapi.py index ced9a2f..6a55b2f 100644 --- a/tests/unit/test_openapi.py +++ b/tests/unit/test_openapi.py @@ -85,3 +85,31 @@ def test_unsupported_file_extension(self, tmp_path): endpoints = parse_openapi_spec(str(spec_file)) assert endpoints == [] + + def test_empty_yaml_spec(self, tmp_path): + """An empty YAML file parses to None and must not crash fixture setup.""" + spec_file = tmp_path / "empty.yaml" + spec_file.write_text("") + + assert parse_openapi_spec(str(spec_file)) == [] + + def test_non_mapping_spec(self, tmp_path): + """A top-level list is not a valid spec and must not crash.""" + spec_file = tmp_path / "list.json" + spec_file.write_text("[1, 2, 3]") + + assert parse_openapi_spec(str(spec_file)) == [] + + def test_non_mapping_paths_section(self, tmp_path): + """A non-mapping paths section returns no endpoints.""" + spec_file = tmp_path / "badpaths.json" + spec_file.write_text(json.dumps({"openapi": "3.0.0", "paths": ["not", "a", "mapping"]})) + + assert parse_openapi_spec(str(spec_file)) == [] + + def test_non_mapping_path_item_is_skipped(self, tmp_path): + """Malformed path items are skipped, valid ones kept.""" + spec_file = tmp_path / "baditem.json" + spec_file.write_text(json.dumps({"paths": {"/users": {"get": {}}, "/bad": "nope"}})) + + assert parse_openapi_spec(str(spec_file)) == ["GET /users"] diff --git a/tests/unit/test_plugin.py b/tests/unit/test_plugin.py index 330ba4b..f2ace77 100644 --- a/tests/unit/test_plugin.py +++ b/tests/unit/test_plugin.py @@ -265,14 +265,14 @@ def test_pytest_configure_without_xdist(self): mock_config.pluginmanager.register.assert_not_called() def test_pytest_configure_without_api_cov_report(self): - """Logging is skipped when api-cov-report is off.""" + """No xdist registration (or logging setup) when api-cov-report is off.""" mock_config = Mock() mock_config.getoption.return_value = False mock_config.pluginmanager.hasplugin.return_value = True pytest_configure(mock_config) - mock_config.pluginmanager.register.assert_called_once() + mock_config.pluginmanager.register.assert_not_called() @pytest.mark.parametrize( ("verbose_level", "expected_log_level"), @@ -358,6 +358,33 @@ def test_pytest_testnodedown_with_existing_worker_data(self): assert "existing_test" in worker_data["/existing"] assert "new_test" in worker_data["/new"] + def test_pytest_testnodedown_crashed_worker_without_workeroutput(self): + """A worker that crashed before producing output must not raise.""" + mock_node = Mock(spec=["config"]) + + plugin = DeferXdistPlugin() + plugin.pytest_testnodedown(mock_node) + + def test_pytest_testnodedown_merges_endpoints_from_all_workers(self): + """Discovered endpoints accumulate across workers instead of first-worker-wins.""" + mock_config = Mock() + mock_config.worker_discovered_endpoints = [] + mock_config.worker_api_call_recorder = {} + + first = Mock() + first.config = mock_config + first.workeroutput = {"discovered_endpoints": ["GET /a", "GET /b"]} + + second = Mock() + second.config = mock_config + second.workeroutput = {"discovered_endpoints": ["GET /b", "GET /c"]} + + plugin = DeferXdistPlugin() + plugin.pytest_testnodedown(first) + plugin.pytest_testnodedown(second) + + assert mock_config.worker_discovered_endpoints == ["GET /a", "GET /b", "GET /c"] + def test_extract_app_from_client_variants(): """Extract app from different client shapes.""" @@ -539,3 +566,33 @@ def getfixturevalue(self, name): assert "GET /users" in coverage_data.discovered_endpoints.endpoints assert "POST /users" in coverage_data.discovered_endpoints.endpoints assert coverage_data.discovered_endpoints.discovery_source == "openapi_spec" + + +def test_coverage_wrapper_supports_context_manager_protocol(): + """`with wrapper:` delegates to the wrapped client and yields the wrapper.""" + from unittest.mock import MagicMock + + from pytest_api_cov.plugin import CoverageWrapper + + client = MagicMock() + wrapper = CoverageWrapper(client, Mock(), "test_ctx") + + with wrapper as entered: + assert entered is wrapper + + client.__enter__.assert_called_once() + client.__exit__.assert_called_once() + + +def test_coverage_wrapper_records_request_method_calls(): + """The .request(method, url) call pattern is recorded with the query stripped.""" + from pytest_api_cov.plugin import CoverageWrapper + + client = Mock() + recorder = Mock() + wrapper = CoverageWrapper(client, recorder, "test_req") + + wrapper.request("GET", "/things?q=1") + + client.request.assert_called_once_with("GET", "/things?q=1") + recorder.record_call.assert_called_once_with("/things", "test_req", "GET") diff --git a/tests/unit/test_report.py b/tests/unit/test_report.py index 765dada..c10e6cf 100644 --- a/tests/unit/test_report.py +++ b/tests/unit/test_report.py @@ -21,9 +21,17 @@ class TestEndpointCategorization: def test_endpoint_to_regex_conversion(self): """Regex creation for Flask and FastAPI style placeholders.""" - assert endpoint_to_regex("/users/").pattern == "^/users/(.+)$" - assert endpoint_to_regex("/items/{item_id}/data").pattern == "^/items/(.+)/data$" + assert endpoint_to_regex("/users/").pattern == "^/users/([^/]+)$" + assert endpoint_to_regex("/items/{item_id}/data").pattern == "^/items/([^/]+)/data$" assert endpoint_to_regex("/static/path").pattern == "^/static/path$" + assert endpoint_to_regex("/files/").pattern == "^/files/(.+)$" + assert endpoint_to_regex("/files/{file_path:path}").pattern == "^/files/(.+)$" + + def test_parameter_matching_is_segment_scoped(self): + """A call to a nested path must not mark a parent parameterised route covered.""" + assert endpoint_to_regex("GET /users/{user_id}").match("GET /users/123") + assert not endpoint_to_regex("GET /users/{user_id}").match("GET /users/123/avatar") + assert endpoint_to_regex("GET /files/").match("GET /files/a/b/c.txt") def test_categorise_endpoints(self): """Standard categorisation with exclusions.""" @@ -346,3 +354,57 @@ def test_write_report_file(self, mock_json_dump, mock_open): mock_open.assert_called_once() assert mock_open.call_args[0] == ("w",) mock_json_dump.assert_called_once_with(report_data, mock_open.return_value.__enter__.return_value, indent=2) + + +class TestGroupingAndDegenerateCases: + """Tests for method grouping and degenerate fail_under handling.""" + + def test_group_endpoints_by_path(self): + """Method-prefixed keys collapse to paths with caller sets merged.""" + from pytest_api_cov.report import group_endpoints_by_path + + endpoints = ["GET /users", "POST /users", "GET /health"] + called = {"GET /users": {"test_a"}, "POST /users": {"test_b"}} + + grouped_endpoints, grouped_calls = group_endpoints_by_path(endpoints, called) + + assert grouped_endpoints == ["/users", "/health"] + assert grouped_calls == {"/users": {"test_a", "test_b"}} + + @patch("pytest_api_cov.report.Console") + def test_generate_report_grouped_methods(self, mock_console_cls): + """With grouping, an endpoint counts covered if any method was tested.""" + mock_console = mock_console_cls.return_value + config = ApiCoverageReportConfig.model_validate({"group_methods_by_endpoint": True}) + discovered = ["GET /users/{id}", "PUT /users/{id}", "DELETE /users/{id}", "GET /users", "POST /users"] + called = {"GET /users/123": {"test_get"}, "POST /users": {"test_post"}} + + status = generate_pytest_api_cov_report(config, called, discovered) + + assert status == 0 + total_print = next(c for c in mock_console.print.call_args_list if "Total API Coverage" in c.args[0]) + assert "100.0%" in total_print.args[0] + + @patch("pytest_api_cov.report.Console") + def test_generate_report_fail_under_with_no_endpoints(self, mock_console_cls): + """A configured fail_under gate must fail when discovery found nothing.""" + mock_console = mock_console_cls.return_value + config = ApiCoverageReportConfig.model_validate({"fail_under": 80.0}) + + status = generate_pytest_api_cov_report(config, {}, []) + + assert status == 1 + fail_print = next(c for c in mock_console.print.call_args_list if "FAIL" in c.args[0]) + assert "No endpoints discovered" in fail_print.args[0] + + @patch("pytest_api_cov.report.Console") + def test_generate_report_fail_under_with_all_endpoints_excluded(self, mock_console_cls): + """Excluding every endpoint makes the gate vacuous rather than a spurious failure.""" + mock_console = mock_console_cls.return_value + config = ApiCoverageReportConfig.model_validate({"fail_under": 80.0, "exclusion_patterns": ["*"]}) + + status = generate_pytest_api_cov_report(config, {}, ["GET /a", "GET /b"]) + + assert status == 0 + note_print = next(c for c in mock_console.print.call_args_list if "excluded" in c.args[0]) + assert "coverage requirement not applied" in note_print.args[0] diff --git a/uv.lock b/uv.lock index 017f07e..0cdd31c 100644 --- a/uv.lock +++ b/uv.lock @@ -756,7 +756,7 @@ requires-dist = [ { name = "flask", marker = "extra == 'flask'", specifier = ">=2.0.0" }, { name = "httpx", marker = "extra == 'fastapi'", specifier = ">=0.20.0" }, { name = "pydantic", specifier = ">=2.0.0" }, - { name = "pytest", specifier = ">=6.0.0" }, + { name = "pytest", specifier = ">=7.0.0" }, { name = "pyyaml", specifier = ">=6.0" }, { name = "rich", specifier = ">=10.0.0" }, { name = "starlette", marker = "extra == 'fastapi'", specifier = ">=0.14.0" }, From 8fae2cd7199722edbc8c3031924aefd4a0389506 Mon Sep 17 00:00:00 2001 From: BarnabasG Date: Wed, 8 Jul 2026 21:32:06 +0100 Subject: [PATCH 2/5] fix: address verified code-review findings on the audit-fix branch Each fix was validated with a dedicated repro check run before (failing) and after (passing) the change; all ten now pass. Django route templates (frameworks.py): - Regex shorthand (\d, \w), bare character classes and bare dots outside groups become placeholders instead of garbage literals; quantifiers are consumed; a trailing '/?' is dropped - Groups whose body can span '/' (bare dot, '/', \S) get a converter so multi-segment values like (?P.*) still match after the single-segment matching change - Dispatch-only CBVs (no verb handlers) keep the default method set instead of vanishing from discovery Recording and discovery: - FastAPI: the requested path is always credited (a redirecting route keeps its own coverage) and a followed redirect also credits the final route - Flask: static-rule exclusion additionally requires the rule to end in /, so user routes named '*.static' are kept - Django detection: isinstance check against BaseHandler so handler subclasses in user modules are detected; optional-framework classes are resolved via sys.modules with failed imports memoised (no repeated finder scans, and robust to pytester's sys.modules snapshotting) Gate semantics (report.py): - All-endpoints-excluded now fails closed when fail_under > 0 (an over-broad exclusion pattern can no longer silently disable the gate) and passes when fail_under == 0 - Empty discovery only hard-fails for fail_under > 0; an explicit 0 is trivially satisfied - Method-scoped exclusion patterns are applied before method grouping so they keep working under --api-cov-group-methods-by-endpoint Tests: 12 new regression tests covering every finding; updated the two tests whose expectations changed (all-excluded gate, redirect recording). --- src/pytest_api_cov/frameworks.py | 147 ++++++++++++++---- src/pytest_api_cov/report.py | 39 +++-- tests/integration/test_django_integration.py | 9 +- .../test_frameworks_integration.py | 61 +++++++- tests/unit/test_frameworks.py | 44 ++++++ tests/unit/test_report.py | 39 ++++- 6 files changed, 287 insertions(+), 52 deletions(-) diff --git a/src/pytest_api_cov/frameworks.py b/src/pytest_api_cov/frameworks.py index 1f98d66..12e0f55 100644 --- a/src/pytest_api_cov/frameworks.py +++ b/src/pytest_api_cov/frameworks.py @@ -2,6 +2,7 @@ from __future__ import annotations +import importlib import re import sys from abc import ABC, abstractmethod @@ -46,9 +47,12 @@ class FlaskAdapter(BaseAdapter): @staticmethod def _is_static_rule(rule: Any) -> bool: - """Match app and blueprint static rules by endpoint name, covering custom static_url_path.""" + """Match app and blueprint static-file rules without dropping user routes sharing the name.""" endpoint = str(getattr(rule, "endpoint", "")) - return endpoint == "static" or endpoint.endswith(".static") + if endpoint != "static" and not endpoint.endswith(".static"): + return False + # Framework static rules always end in the filename path converter. + return str(getattr(rule, "rule", "")).endswith("/") def get_endpoints(self) -> list[str]: """Return list of 'METHOD /path' strings.""" @@ -157,28 +161,65 @@ def get_tracked_client(self, recorder: ApiCallRecorder | None, test_name: str) - class TrackingFastAPIClient(TestClient): def send(self, *args: Any, **kwargs: Any) -> Any: request = args[0] + method = request.method.upper() + original_path = request.url.path try: response = super().send(*args, **kwargs) except BaseException: if recorder is not None: - recorder.record_call(request.url.path, test_name, request.method.upper()) + recorder.record_call(original_path, test_name, method) raise if recorder is not None: - # httpx follows redirects inside send(); response.request points at the - # final request, so followed slash-redirects record the real route path. - final_request = getattr(response, "request", request) - recorder.record_call(final_request.url.path, test_name, final_request.method.upper()) + # The requested endpoint always gets credit (it may itself be a + # redirecting route); httpx follows redirects inside send(), so a + # followed redirect also credits the final route it landed on. + recorder.record_call(original_path, test_name, method) + final_request = getattr(response, "request", None) + if final_request is not None: + final_path = getattr(getattr(final_request, "url", None), "path", None) + if final_path and final_path != original_path: + recorder.record_call(final_path, test_name, final_request.method.upper()) return response return TrackingFastAPIClient(self.app) +_REGEX_SHORTHAND_CLASSES = frozenset("dDwWsS") + + +def _consume_quantifier(route: str, i: int) -> int: + """Return the index just past a regex quantifier starting at ``i``, if any.""" + if i < len(route) and route[i] in "+*?": + return i + 1 + if i < len(route) and route[i] == "{": + end = route.find("}", i) + if end != -1 and re.fullmatch(r"\{\d+(,\d*)?\}", route[i : end + 1]): + return end + 1 + return i + + +def _group_placeholder(group: str, param_count: int) -> tuple[str, int]: + """Choose a placeholder for a regex group; bodies that can span '/' get a path converter.""" + named = re.match(r"\(\?P<(\w+)>", group) + body = group[named.end() : -1] if named else group[1:-1] + multi_segment = "/" in body or re.search(r"(?" if multi_segment else f"<{name}>"), param_count + + def _django_route_to_template(route: str) -> str: r"""Convert a Django route string to a matchable template. - ``path()`` routes pass through unchanged; ``re_path()`` regex groups - (``(?P[0-9]{4})``) become ```` placeholders and escaped - literals (``\.``) are unescaped so recorded request paths can match. + ``path()`` routes pass through unchanged. In ``re_path()`` regexes, groups + (``(?P[0-9]{4})``), shorthand classes (``\d+``), bare character + classes (``[0-9]+``) and bare dots become placeholders (``path:`` variants + when the pattern can span ``/``); escaped literals (``\.``) are unescaped + and bare quantifiers (a trailing ``/?``) are dropped, so recorded request + paths can match the template. """ out: list[str] = [] i = 0 @@ -187,8 +228,14 @@ def _django_route_to_template(route: str) -> str: while i < n: char = route[i] if char == "\\" and i + 1 < n: - out.append(route[i + 1]) - i += 2 + escaped = route[i + 1] + if escaped in _REGEX_SHORTHAND_CLASSES: + param_count += 1 + out.append(f"") + i = _consume_quantifier(route, i + 2) + else: + out.append(escaped) + i += 2 elif char == "(": depth = 0 j = i @@ -209,13 +256,25 @@ def _django_route_to_template(route: str) -> str: if depth == 0: break j += 1 - named = re.match(r"\(\?P<(\w+)>", route[i : j + 1]) - if named: - out.append(f"<{named.group(1)}>") - else: - param_count += 1 - out.append(f"") - i = j + 1 + placeholder, param_count = _group_placeholder(route[i : j + 1], param_count) + out.append(placeholder) + i = _consume_quantifier(route, j + 1) + elif char == "[": + j = i + 1 + while j < n and route[j] != "]": + j += 2 if route[j] == "\\" else 1 + param_count += 1 + out.append(f"") + i = _consume_quantifier(route, j + 1) + elif char == ".": + end = _consume_quantifier(route, i + 1) + param_count += 1 + # A quantified dot (.* / .+) can cross path segments. + out.append(f"" if end > i + 1 else f"") + i = end + elif char in "+*?": + # Bare quantifier on the preceding literal (e.g. a trailing '/?'): drop it. + i += 1 else: out.append(char) i += 1 @@ -245,7 +304,11 @@ def _extract_patterns(patterns: list[Any], prefix: str = "") -> None: if view_class is not None and hasattr(view_class, "http_method_names"): # Only count methods the class actually implements (mirrors # View._allowed_methods), not the full http_method_names list. - methods = {m.upper() for m in view_class.http_method_names if hasattr(view_class, m)} + implemented = {m.upper() for m in view_class.http_method_names if hasattr(view_class, m)} + if implemented - {"HEAD", "OPTIONS"}: + methods = implemented + # else: dispatch()-only view — keep the default set so the + # endpoint stays discoverable at all. endpoints.extend(f"{method} {full_path}" for method in methods if method not in ("HEAD", "OPTIONS")) @@ -286,24 +349,40 @@ def _unwrap_wsgi_app(app: Any) -> Any: return None -def _detect_by_isinstance(app: Any) -> SupportedFramework | None: - """Detect Flask/FastAPI apps (including subclasses) via isinstance checks.""" - try: - from flask import Flask +_FRAMEWORK_CLASS_SPECS: tuple[tuple[SupportedFramework, str, str], ...] = ( + (SupportedFramework.FLASK, "flask", "Flask"), + (SupportedFramework.FASTAPI, "fastapi", "FastAPI"), + (SupportedFramework.DJANGO, "django.core.handlers.base", "BaseHandler"), +) - if isinstance(app, Flask): - return SupportedFramework.FLASK - except ImportError: - pass +_import_failed: set[str] = set() - try: - from fastapi import FastAPI - if isinstance(app, FastAPI): - return SupportedFramework.FASTAPI - except ImportError: - pass +def _optional_class(module_name: str, attr: str) -> type[Any] | None: + """Resolve a class from an optional dependency. + Resolves through sys.modules so reloaded modules stay consistent, and + remembers failed imports so missing frameworks are only probed once. + """ + if module_name in _import_failed: + return None + module = sys.modules.get(module_name) + if module is None: + try: + module = importlib.import_module(module_name) + except ImportError: + _import_failed.add(module_name) + return None + cls = getattr(module, attr, None) + return cls if isinstance(cls, type) else None + + +def _detect_by_isinstance(app: Any) -> SupportedFramework | None: + """Detect framework apps (including subclasses) via isinstance checks.""" + for framework, module_name, attr in _FRAMEWORK_CLASS_SPECS: + framework_class = _optional_class(module_name, attr) + if framework_class is not None and isinstance(app, framework_class): + return framework return None diff --git a/src/pytest_api_cov/report.py b/src/pytest_api_cov/report.py index f7a9d1c..df3f7fa 100644 --- a/src/pytest_api_cov/report.py +++ b/src/pytest_api_cov/report.py @@ -217,7 +217,8 @@ def generate_pytest_api_cov_report( console = Console() if not discovered_endpoints: - if api_cov_config.fail_under is not None: + # A truthy threshold (> 0) cannot be met without endpoints; an explicit 0 can. + if api_cov_config.fail_under: console.print( f"\n[bold red]FAIL: No endpoints discovered but --api-cov-fail-under={api_cov_config.fail_under} " "is set. Check your app/client fixtures or OpenAPI spec.[/bold red]" @@ -230,13 +231,16 @@ def generate_pytest_api_cov_report( console.print(f"\n\n[bold blue]{separator} API Coverage Report {separator}[/bold blue]") if api_cov_config.group_methods_by_endpoint: - discovered_endpoints, called_data = group_endpoints_by_path(discovered_endpoints, called_data) - - covered, uncovered, excluded = categorise_endpoints( - discovered_endpoints, - called_data, - api_cov_config.exclusion_patterns, - ) + # Apply (possibly method-scoped) exclusions before collapsing methods away. + _, kept, excluded = categorise_endpoints(discovered_endpoints, {}, api_cov_config.exclusion_patterns) + grouped_endpoints, called_data = group_endpoints_by_path(kept, called_data) + covered, uncovered, _ = categorise_endpoints(grouped_endpoints, called_data, []) + else: + covered, uncovered, excluded = categorise_endpoints( + discovered_endpoints, + called_data, + api_cov_config.exclusion_patterns, + ) if api_cov_config.show_uncovered_endpoints: print_endpoints( @@ -271,11 +275,20 @@ def generate_pytest_api_cov_report( if api_cov_config.fail_under is None: console.print(f"\n[bold green]Total API Coverage: {coverage}%[/bold green]") elif not covered and not uncovered: - # Every endpoint was excluded: nothing is measurable, so the gate is vacuous. - console.print( - f"\n[bold yellow]All {len(excluded)} discovered endpoints are excluded; " - "coverage requirement not applied.[/bold yellow]" - ) + # Every endpoint was excluded: nothing is measurable. Fail closed when a + # real threshold is set, so an over-broad pattern cannot disable the gate. + if api_cov_config.fail_under: + console.print( + f"\n[bold red]FAIL: All {len(excluded)} discovered endpoints are excluded, so no coverage " + f"can be measured against the requirement of {api_cov_config.fail_under}%. " + "Loosen the exclusion patterns or remove fail_under.[/bold red]" + ) + status = 1 + else: + console.print( + f"\n[bold yellow]All {len(excluded)} discovered endpoints are excluded; " + "coverage requirement of 0% is trivially met.[/bold yellow]" + ) elif coverage < api_cov_config.fail_under: console.print( f"\n[bold red]FAIL: Required coverage of {api_cov_config.fail_under}% not met. " diff --git a/tests/integration/test_django_integration.py b/tests/integration/test_django_integration.py index 48bc7b5..8f8a00d 100644 --- a/tests/integration/test_django_integration.py +++ b/tests/integration/test_django_integration.py @@ -74,11 +74,16 @@ class GetOnlyView(View): def get(self, request): return JsonResponse({"ok": True}) + class DispatchOnlyView(View): + def dispatch(self, request, *args, **kwargs): + return JsonResponse({"ok": True}) + def year_view(request, year): return JsonResponse({"year": year}) urlpatterns = [ path("only-get/", GetOnlyView.as_view()), + path("dispatch-only/", DispatchOnlyView.as_view()), re_path(r"^articles/(?P[0-9]{4})/$", year_view), ] """ @@ -119,5 +124,7 @@ def test_articles(coverage_client): # CBVs only count implemented handlers; FBVs keep the 5-method default. assert "POST /only-get/" not in output assert "POST /articles//" in output - assert "Total API Coverage: 16.67%" in output + # Dispatch-only CBVs stay discoverable with the default method set. + assert "GET /dispatch-only/" in output + assert "Total API Coverage: 9.09%" in output assert result.ret == 0 diff --git a/tests/integration/test_frameworks_integration.py b/tests/integration/test_frameworks_integration.py index 08ebe98..5ae09e6 100644 --- a/tests/integration/test_frameworks_integration.py +++ b/tests/integration/test_frameworks_integration.py @@ -357,5 +357,64 @@ def items(): response = client.get("/items", follow_redirects=True) assert response.status_code == 200 + # Both the requested path and the redirect target get credit. assert "GET /items/" in recorder - assert "GET /items" not in recorder + assert "GET /items" in recorder + + def test_redirecting_endpoint_keeps_its_own_coverage_credit(self): + """A route that returns RedirectResponse is itself recorded as covered.""" + try: + from fastapi import FastAPI + from fastapi.responses import RedirectResponse + except ImportError: + pytest.skip("FastAPI not available for integration testing") + + app = FastAPI() + + @app.get("/old") + def old(): + return RedirectResponse("/new") + + @app.get("/new") + def new(): + return {"ok": True} + + recorder = ApiCallRecorder() + client = FastAPIAdapter(app).get_tracked_client(recorder, "test_redirect_source") + + response = client.get("/old", follow_redirects=True) + + assert response.status_code == 200 + assert "GET /old" in recorder + assert "GET /new" in recorder + + def test_user_route_with_static_endpoint_name_is_kept(self): + """A real Flask route whose endpoint name ends in '.static' must not be dropped.""" + try: + from flask import Flask + except ImportError: + pytest.skip("Flask not available for integration testing") + + app = Flask(__name__, static_folder=None) + app.add_url_rule("/assets/report", endpoint="docs.static", view_func=lambda: "hi") + + endpoints = FlaskAdapter(app).get_endpoints() + + assert "GET /assets/report" in endpoints + + def test_django_handler_subclass_in_user_module_is_detected(self): + """WSGIHandler subclasses defined outside the django package are detected.""" + try: + from django.core.handlers.wsgi import WSGIHandler + except ImportError: + pytest.skip("Django not available for integration testing") + + from pytest_api_cov.frameworks import DjangoAdapter, get_framework_adapter + + class MyHandler(WSGIHandler): + pass + + MyHandler.__module__ = "mydjango_utils.handlers" + instance = object.__new__(MyHandler) + + assert isinstance(get_framework_adapter(instance), DjangoAdapter) diff --git a/tests/unit/test_frameworks.py b/tests/unit/test_frameworks.py index e519352..e636e9f 100644 --- a/tests/unit/test_frameworks.py +++ b/tests/unit/test_frameworks.py @@ -253,3 +253,47 @@ def test_path_route_passes_through(self): from pytest_api_cov.frameworks import _django_route_to_template assert _django_route_to_template("articles//") == "articles//" + + def test_shorthand_class_outside_group_becomes_placeholder(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template(r"v\d+/users/") == "v/users/" + + def test_bare_character_class_becomes_placeholder(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template("v[0-9]+/users/") == "v/users/" + + def test_multi_segment_group_gets_path_converter(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template("files/(?P.*)") == "files/" + + def test_trailing_optional_slash_quantifier_is_dropped(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template(r"articles/(?P[\w-]+)/?") == "articles//" + + def test_failed_framework_import_is_probed_only_once(self, monkeypatch): + import pytest_api_cov.frameworks as frameworks + + calls = [] + + def counting_import(name, *args, **kwargs): + calls.append(name) + raise ImportError(name) + + monkeypatch.setattr(frameworks.importlib, "import_module", counting_import) + monkeypatch.setattr(frameworks, "_import_failed", set()) + + assert frameworks._optional_class("not_a_real_framework_xyz", "App") is None + assert frameworks._optional_class("not_a_real_framework_xyz", "App") is None + assert calls.count("not_a_real_framework_xyz") == 1 + + def test_optional_class_resolves_current_module_object(self): + import sys + + from pytest_api_cov.frameworks import _optional_class + + flask_class = _optional_class("flask", "Flask") + assert flask_class is sys.modules["flask"].Flask diff --git a/tests/unit/test_report.py b/tests/unit/test_report.py index c10e6cf..7edca76 100644 --- a/tests/unit/test_report.py +++ b/tests/unit/test_report.py @@ -399,12 +399,45 @@ def test_generate_report_fail_under_with_no_endpoints(self, mock_console_cls): @patch("pytest_api_cov.report.Console") def test_generate_report_fail_under_with_all_endpoints_excluded(self, mock_console_cls): - """Excluding every endpoint makes the gate vacuous rather than a spurious failure.""" + """A real threshold fails closed when exclusions leave nothing measurable.""" mock_console = mock_console_cls.return_value config = ApiCoverageReportConfig.model_validate({"fail_under": 80.0, "exclusion_patterns": ["*"]}) status = generate_pytest_api_cov_report(config, {}, ["GET /a", "GET /b"]) + assert status == 1 + fail_print = next(c for c in mock_console.print.call_args_list if "FAIL" in c.args[0]) + assert "All 2 discovered endpoints are excluded" in fail_print.args[0] + + @patch("pytest_api_cov.report.Console") + def test_generate_report_fail_under_zero_with_all_endpoints_excluded(self, mock_console_cls): + """An explicit 0% threshold is trivially met even when everything is excluded.""" + config = ApiCoverageReportConfig.model_validate({"fail_under": 0.0, "exclusion_patterns": ["*"]}) + + status = generate_pytest_api_cov_report(config, {}, ["GET /a"]) + + assert status == 0 + + @patch("pytest_api_cov.report.Console") + def test_generate_report_fail_under_zero_with_no_endpoints(self, mock_console_cls): + """An explicit 0% threshold does not hard-fail on empty discovery.""" + config = ApiCoverageReportConfig.model_validate({"fail_under": 0.0}) + + status = generate_pytest_api_cov_report(config, {}, []) + + assert status == 0 + + @patch("pytest_api_cov.report.Console") + def test_method_scoped_exclusions_apply_before_grouping(self, mock_console_cls): + """Method-scoped exclusion patterns still work with group-methods-by-endpoint.""" + config = ApiCoverageReportConfig.model_validate( + { + "group_methods_by_endpoint": True, + "exclusion_patterns": ["GET /health"], + "fail_under": 100.0, + } + ) + + status = generate_pytest_api_cov_report(config, {"GET /api": {"t"}}, ["GET /health", "GET /api"]) + assert status == 0 - note_print = next(c for c in mock_console.print.call_args_list if "excluded" in c.args[0]) - assert "coverage requirement not applied" in note_print.args[0] From 49e162afada8e6071e7369c8bd9119024de2b57c Mon Sep 17 00:00:00 2001 From: BarnabasG Date: Wed, 8 Jul 2026 21:50:52 +0100 Subject: [PATCH 3/5] refactor: simplification pass over the audit-fix changes plugin.py: - Extract _tracked_client_flow: create_coverage_fixture and coverage_client shared ~45 duplicated lines of client-resolution flow that had already drifted; both now delegate to one generator (coverage_client additionally gains create_coverage_fixture's catch around tracked-client construction) - Share DEFAULT_CLIENT_FIXTURE_NAMES with config.py instead of a second hardcoded tuple; collapse the doubled --api-cov-report getoption in pytest_configure; 'except (FixtureLookupError, Exception)' -> Exception; bare partition() replaces three 'if "?" in url' conditionals; xdist endpoint merge uses dict.fromkeys for ordered dedup - Parse the OpenAPI spec once per session: a spec yielding zero endpoints was re-read and re-parsed on every test (new SessionData.openapi_discovery_attempted flag) report.py: - Split categorise_endpoints into _partition_excluded + _match_covered (categorise_endpoints stays as their composition), collapsing the quadruplicated exclusion/negation match loops into one _matches_any helper and turning generate's grouping branch into a linear pipeline with no sentinel arguments - _split_endpoint helper replaces four inline METHOD/path split idioms; group_endpoints_by_path uses dict.fromkeys; _compile_exclusion_patterns returns tuples instead of Optionals; footer now derives from the header string (was 2 chars wider via magic arithmetic); endpoint_to_regex cache unbounded so >512-endpoint sessions stop thrashing frameworks.py: - Merge the APIRoute and plain-Route discovery branches; _SKIPPED_METHODS constant replaces four ('HEAD', 'OPTIONS') literals - Django template scanner: quantifier matching via one compiled regex, _skip_class helper replaces duplicated character-class handling, and an itertools.count param namer removes the param_count tuple-threading - Bind the recorder once after the None guard in all three tracking clients, deleting dead 'if recorder is not None' checks and a type-ignore - Inline single-caller _detect_by_isinstance into _detect_framework; drop the duplicate None check in is_supported_framework openapi.py: logger.exception implies exc_info; drop the redundant argument. --- src/pytest_api_cov/config.py | 4 +- src/pytest_api_cov/frameworks.py | 135 +++++++++++------------ src/pytest_api_cov/models.py | 1 + src/pytest_api_cov/openapi.py | 2 +- src/pytest_api_cov/plugin.py | 183 +++++++++++++------------------ src/pytest_api_cov/report.py | 144 +++++++++++------------- 6 files changed, 209 insertions(+), 260 deletions(-) diff --git a/src/pytest_api_cov/config.py b/src/pytest_api_cov/config.py index e54cb27..761f16d 100644 --- a/src/pytest_api_cov/config.py +++ b/src/pytest_api_cov/config.py @@ -9,6 +9,8 @@ import tomli from pydantic import BaseModel, ConfigDict, Field +DEFAULT_CLIENT_FIXTURE_NAMES = ("client", "test_client", "api_client", "app_client") + class ApiCoverageReportConfig(BaseModel): """Configuration model for API coverage reporting.""" @@ -24,7 +26,7 @@ class ApiCoverageReportConfig(BaseModel): force_sugar: bool = Field(default=False, alias="api-cov-force-sugar") force_sugar_disabled: bool = Field(default=False, alias="api-cov-force-sugar-disabled") client_fixture_names: list[str] = Field( - ["client", "test_client", "api_client", "app_client"], alias="api-cov-client-fixture-names" + default_factory=lambda: list(DEFAULT_CLIENT_FIXTURE_NAMES), alias="api-cov-client-fixture-names" ) group_methods_by_endpoint: bool = Field(default=False, alias="api-cov-group-methods-by-endpoint") openapi_spec: str | None = Field(None, alias="api-cov-openapi-spec") diff --git a/src/pytest_api_cov/frameworks.py b/src/pytest_api_cov/frameworks.py index 12e0f55..e088e5a 100644 --- a/src/pytest_api_cov/frameworks.py +++ b/src/pytest_api_cov/frameworks.py @@ -6,6 +6,7 @@ import re import sys from abc import ABC, abstractmethod +from itertools import count from typing import TYPE_CHECKING, Any if sys.version_info >= (3, 11): @@ -23,8 +24,13 @@ class SupportedFramework(StrEnum): if TYPE_CHECKING: + from collections.abc import Callable + from .models import ApiCallRecorder +# Auto-added companions of GET et al. that would inflate the endpoint count. +_SKIPPED_METHODS = frozenset({"HEAD", "OPTIONS"}) + class BaseAdapter(ABC): """Abstract base for framework adapters.""" @@ -61,7 +67,7 @@ def get_endpoints(self) -> list[str]: for rule in self.app.url_map.iter_rules() if not self._is_static_rule(rule) for method in rule.methods - if method not in ("HEAD", "OPTIONS") + if method not in _SKIPPED_METHODS ] return sorted(endpoints) @@ -76,6 +82,7 @@ def get_tracked_client(self, recorder: ApiCallRecorder | None, test_name: str) - if recorder is None: return self.app.test_client() + active_recorder = recorder url_adapter = None if hasattr(self.app.url_map, "bind"): url_adapter = self.app.url_map.bind("") @@ -105,7 +112,7 @@ def open(self, *args: Any, **kwargs: Any) -> Any: if isinstance(path, str) and url_adapter is not None: rule = _match_rule(path.partition("?")[0], method, bool(kwargs.get("follow_redirects"))) if rule is not None: - recorder.record_call(rule.rule, test_name, method) # type: ignore[union-attr] + active_recorder.record_call(rule.rule, test_name, method) return super().open(*args, **kwargs) return TrackingFlaskClient(self.app, self.app.response_class) @@ -126,18 +133,12 @@ def _collect_routes(self, routes: list[Any], prefix: str, endpoints: list[str]) from starlette.routing import Mount, Route for route in routes: - if isinstance(route, APIRoute): - endpoints.extend( - f"{method} {prefix}{route.path}" for method in route.methods if method not in ("HEAD", "OPTIONS") - ) - elif isinstance(route, Route) and not isinstance(route, Mount): - # Plain Starlette routes (add_route, mounted Starlette apps). The auto-generated - # docs routes (/docs, /openapi.json, ...) carry include_in_schema=False. - if not getattr(route, "include_in_schema", True): - continue + # APIRoutes always count; plain Starlette routes count unless flagged out of + # the schema (the auto-generated /docs, /openapi.json, ... routes are). + if isinstance(route, APIRoute) or (isinstance(route, Route) and getattr(route, "include_in_schema", True)): methods = route.methods or {"GET"} endpoints.extend( - f"{method} {prefix}{route.path}" for method in methods if method not in ("HEAD", "OPTIONS") + f"{method} {prefix}{route.path}" for method in methods if method not in _SKIPPED_METHODS ) elif isinstance(route, Mount): mount_prefix = prefix + route.path @@ -158,6 +159,8 @@ def get_tracked_client(self, recorder: ApiCallRecorder | None, test_name: str) - if recorder is None: return TestClient(self.app) + active_recorder = recorder + class TrackingFastAPIClient(TestClient): def send(self, *args: Any, **kwargs: Any) -> Any: request = args[0] @@ -166,49 +169,47 @@ def send(self, *args: Any, **kwargs: Any) -> Any: try: response = super().send(*args, **kwargs) except BaseException: - if recorder is not None: - recorder.record_call(original_path, test_name, method) + active_recorder.record_call(original_path, test_name, method) raise - if recorder is not None: - # The requested endpoint always gets credit (it may itself be a - # redirecting route); httpx follows redirects inside send(), so a - # followed redirect also credits the final route it landed on. - recorder.record_call(original_path, test_name, method) - final_request = getattr(response, "request", None) - if final_request is not None: - final_path = getattr(getattr(final_request, "url", None), "path", None) - if final_path and final_path != original_path: - recorder.record_call(final_path, test_name, final_request.method.upper()) + # The requested endpoint always gets credit (it may itself be a + # redirecting route); httpx follows redirects inside send(), so a + # followed redirect also credits the final route it landed on. + active_recorder.record_call(original_path, test_name, method) + final_request = getattr(response, "request", None) + if final_request is not None: + final_path = getattr(getattr(final_request, "url", None), "path", None) + if final_path and final_path != original_path: + active_recorder.record_call(final_path, test_name, final_request.method.upper()) return response return TrackingFastAPIClient(self.app) _REGEX_SHORTHAND_CLASSES = frozenset("dDwWsS") +_QUANTIFIER_PATTERN = re.compile(r"[+*?]|\{\d+(?:,\d*)?\}") def _consume_quantifier(route: str, i: int) -> int: """Return the index just past a regex quantifier starting at ``i``, if any.""" - if i < len(route) and route[i] in "+*?": - return i + 1 - if i < len(route) and route[i] == "{": - end = route.find("}", i) - if end != -1 and re.fullmatch(r"\{\d+(,\d*)?\}", route[i : end + 1]): - return end + 1 - return i + match = _QUANTIFIER_PATTERN.match(route, i) + return match.end() if match else i -def _group_placeholder(group: str, param_count: int) -> tuple[str, int]: +def _skip_class(route: str, i: int) -> int: + """Return the index of the ']' closing the character class opened at ``route[i]``.""" + j = i + 1 + while j < len(route) and route[j] != "]": + j += 2 if route[j] == "\\" else 1 + return j + + +def _group_placeholder(group: str, next_param: Callable[[], str]) -> str: """Choose a placeholder for a regex group; bodies that can span '/' get a path converter.""" named = re.match(r"\(\?P<(\w+)>", group) body = group[named.end() : -1] if named else group[1:-1] multi_segment = "/" in body or re.search(r"(?" if multi_segment else f"<{name}>"), param_count + name = named.group(1) if named else next_param() + return f"" if multi_segment else f"<{name}>" def _django_route_to_template(route: str) -> str: @@ -223,15 +224,18 @@ def _django_route_to_template(route: str) -> str: """ out: list[str] = [] i = 0 - param_count = 0 n = len(route) + param_counter = count(1) + + def next_param() -> str: + return f"param{next(param_counter)}" + while i < n: char = route[i] if char == "\\" and i + 1 < n: escaped = route[i + 1] if escaped in _REGEX_SHORTHAND_CLASSES: - param_count += 1 - out.append(f"") + out.append(f"<{next_param()}>") i = _consume_quantifier(route, i + 2) else: out.append(escaped) @@ -239,16 +243,13 @@ def _django_route_to_template(route: str) -> str: elif char == "(": depth = 0 j = i - in_class = False while j < n: inner = route[j] if inner == "\\": j += 2 continue - if in_class: - in_class = inner != "]" - elif inner == "[": - in_class = True + if inner == "[": + j = _skip_class(route, j) elif inner == "(": depth += 1 elif inner == ")": @@ -256,21 +257,16 @@ def _django_route_to_template(route: str) -> str: if depth == 0: break j += 1 - placeholder, param_count = _group_placeholder(route[i : j + 1], param_count) - out.append(placeholder) + out.append(_group_placeholder(route[i : j + 1], next_param)) i = _consume_quantifier(route, j + 1) elif char == "[": - j = i + 1 - while j < n and route[j] != "]": - j += 2 if route[j] == "\\" else 1 - param_count += 1 - out.append(f"") + j = _skip_class(route, i) + out.append(f"<{next_param()}>") i = _consume_quantifier(route, j + 1) elif char == ".": end = _consume_quantifier(route, i + 1) - param_count += 1 # A quantified dot (.* / .+) can cross path segments. - out.append(f"" if end > i + 1 else f"") + out.append(f"" if end > i + 1 else f"<{next_param()}>") i = end elif char in "+*?": # Bare quantifier on the preceding literal (e.g. a trailing '/?'): drop it. @@ -305,12 +301,12 @@ def _extract_patterns(patterns: list[Any], prefix: str = "") -> None: # Only count methods the class actually implements (mirrors # View._allowed_methods), not the full http_method_names list. implemented = {m.upper() for m in view_class.http_method_names if hasattr(view_class, m)} - if implemented - {"HEAD", "OPTIONS"}: + if implemented - _SKIPPED_METHODS: methods = implemented # else: dispatch()-only view — keep the default set so the # endpoint stays discoverable at all. - endpoints.extend(f"{method} {full_path}" for method in methods if method not in ("HEAD", "OPTIONS")) + endpoints.extend(f"{method} {full_path}" for method in methods if method not in _SKIPPED_METHODS) elif isinstance(pattern, URLResolver): route = _django_route_to_template(str(pattern.pattern).strip("^$")) @@ -326,13 +322,14 @@ def get_tracked_client(self, recorder: ApiCallRecorder | None, test_name: str) - if recorder is None: return Client() + active_recorder = recorder + class TrackingDjangoClient(Client): # type: ignore[misc] def request(self, **request: Any) -> Any: method = request.get("REQUEST_METHOD", "GET").upper() path = request.get("PATH_INFO", "/") - if recorder is not None: - recorder.record_call(path, test_name, method) + active_recorder.record_call(path, test_name, method) return super().request(**request) @@ -377,25 +374,17 @@ def _optional_class(module_name: str, attr: str) -> type[Any] | None: return cls if isinstance(cls, type) else None -def _detect_by_isinstance(app: Any) -> SupportedFramework | None: - """Detect framework apps (including subclasses) via isinstance checks.""" - for framework, module_name, attr in _FRAMEWORK_CLASS_SPECS: - framework_class = _optional_class(module_name, attr) - if framework_class is not None and isinstance(app, framework_class): - return framework - return None - - def _detect_framework(app: Any) -> SupportedFramework | None: """Detect the framework, supporting app subclasses via isinstance checks.""" if app is None: return None - framework = _detect_by_isinstance(app) - if framework is not None: - return framework + for framework, module_name, attr in _FRAMEWORK_CLASS_SPECS: + framework_class = _optional_class(module_name, attr) + if framework_class is not None and isinstance(app, framework_class): + return framework - # Name-based fallback for Django handlers and duck-typed apps. + # Name-based fallback for duck-typed apps (e.g. mocks in test suites). app_type = type(app).__name__ module_name = getattr(type(app), "__module__", "").split(".")[0] @@ -412,8 +401,6 @@ def _detect_framework(app: Any) -> SupportedFramework | None: def is_supported_framework(app: Any) -> bool: """Check if the app is a supported framework.""" - if app is None: - return False return _detect_framework(app) is not None diff --git a/src/pytest_api_cov/models.py b/src/pytest_api_cov/models.py index 7883710..2bcee82 100644 --- a/src/pytest_api_cov/models.py +++ b/src/pytest_api_cov/models.py @@ -104,6 +104,7 @@ class SessionData(BaseModel): recorder: ApiCallRecorder = Field(default_factory=ApiCallRecorder) discovered_endpoints: EndpointDiscovery = Field(default_factory=EndpointDiscovery) discovery_complete: bool = Field(default=False) + openapi_discovery_attempted: bool = Field(default=False) def record_call(self, endpoint: str, test_name: str, method: str = "GET") -> None: """Record an API call.""" diff --git a/src/pytest_api_cov/openapi.py b/src/pytest_api_cov/openapi.py index 6b766e8..9c7ec96 100644 --- a/src/pytest_api_cov/openapi.py +++ b/src/pytest_api_cov/openapi.py @@ -28,7 +28,7 @@ def parse_openapi_spec(path: str) -> list[str]: spec = json.load(f) except Exception: - logger.exception("Failed to parse OpenAPI spec", exc_info=True) + logger.exception("Failed to parse OpenAPI spec") return [] if not isinstance(spec, dict): diff --git a/src/pytest_api_cov/plugin.py b/src/pytest_api_cov/plugin.py index 16679c2..d1ffc41 100644 --- a/src/pytest_api_cov/plugin.py +++ b/src/pytest_api_cov/plugin.py @@ -5,7 +5,7 @@ import pytest -from .config import ApiCoverageReportConfig, get_pytest_api_cov_report_config +from .config import DEFAULT_CLIENT_FIXTURE_NAMES, ApiCoverageReportConfig, get_pytest_api_cov_report_config from .frameworks import get_framework_adapter, is_supported_framework from .models import SessionData from .openapi import parse_openapi_spec @@ -17,11 +17,13 @@ def _discover_openapi_endpoints(config: ApiCoverageReportConfig, coverage_data: SessionData) -> None: """Discover endpoints from OpenAPI spec if configured.""" - if coverage_data.discovery_complete: + if coverage_data.discovery_complete or coverage_data.openapi_discovery_attempted: return if not config.openapi_spec or coverage_data.discovered_endpoints.endpoints: return + # The spec is session-constant config: parse it once, not per test. + coverage_data.openapi_discovery_attempted = True endpoints = parse_openapi_spec(config.openapi_spec) if not endpoints: logger.warning(f"> No endpoints found in OpenAPI spec: {config.openapi_spec}") @@ -85,20 +87,22 @@ def pytest_addoption(parser: pytest.Parser) -> None: def pytest_configure(config: pytest.Config) -> None: """Configure the pytest session and logging.""" - if config.getoption("--api-cov-report"): - verbosity = config.option.verbose + if not config.getoption("--api-cov-report"): + return - if verbosity >= 2: - log_level = logging.DEBUG - elif verbosity >= 1: - log_level = logging.INFO - else: - log_level = logging.WARNING + verbosity = config.option.verbose - logger.setLevel(log_level) - logger.info("Initializing API coverage plugin...") + if verbosity >= 2: + log_level = logging.DEBUG + elif verbosity >= 1: + log_level = logging.INFO + else: + log_level = logging.WARNING - if config.getoption("--api-cov-report") and config.pluginmanager.hasplugin("xdist"): + logger.setLevel(log_level) + logger.info("Initializing API coverage plugin...") + + if config.pluginmanager.hasplugin("xdist"): config.pluginmanager.register(DeferXdistPlugin(), "defer_xdist_plugin") @@ -118,6 +122,59 @@ def _try_get_fixture(request: pytest.FixtureRequest, names: tuple[str, ...] | li return None +def _tracked_client_flow( + request: pytest.FixtureRequest, + coverage_data: SessionData, + fixture_name: str, + existing_client: Any | None, +) -> Any: + """Coverage-enabled flow shared by coverage_client and create_coverage_fixture. + + Discovers endpoints, then yields (in order of preference) the wrapped + existing client, a fresh tracked client built from the app, or None. + """ + config = get_pytest_api_cov_report_config(request.config) + _discover_openapi_endpoints(config, coverage_data) + + if existing_client is None: + for name in config.client_fixture_names: + try: + existing_client = request.getfixturevalue(name) + logger.info(f"> Found client fixture '{name}' for '{fixture_name}'") + break + except pytest.FixtureLookupError: + continue + + app = extract_app_from_client(existing_client) if existing_client is not None else None + if app is None: + try: + app = request.getfixturevalue("app") + except pytest.FixtureLookupError: + app = None + + _discover_app_endpoints(app, coverage_data, fixture_name) + + if existing_client is not None: + yield wrap_client_with_coverage(existing_client, coverage_data.recorder, request.node.name) + return + + if app is not None: + try: + adapter = get_framework_adapter(app) + client = adapter.get_tracked_client(coverage_data.recorder, request.node.name) + except Exception as e: # noqa: BLE001 + logger.warning(f"> Failed to create tracked client for '{fixture_name}': {e}") + else: + yield client + return + + # Last resort - yield None but don't skip, so tests still run + logger.warning( + f"> '{fixture_name}' could not provide a client; tests will run without API coverage for this fixture." + ) + yield None + + def create_coverage_fixture(fixture_name: str, existing_fixture_name: str | None = None) -> Any: """Create a coverage-enabled fixture with a custom name. @@ -166,51 +223,7 @@ def fixture_func(request: pytest.FixtureRequest) -> Any: yield client return - config = get_pytest_api_cov_report_config(request.config) - _discover_openapi_endpoints(config, coverage_data) - - if existing_client is None: - for name in config.client_fixture_names: - try: - existing_client = request.getfixturevalue(name) - logger.info(f"> Found client fixture '{name}' for '{fixture_name}'") - break - except pytest.FixtureLookupError: - continue - - app = None - if existing_client is not None: - app = extract_app_from_client(existing_client) - - if app is None: - try: - app = request.getfixturevalue("app") - except pytest.FixtureLookupError: - app = None - - _discover_app_endpoints(app, coverage_data, fixture_name) - - if existing_client is not None: - wrapped = wrap_client_with_coverage(existing_client, coverage_data.recorder, request.node.name) - yield wrapped - return - - if app is not None: - try: - adapter = get_framework_adapter(app) - client = adapter.get_tracked_client(coverage_data.recorder, request.node.name) - except Exception as e: # noqa: BLE001 - logger.warning(f"> Failed to create tracked client for '{fixture_name}': {e}") - else: - yield client - return - - # Last resort - yield None but don't skip, so tests still run - logger.warning( - f"> create_coverage_fixture('{fixture_name}') could not provide a client; " - "tests will run without API coverage for this fixture." - ) - yield None + yield from _tracked_client_flow(request, coverage_data, fixture_name, existing_client) fixture_func.__name__ = fixture_name return pytest.fixture(fixture_func) @@ -234,16 +247,15 @@ def _extract_path_and_method(self, name: str, args: Any, kwargs: Any) -> tuple[s req_method = (args[0] if args else kwargs.get("method", "GET")).upper() req_url = args[1] if len(args) > 1 else kwargs.get("url") if isinstance(req_url, str): - return req_url if "?" not in req_url else req_url.partition("?")[0], req_method + return req_url.partition("?")[0], req_method return None # .get(url), .post(url), .open(url), etc. - url is first arg if args: first = args[0] if isinstance(first, str): - path = first if "?" not in first else first.partition("?")[0] method = kwargs.get("method", name).upper() - return path, ("GET" if method == "OPEN" else method) + return first.partition("?")[0], ("GET" if method == "OPEN" else method) if hasattr(first, "url") and hasattr(first.url, "path"): try: @@ -254,9 +266,8 @@ def _extract_path_and_method(self, name: str, args: Any, kwargs: Any) -> tuple[s if kwargs: path_kw = kwargs.get("path") or kwargs.get("url") or kwargs.get("uri") if isinstance(path_kw, str): - path = path_kw if "?" not in path_kw else path_kw.partition("?")[0] method = kwargs.get("method", name).upper() - return path, ("GET" if method == "OPEN" else method) + return path_kw.partition("?")[0], ("GET" if method == "OPEN" else method) return None @@ -312,7 +323,7 @@ def wrap_client_with_coverage(client: Any, recorder: Any, test_name: str) -> Any def _coverage_client_impl(request: pytest.FixtureRequest) -> Any: - """Inner generator shared by coverage_client and create_coverage_fixture.""" + """Inner generator behind the coverage_client fixture.""" session = request.node.session coverage_enabled = bool(session.config.getoption("--api-cov-report")) @@ -320,51 +331,20 @@ def _coverage_client_impl(request: pytest.FixtureRequest) -> Any: if not coverage_enabled or coverage_data is None: # Try common client fixture names then app fixture - found = _try_get_fixture(request, ("client", "test_client", "api_client", "app_client")) + found = _try_get_fixture(request, DEFAULT_CLIENT_FIXTURE_NAMES) if found is not None: yield found return try: app = request.getfixturevalue("app") adapter = get_framework_adapter(app) - except (pytest.FixtureLookupError, Exception): # noqa: BLE001 + except Exception: # noqa: BLE001 yield None else: yield adapter.get_tracked_client(None, request.node.name) return - config = get_pytest_api_cov_report_config(request.config) - _discover_openapi_endpoints(config, coverage_data) - - # Find a client fixture - client = _try_get_fixture(request, config.client_fixture_names) - if client is not None: - logger.info("> Found client fixture") - - app = extract_app_from_client(client) if client else None - if app is None: - try: - app = request.getfixturevalue("app") - except pytest.FixtureLookupError: - app = None - - _discover_app_endpoints(app, coverage_data, "coverage_client") - - if client is not None: - yield wrap_client_with_coverage(client, coverage_data.recorder, request.node.name) - return - - if app is not None: - try: - adapter = get_framework_adapter(app) - except Exception as e: # noqa: BLE001 - logger.warning(f"> Failed to create tracked client: {e}") - else: - yield adapter.get_tracked_client(coverage_data.recorder, request.node.name) - return - - logger.warning("> coverage_client could not provide a client; tests will run without API coverage.") - yield None + yield from _tracked_client_flow(request, coverage_data, "coverage_client", None) @pytest.fixture @@ -438,10 +418,5 @@ def pytest_testnodedown(self, node: Any) -> None: if discovered_endpoints: current_endpoints = getattr(node.config, "worker_discovered_endpoints", []) - seen = set(current_endpoints) - for endpoint in discovered_endpoints: - if endpoint not in seen: - seen.add(endpoint) - current_endpoints.append(endpoint) - node.config.worker_discovered_endpoints = current_endpoints + node.config.worker_discovered_endpoints = list(dict.fromkeys([*current_endpoints, *discovered_endpoints])) logger.debug(f"> Merged discovered endpoints from worker: {discovered_endpoints}") diff --git a/src/pytest_api_cov/report.py b/src/pytest_api_cov/report.py index df3f7fa..8db923c 100644 --- a/src/pytest_api_cov/report.py +++ b/src/pytest_api_cov/report.py @@ -15,7 +15,7 @@ from .config import ApiCoverageReportConfig -@lru_cache(maxsize=512) +@lru_cache(maxsize=None) def endpoint_to_regex(endpoint: str) -> Pattern[str]: """Create a regex pattern from an endpoint by replacing dynamic segments. @@ -42,6 +42,14 @@ def contains_escape_characters(endpoint: str) -> bool: return ("<" in endpoint and ">" in endpoint) or ("{" in endpoint and "}" in endpoint) +def _split_endpoint(endpoint: str) -> tuple[str | None, str]: + """Split 'METHOD /path' into (METHOD, path); method is None when absent.""" + if " " in endpoint: + method, path = endpoint.split(" ", 1) + return method.upper(), path + return None, endpoint + + def _compile_exclusion_pattern(pat: str) -> tuple[frozenset[str] | None, Pattern[str]]: """Compile a single exclusion pattern into a (methods, regex) pair.""" path_pattern = pat.strip() @@ -60,17 +68,55 @@ def _compile_exclusion_pattern(pat: str) -> tuple[frozenset[str] | None, Pattern @lru_cache(maxsize=128) def _compile_exclusion_patterns( patterns: tuple[str, ...], -) -> tuple[_CompiledPatterns | None, _CompiledPatterns | None]: +) -> tuple[_CompiledPatterns, _CompiledPatterns]: """Compile and cache exclusion/negation patterns. Accepts a tuple (hashable) so the result can be cached across calls. """ - exclusion_only = [p for p in patterns if not p.startswith("!")] - negation_only = [p[1:] for p in patterns if p.startswith("!")] + exclusions = tuple(_compile_exclusion_pattern(p) for p in patterns if not p.startswith("!")) + negations = tuple(_compile_exclusion_pattern(p[1:]) for p in patterns if p.startswith("!")) + return exclusions, negations + + +def _matches_any(compiled: _CompiledPatterns, method: str | None, path_only: str, endpoint: str) -> bool: + """Check an endpoint against compiled (methods, regex) patterns.""" + for methods_set, regex in compiled: + if methods_set and (not method or method not in methods_set): + continue + if regex.match(path_only) or regex.match(endpoint): + return True + return False - compiled_exclusions = tuple(_compile_exclusion_pattern(p) for p in exclusion_only) if exclusion_only else None - compiled_negations = tuple(_compile_exclusion_pattern(p) for p in negation_only) if negation_only else None - return compiled_exclusions, compiled_negations + +def _partition_excluded(endpoints: list[str], exclusion_patterns: list[str]) -> tuple[list[str], list[str]]: + """Split endpoints into (kept, excluded); negation patterns override exclusions.""" + if not exclusion_patterns: + return list(endpoints), [] + + compiled_exclusions, compiled_negations = _compile_exclusion_patterns(tuple(exclusion_patterns)) + kept: list[str] = [] + excluded: list[str] = [] + for endpoint in endpoints: + method, path_only = _split_endpoint(endpoint) + is_excluded = _matches_any(compiled_exclusions, method, path_only, endpoint) and not _matches_any( + compiled_negations, method, path_only, endpoint + ) + (excluded if is_excluded else kept).append(endpoint) + return kept, excluded + + +def _match_covered(endpoints: list[str], called_data: dict[str, set[str]]) -> tuple[list[str], list[str]]: + """Split endpoints into (covered, uncovered) against the recorded call keys.""" + covered: list[str] = [] + uncovered: list[str] = [] + for endpoint in endpoints: + if contains_escape_characters(endpoint): + pattern = endpoint_to_regex(endpoint) + is_covered = any(pattern.match(ep) for ep in called_data) + else: + is_covered = endpoint in called_data + (covered if is_covered else uncovered).append(endpoint) + return covered, uncovered def categorise_endpoints( @@ -84,57 +130,8 @@ def categorise_endpoints( HTTP method prefixes. Pattern order matters: exclusions first, then negations override them. """ - covered: list[str] = [] - uncovered: list[str] = [] - excluded: list[str] = [] - - if not exclusion_patterns: - compiled_exclusions = None - compiled_negations = None - else: - compiled_exclusions, compiled_negations = _compile_exclusion_patterns(tuple(exclusion_patterns)) - - for endpoint in endpoints: - is_excluded = False - endpoint_method = None - path_only = endpoint - if " " in endpoint: - endpoint_method, path_only = endpoint.split(" ", 1) - endpoint_method = endpoint_method.upper() - - if compiled_exclusions: - for methods_set, regex in compiled_exclusions: - if methods_set: - if not endpoint_method or endpoint_method not in methods_set: - continue - if regex.match(path_only) or regex.match(endpoint): - is_excluded = True - break - elif regex.match(path_only) or regex.match(endpoint): - is_excluded = True - break - - if is_excluded and compiled_negations: - for methods_set, regex in compiled_negations: - if methods_set: - if not endpoint_method or endpoint_method not in methods_set: - continue - if regex.match(path_only) or regex.match(endpoint): - is_excluded = False - break - elif regex.match(path_only) or regex.match(endpoint): - is_excluded = False - break - - if is_excluded: - excluded.append(endpoint) - continue - if contains_escape_characters(endpoint): - pattern = endpoint_to_regex(endpoint) - is_covered = any(pattern.match(ep) for ep in called_data) - else: - is_covered = endpoint in called_data - covered.append(endpoint) if is_covered else uncovered.append(endpoint) + kept, excluded = _partition_excluded(endpoints, exclusion_patterns) + covered, uncovered = _match_covered(kept, called_data) return covered, uncovered, excluded @@ -143,18 +140,11 @@ def group_endpoints_by_path( called_data: dict[str, set[str]], ) -> tuple[list[str], dict[str, set[str]]]: """Collapse 'METHOD /path' keys to '/path', merging caller sets across methods.""" - grouped_endpoints: list[str] = [] - seen: set[str] = set() - for endpoint in endpoints: - path = endpoint.split(" ", 1)[1] if " " in endpoint else endpoint - if path not in seen: - seen.add(path) - grouped_endpoints.append(path) + grouped_endpoints = list(dict.fromkeys(_split_endpoint(endpoint)[1] for endpoint in endpoints)) grouped_calls: dict[str, set[str]] = {} for key, callers in called_data.items(): - path = key.split(" ", 1)[1] if " " in key else key - grouped_calls.setdefault(path, set()).update(callers) + grouped_calls.setdefault(_split_endpoint(key)[1], set()).update(callers) return grouped_endpoints, grouped_calls @@ -227,20 +217,14 @@ def generate_pytest_api_cov_report( console.print("\n[bold red]No endpoints discovered. Please check your test setup.[/bold red]") return 0 - separator = "=" * 20 - console.print(f"\n\n[bold blue]{separator} API Coverage Report {separator}[/bold blue]") + header = f"{'=' * 20} API Coverage Report {'=' * 20}" + console.print(f"\n\n[bold blue]{header}[/bold blue]") + kept, excluded = _partition_excluded(discovered_endpoints, api_cov_config.exclusion_patterns) if api_cov_config.group_methods_by_endpoint: - # Apply (possibly method-scoped) exclusions before collapsing methods away. - _, kept, excluded = categorise_endpoints(discovered_endpoints, {}, api_cov_config.exclusion_patterns) - grouped_endpoints, called_data = group_endpoints_by_path(kept, called_data) - covered, uncovered, _ = categorise_endpoints(grouped_endpoints, called_data, []) - else: - covered, uncovered, excluded = categorise_endpoints( - discovered_endpoints, - called_data, - api_cov_config.exclusion_patterns, - ) + # Exclusions (possibly method-scoped) apply before methods are collapsed away. + kept, called_data = group_endpoints_by_path(kept, called_data) + covered, uncovered = _match_covered(kept, called_data) if api_cov_config.show_uncovered_endpoints: print_endpoints( @@ -316,5 +300,5 @@ def generate_pytest_api_cov_report( write_report_file(final_report, api_cov_config.report_path) console.print(f"\n[grey50]JSON report saved to {api_cov_config.report_path}[/grey50]") - console.print(f"[bold blue]{'=' * (42 + len(' API Coverage Report '))}[/bold blue]\n") + console.print(f"[bold blue]{'=' * len(header)}[/bold blue]\n") return status From 71189c2468e54f5cec3ef3dcd03cfba6a74ff62e Mon Sep 17 00:00:00 2001 From: BarnabasG Date: Wed, 8 Jul 2026 22:03:16 +0100 Subject: [PATCH 4/5] 1.4.0 many many fixes --- pyproject.toml | 2 +- uv.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e8502e6..d302e81 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pytest-api-cov" -version = "1.3.8" +version = "1.4.0" description = "Pytest Plugin to provide API Coverage statistics for Python Web Frameworks" readme = "README.md" authors = [{ name = "Barnaby Gill", email = "barnabasgill@gmail.com" }] diff --git a/uv.lock b/uv.lock index 0cdd31c..efea6ac 100644 --- a/uv.lock +++ b/uv.lock @@ -705,7 +705,7 @@ wheels = [ [[package]] name = "pytest-api-cov" -version = "1.3.8" +version = "1.4.0" source = { editable = "." } dependencies = [ { name = "backports-strenum", marker = "python_full_version < '3.11'" }, From 29c017c3dbdc75135c20ce630ebed6ed97bcb8dc Mon Sep 17 00:00:00 2001 From: BarnabasG Date: Wed, 8 Jul 2026 23:27:52 +0100 Subject: [PATCH 5/5] refactor: parse Django re_path routes with the stdlib regex parser Replaces the hand-rolled character scanner in _django_route_to_template (and its _consume_quantifier/_skip_class/_group_placeholder helpers) with a walk of the stdlib regex parse tree (re._parser on 3.11+, sre_parse on 3.10). Every rule now reads off a node type instead of character arithmetic, and multi-segment detection inspects what the pattern can actually match rather than grepping its source text. Behavioral deltas, all covered by new tests: - (?P[^/]+) is now correctly single-segment (the old text heuristic saw the '/' inside the negated class and emitted ) - top-level alternation (legacy|new) degrades to a matchable placeholder instead of passing through as an unmatchable literal - bounded repeats (a{2,4}) become placeholders instead of leaking regex syntax into the template - unparseable input passes through verbatim, so a future change to the private parser API degrades to unconverted routes rather than crashing (discovery already guards adapter failures) Known trade-off, accepted deliberately: re._parser is a private stdlib API. It has been stable for two decades, a large ecosystem (hypothesis, sre_yield, exrex) walks the same tree, CI covers 3.10-3.14, and the verbatim fallback bounds the failure mode. Verified on 3.12 and on a fresh 3.10 environment (sre_parse branch). --- src/pytest_api_cov/frameworks.py | 161 +++++++++++++++---------------- tests/unit/test_frameworks.py | 20 ++++ 2 files changed, 100 insertions(+), 81 deletions(-) diff --git a/src/pytest_api_cov/frameworks.py b/src/pytest_api_cov/frameworks.py index e088e5a..edcd5b0 100644 --- a/src/pytest_api_cov/frameworks.py +++ b/src/pytest_api_cov/frameworks.py @@ -3,7 +3,6 @@ from __future__ import annotations import importlib -import re import sys from abc import ABC, abstractmethod from itertools import count @@ -11,7 +10,12 @@ if sys.version_info >= (3, 11): from enum import StrEnum + + # The regex parser module was renamed from sre_parse in 3.11; typeshed does not declare it. + from re import _parser as _sre_parser # type: ignore[attr-defined] else: + import sre_parse as _sre_parser + from backports.strenum import StrEnum @@ -24,8 +28,6 @@ class SupportedFramework(StrEnum): if TYPE_CHECKING: - from collections.abc import Callable - from .models import ApiCallRecorder # Auto-added companions of GET et al. that would inflate the endpoint count. @@ -185,96 +187,93 @@ def send(self, *args: Any, **kwargs: Any) -> Any: return TrackingFastAPIClient(self.app) -_REGEX_SHORTHAND_CLASSES = frozenset("dDwWsS") -_QUANTIFIER_PATTERN = re.compile(r"[+*?]|\{\d+(?:,\d*)?\}") - - -def _consume_quantifier(route: str, i: int) -> int: - """Return the index just past a regex quantifier starting at ``i``, if any.""" - match = _QUANTIFIER_PATTERN.match(route, i) - return match.end() if match else i - - -def _skip_class(route: str, i: int) -> int: - """Return the index of the ']' closing the character class opened at ``route[i]``.""" - j = i + 1 - while j < len(route) and route[j] != "]": - j += 2 if route[j] == "\\" else 1 - return j - - -def _group_placeholder(group: str, next_param: Callable[[], str]) -> str: - """Choose a placeholder for a regex group; bodies that can span '/' get a path converter.""" - named = re.match(r"\(\?P<(\w+)>", group) - body = group[named.end() : -1] if named else group[1:-1] - multi_segment = "/" in body or re.search(r"(?" if multi_segment else f"<{name}>" +def _class_matches_slash(items: Any) -> bool: + """Report whether a character class (the ``av`` of an IN node) can match '/'.""" + slash = ord("/") + negated = bool(items) and items[0][0] is _sre_parser.NEGATE + matched = any( + (op is _sre_parser.LITERAL and av == slash) + or (op is _sre_parser.RANGE and av[0] <= slash <= av[1]) + or ( + op is _sre_parser.CATEGORY + and av + in ( + _sre_parser.CATEGORY_NOT_WORD, + _sre_parser.CATEGORY_NOT_DIGIT, + _sre_parser.CATEGORY_NOT_SPACE, + ) + ) + for op, av in items + ) + return not matched if negated else matched + + +def _can_match_slash(nodes: Any) -> bool: + """Report whether this parsed regex subtree can match '/', i.e. span path segments.""" + return any( + op is _sre_parser.ANY + or (op is _sre_parser.LITERAL and av == ord("/")) + or (op is _sre_parser.NOT_LITERAL and av != ord("/")) + or (op is _sre_parser.IN and _class_matches_slash(av)) + or (op is _sre_parser.SUBPATTERN and _can_match_slash(av[3])) + or (op is _sre_parser.BRANCH and any(_can_match_slash(branch) for branch in av[1])) + or (op in (_sre_parser.MAX_REPEAT, _sre_parser.MIN_REPEAT) and _can_match_slash(av[2])) + for op, av in nodes + ) def _django_route_to_template(route: str) -> str: r"""Convert a Django route string to a matchable template. - ``path()`` routes pass through unchanged. In ``re_path()`` regexes, groups - (``(?P[0-9]{4})``), shorthand classes (``\d+``), bare character - classes (``[0-9]+``) and bare dots become placeholders (``path:`` variants - when the pattern can span ``/``); escaped literals (``\.``) are unescaped - and bare quantifiers (a trailing ``/?``) are dropped, so recorded request - paths can match the template. + ``path()`` routes (pure literals) pass through unchanged. ``re_path()`` + regexes are parsed with the stdlib regex parser, and every dynamic + construct — groups (``(?P[0-9]{4})``), classes (``[0-9]+``), + shorthand (``\d+``), dots, alternations — becomes a placeholder, + ```` when it can span ``/``. Escaped literals (``\.``) are + unescaped and an optional trailing ``/?`` keeps its literal, so recorded + request paths can match the template. Unparseable input passes through + verbatim. """ - out: list[str] = [] - i = 0 - n = len(route) + try: + parsed = _sre_parser.parse(route) + except Exception: # noqa: BLE001 - not a regex (e.g. a literal path() route with specials) + return route + + group_names = {number: name for name, number in parsed.state.groupdict.items()} param_counter = count(1) def next_param() -> str: return f"param{next(param_counter)}" - while i < n: - char = route[i] - if char == "\\" and i + 1 < n: - escaped = route[i + 1] - if escaped in _REGEX_SHORTHAND_CLASSES: - out.append(f"<{next_param()}>") - i = _consume_quantifier(route, i + 2) + def placeholder(nodes: Any, name: str | None = None) -> str: + name = name or next_param() + return f"" if _can_match_slash(nodes) else f"<{name}>" + + def emit(nodes: Any) -> str: + out: list[str] = [] + for op, av in nodes: + if op is _sre_parser.LITERAL: + # Escaped literals (\.) arrive pre-unescaped from the parser. + out.append(chr(av)) + elif op is _sre_parser.AT: + continue # anchors (^, $, \b) never appear in request paths + elif op is _sre_parser.SUBPATTERN: + group_number, _add_flags, _del_flags, body = av + out.append(placeholder(body, group_names.get(group_number))) + elif op in (_sre_parser.MAX_REPEAT, _sre_parser.MIN_REPEAT): + _min_count, max_count, body = av + if max_count == 1 and len(body) == 1 and body[0][0] is _sre_parser.LITERAL: + out.append(chr(body[0][1])) # an optional literal ('/?') keeps its literal + elif len(body) == 1 and body[0][0] is _sre_parser.SUBPATTERN: + out.append(emit(body)) # '(...)?' is just the group placeholder + else: + out.append(placeholder(body)) # \d+, [0-9]+, .*, a{2,4}, ... else: - out.append(escaped) - i += 2 - elif char == "(": - depth = 0 - j = i - while j < n: - inner = route[j] - if inner == "\\": - j += 2 - continue - if inner == "[": - j = _skip_class(route, j) - elif inner == "(": - depth += 1 - elif inner == ")": - depth -= 1 - if depth == 0: - break - j += 1 - out.append(_group_placeholder(route[i : j + 1], next_param)) - i = _consume_quantifier(route, j + 1) - elif char == "[": - j = _skip_class(route, i) - out.append(f"<{next_param()}>") - i = _consume_quantifier(route, j + 1) - elif char == ".": - end = _consume_quantifier(route, i + 1) - # A quantified dot (.* / .+) can cross path segments. - out.append(f"" if end > i + 1 else f"<{next_param()}>") - i = end - elif char in "+*?": - # Bare quantifier on the preceding literal (e.g. a trailing '/?'): drop it. - i += 1 - else: - out.append(char) - i += 1 - return "".join(out) + # IN, ANY, BRANCH, NOT_LITERAL — and any opcode a future Python adds. + out.append(placeholder([(op, av)])) + return "".join(out) + + return emit(parsed) class DjangoAdapter(BaseAdapter): diff --git a/tests/unit/test_frameworks.py b/tests/unit/test_frameworks.py index e636e9f..c578e97 100644 --- a/tests/unit/test_frameworks.py +++ b/tests/unit/test_frameworks.py @@ -254,6 +254,26 @@ def test_path_route_passes_through(self): assert _django_route_to_template("articles//") == "articles//" + def test_negated_class_containing_slash_stays_single_segment(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template("x/(?P[^/]+)/") == "x//" + + def test_top_level_alternation_becomes_placeholder(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template("legacy|new") == "" + + def test_bounded_repeat_becomes_placeholder(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template("a{2,4}/end/") == "/end/" + + def test_unparseable_route_passes_through(self): + from pytest_api_cov.frameworks import _django_route_to_template + + assert _django_route_to_template("bad[route") == "bad[route" + def test_shorthand_class_outside_group_becomes_placeholder(self): from pytest_api_cov.frameworks import _django_route_to_template