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..d302e81 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,17 +1,37 @@ [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" }] 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..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") @@ -60,7 +62,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 +73,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 +86,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..edcd5b0 100644 --- a/src/pytest_api_cov/frameworks.py +++ b/src/pytest_api_cov/frameworks.py @@ -2,13 +2,20 @@ from __future__ import annotations +import importlib import sys from abc import ABC, abstractmethod +from itertools import count from typing import TYPE_CHECKING, Any 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 @@ -23,6 +30,9 @@ class SupportedFramework(StrEnum): if TYPE_CHECKING: 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.""" @@ -43,42 +53,68 @@ 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-file rules without dropping user routes sharing the name.""" + endpoint = str(getattr(rule, "endpoint", "")) + 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.""" - 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") + if method not in _SKIPPED_METHODS ] return sorted(endpoints) 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() + active_recorder = recorder url_adapter = None 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: + active_recorder.record_call(rule.rule, test_name, method) return super().open(*args, **kwargs) return TrackingFlaskClient(self.app, self.app.response_class) @@ -96,12 +132,15 @@ 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): + # 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 route.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 @@ -122,18 +161,121 @@ 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] - 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) + method = request.method.upper() + original_path = request.url.path + try: + response = super().send(*args, **kwargs) + except BaseException: + active_recorder.record_call(original_path, test_name, method) + raise + # 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) +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 (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. + """ + 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)}" + + 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: + # 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): """Adapter for Django applications.""" @@ -147,19 +289,26 @@ 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. + implemented = {m.upper() for m in view_class.http_method_names if hasattr(view_class, m)} + 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 = 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) @@ -172,13 +321,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) @@ -195,8 +345,45 @@ def _unwrap_wsgi_app(app: Any) -> Any: return None +_FRAMEWORK_CLASS_SPECS: tuple[tuple[SupportedFramework, str, str], ...] = ( + (SupportedFramework.FLASK, "flask", "Flask"), + (SupportedFramework.FASTAPI, "fastapi", "FastAPI"), + (SupportedFramework.DJANGO, "django.core.handlers.base", "BaseHandler"), +) + +_import_failed: set[str] = set() + + +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_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 + + 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 duck-typed apps (e.g. mocks in test suites). app_type = type(app).__name__ module_name = getattr(type(app), "__module__", "").split(".")[0] @@ -205,7 +392,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 @@ -213,8 +400,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 e8f20b6..9c7ec96 100644 --- a/src/pytest_api_cov/openapi.py +++ b/src/pytest_api_cov/openapi.py @@ -28,11 +28,22 @@ 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): + 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..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,18 +87,20 @@ 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 + + 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 @@ -278,6 +289,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.""" @@ -288,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")) @@ -296,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 @@ -395,8 +399,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 +416,7 @@ 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", []) + 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/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..8db923c 100644 --- a/src/pytest_api_cov/report.py +++ b/src/pytest_api_cov/report.py @@ -15,12 +15,26 @@ 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.""" - 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: @@ -28,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() @@ -46,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 + + +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 + - 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 _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( @@ -70,58 +130,23 @@ def categorise_endpoints( HTTP method prefixes. Pattern order matters: exclusions first, then negations override them. """ - covered: list[str] = [] - uncovered: list[str] = [] - excluded: list[str] = [] + kept, excluded = _partition_excluded(endpoints, exclusion_patterns) + covered, uncovered = _match_covered(kept, called_data) + return covered, uncovered, excluded - 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) - 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(dict.fromkeys(_split_endpoint(endpoint)[1] for endpoint in endpoints)) + + grouped_calls: dict[str, set[str]] = {} + for key, callers in called_data.items(): + grouped_calls.setdefault(_split_endpoint(key)[1], set()).update(callers) + + return grouped_endpoints, grouped_calls def print_endpoints( @@ -182,17 +207,24 @@ def generate_pytest_api_cov_report( console = Console() if not discovered_endpoints: + # 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]" + ) + 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]") + header = f"{'=' * 20} API Coverage Report {'=' * 20}" + console.print(f"\n\n[bold blue]{header}[/bold blue]") - covered, uncovered, excluded = categorise_endpoints( - discovered_endpoints, - called_data, - api_cov_config.exclusion_patterns, - ) + kept, excluded = _partition_excluded(discovered_endpoints, api_cov_config.exclusion_patterns) + if api_cov_config.group_methods_by_endpoint: + # 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( @@ -226,6 +258,21 @@ 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. 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. " @@ -253,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 diff --git a/tests/integration/test_django_integration.py b/tests/integration/test_django_integration.py index 4bb5d37..8f8a00d 100644 --- a/tests/integration/test_django_integration.py +++ b/tests/integration/test_django_integration.py @@ -60,3 +60,71 @@ 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}) + + 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), + ] + """ + ) + + 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 + # 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 5deb869..5ae09e6 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,241 @@ 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 + # Both the requested path and the redirect target get credit. + assert "GET /items/" 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_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..c578e97 100644 --- a/tests/unit/test_frameworks.py +++ b/tests/unit/test_frameworks.py @@ -219,3 +219,101 @@ 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//" + + 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 + + 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_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..7edca76 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,90 @@ 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): + """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 diff --git a/uv.lock b/uv.lock index 017f07e..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'" }, @@ -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" },