cleanup, format

Author: BarnabasG

src/pytest_api_cov/cli.py

@@ -1,4 +1,4 @@
-"""pytest-api-cov CLI commands for setup and configuration."""
+"""CLI commands for setup and configuration."""
 
 import argparse
 import os
@@ -22,19 +22,16 @@ def detect_framework_and_app() -> Optional[tuple[str, str, str]]:
     for filename, attr_names in common_patterns:
         if os.path.exists(filename):
             try:
-                # Read file to detect framework
                 with open(filename, "r") as f:
                     content = f.read()
 
-                # Simple detection based on imports
                 if "from fastapi import" in content or "import fastapi" in content:
                     framework = "FastAPI"
                 elif "from flask import" in content or "import flask" in content:
                     framework = "Flask"
                 else:
                     continue
 
-                # Look for app variable in file
                 for attr_name in attr_names:
                     if f"{attr_name} = " in content:
                         return framework, filename, attr_name
@@ -97,14 +94,12 @@ def cmd_init():
     print("🚀 pytest-api-cov Setup Wizard")
     print("=" * 40)
 
-    # Detect existing setup
     detection_result = detect_framework_and_app()
 
     if detection_result:
         framework, file_path, app_variable = detection_result
         print(f"✅ Detected {framework} app in {file_path} (variable: {app_variable})")
 
-        # Check if conftest.py already exists
         conftest_exists = os.path.exists("conftest.py")
         if conftest_exists:
             print("⚠️  conftest.py already exists")
@@ -118,7 +113,6 @@ def cmd_init():
                 f.write(conftest_content)
             print("✅ Created conftest.py")
 
-        # Check if pyproject.toml exists
         pyproject_exists = os.path.exists("pyproject.toml")
         if pyproject_exists:
             print("ℹ️  pyproject.toml already exists")
@@ -183,7 +177,6 @@ def main():
 
     subparsers = parser.add_subparsers(dest="command", help="Available commands")
 
-    # Init command
     subparsers.add_parser("init", help="Initialize pytest-api-cov setup")
 
     args = parser.parse_args()

src/pytest_api_cov/config.py

@@ -1,4 +1,4 @@
-"""src/pytest_api_cov/config.py: Handles the configuration for the API coverage report."""
+"""Configuration handling for the API coverage report."""
 
 import sys
 from typing import Any, Dict, List, Optional
@@ -8,7 +8,7 @@
 
 
 class ApiCoverageReportConfig(BaseModel):
-    """A Pydantic model for all API coverage configuration."""
+    """Configuration model for API coverage reporting."""
 
     model_config = ConfigDict(populate_by_name=True)
 
@@ -33,7 +33,7 @@ def read_toml_config() -> Dict[str, Any]:
 
 
 def read_session_config(session_config: Any) -> Dict[str, Any]:
-    """Read configuration from the pytest session config (command-line flags)."""
+    """Read configuration from pytest session config (command-line flags)."""
     cli_options = {
         "api-cov-fail-under": "fail_under",
         "api-cov-show-uncovered-endpoints": "show_uncovered_endpoints",
@@ -47,7 +47,6 @@ def read_session_config(session_config: Any) -> Dict[str, Any]:
     config = {}
     for opt, key in cli_options.items():
         value = session_config.getoption(f"--{opt}")
-        # We only want to include options that were actually provided or are booleans
         if value is not None and value != [] and value is not False:
             config[key] = value
     return config
@@ -68,10 +67,8 @@ def get_pytest_api_cov_report_config(session_config: Any) -> ApiCoverageReportCo
     toml_config = read_toml_config()
     cli_config = read_session_config(session_config)
 
-    # Start with defaults, then apply TOML, then apply CLI args
     final_config = {**toml_config, **cli_config}
 
-    # Set force_sugar based on environment if not explicitly set
     if final_config.get("force_sugar_disabled"):
         final_config["force_sugar"] = False
     elif "force_sugar" not in final_config:

src/pytest_api_cov/frameworks.py

@@ -1,12 +1,11 @@
-"""src/pytest_api_cov/frameworks.py"""
+"""Framework adapters for Flask and FastAPI."""
 
 from typing import TYPE_CHECKING, Any, List, Optional
 
 if TYPE_CHECKING:
     from .models import ApiCallRecorder
 
 
-# --- Base Adapter ---
 class BaseAdapter:
     def __init__(self, app: Any):
         self.app = app
@@ -20,10 +19,8 @@ def get_tracked_client(self, recorder: Optional["ApiCallRecorder"], test_name: s
         raise NotImplementedError
 
 
-# --- Flask Adapter ---
 class FlaskAdapter(BaseAdapter):
     def get_endpoints(self) -> List[str]:
-        # Exclude static and other non-API endpoints
         excluded_rules = ("/static/<path:filename>",)
         return sorted([rule.rule for rule in self.app.url_map.iter_rules() if rule.rule not in excluded_rules])
 
@@ -38,20 +35,16 @@ def open(self, *args, **kwargs) -> Any:
                 path = kwargs.get("path") or (args[0] if args else None)
                 if path and hasattr(self.application.url_map, "bind"):
                     try:
-                        # Fetch the endpoint *name*, e.g., 'root' not '/'
                         endpoint_name, _ = self.application.url_map.bind("").match(path, method=kwargs.get("method"))
-                        # Find the rule object associated with that endpoint name
                         endpoint_rule_string = next(self.application.url_map.iter_rules(endpoint_name)).rule
                         recorder.record_call(endpoint_rule_string, test_name)  # type: ignore[union-attr]
                     except Exception:
-                        # Fallback for paths that might not match a rule
                         pass
                 return super().open(*args, **kwargs)
 
         return TrackingFlaskClient(self.app, self.app.response_class)
 
 
-# --- FastAPI Adapter ---
 class FastAPIAdapter(BaseAdapter):
     def get_endpoints(self) -> List[str]:
         from fastapi.routing import APIRoute
@@ -64,17 +57,15 @@ def get_tracked_client(self, recorder: Optional["ApiCallRecorder"], test_name: s
         if recorder is None:
             return TestClient(self.app)
 
-        # FastAPI patches the 'send' method of the underlying client
         class TrackingFastAPIClient(TestClient):
             def send(self, *args, **kwargs) -> Any:
                 request = args[0]
-                recorder.record_call(request.url.path, test_name)  # type: ignore[union-attr]
+                recorder.record_call(request.url.path, test_name)
                 return super().send(*args, **kwargs)
 
         return TrackingFastAPIClient(self.app)
 
 
-# --- Factory Function ---
 def get_framework_adapter(app: Any) -> BaseAdapter:
     """Detects the framework and returns the appropriate adapter."""
     app_type = type(app).__name__

src/pytest_api_cov/models.py

@@ -1,6 +1,5 @@
 """Data models for pytest-api-cov."""
 
-from collections import defaultdict
 from typing import Any, Dict, Iterator, List, Set, Tuple
 
 from pydantic import BaseModel, Field
@@ -11,7 +10,6 @@ class ApiCallRecorder(BaseModel):
 
     model_config = {"arbitrary_types_allowed": True}
 
-    # Map of endpoint paths to sets of test function names that called them
     calls: Dict[str, Set[str]] = Field(default_factory=dict)
 
     def record_call(self, endpoint: str, test_name: str) -> None:
@@ -70,7 +68,7 @@ class EndpointDiscovery(BaseModel):
     """Model for discovered API endpoints."""
 
     endpoints: List[str] = Field(default_factory=list)
-    discovery_source: str = Field(default="unknown")  # e.g., "flask_adapter", "fastapi_adapter"
+    discovery_source: str = Field(default="unknown")
 
     def add_endpoint(self, endpoint: str) -> None:
         """Add a discovered endpoint."""
@@ -86,7 +84,7 @@ def __len__(self) -> int:
         """Return number of discovered endpoints."""
         return len(self.endpoints)
 
-    def __iter__(self) -> Iterator[str]:
+    def __iter__(self) -> Iterator[str]:  # type: ignore[override]
         """Iterate over discovered endpoints."""
         return iter(self.endpoints)
 
@@ -109,20 +107,16 @@ def add_discovered_endpoint(self, endpoint: str, source: str = "unknown") -> Non
 
     def merge_worker_data(self, worker_recorder: Dict[str, Any], worker_endpoints: List[str]) -> None:
         """Merge data from a worker process."""
-        # Handle recorder data
         if isinstance(worker_recorder, dict):
-            # Check if all values are lists (serializable format)
             all_lists = worker_recorder and all(isinstance(v, list) for v in worker_recorder.values())
             if all_lists:
                 worker_api_recorder = ApiCallRecorder.from_serializable(worker_recorder)
             else:
-                # Handle raw dict format or mixed types
                 calls = {k: set(v) if isinstance(v, (list, set)) else {v} for k, v in worker_recorder.items()}
                 worker_api_recorder = ApiCallRecorder(calls=calls)
 
             self.recorder.merge(worker_api_recorder)
 
-        # Handle discovered endpoints
         if worker_endpoints:
             worker_discovery = EndpointDiscovery(endpoints=worker_endpoints, discovery_source="worker")
             self.discovered_endpoints.merge(worker_discovery)

src/pytest_api_cov/plugin.py

@@ -1,4 +1,4 @@
-"""src/pytest_api_cov/plugin.py"""
+"""pytest plugin for API coverage tracking."""
 
 import importlib
 import importlib.util
@@ -32,7 +32,6 @@ def auto_discover_app() -> Optional[Any]:
     """Automatically discover Flask/FastAPI apps in common locations."""
     logger.debug("> Auto-discovering app in common locations...")
 
-    # Common file patterns and variable names to check
     common_patterns = [
         ("app.py", ["app", "application", "main"]),
         ("main.py", ["app", "application", "main"]),
@@ -45,14 +44,12 @@ def auto_discover_app() -> Optional[Any]:
         if os.path.exists(filename):
             logger.debug(f"> Found {filename}, checking for app variables...")
             try:
-                # Import the module
-                module_name = filename[:-3]  # Remove .py extension
+                module_name = filename[:-3]  # .py extension
                 spec = importlib.util.spec_from_file_location(module_name, filename)
                 if spec and spec.loader:
                     module = importlib.util.module_from_spec(spec)
                     spec.loader.exec_module(module)
 
-                    # Check each possible app variable name
                     for attr_name in attr_names:
                         if hasattr(module, attr_name):
                             app = getattr(module, attr_name)
@@ -112,22 +109,18 @@ def pytest_addoption(parser: pytest.Parser) -> None:
 
 def pytest_configure(config: pytest.Config) -> None:
     """Configure the pytest session and logging."""
-    # Configure logging based on verbosity level
     if config.getoption("--api-cov-report"):
         verbosity = config.option.verbose
 
-        # Set up logging level based on pytest verbosity
         if verbosity >= 2:  # -vv or more
             log_level = logging.DEBUG
         elif verbosity >= 1:  # -v
             log_level = logging.INFO
-        else:  # normal run
+        else:
             log_level = logging.WARNING
 
-        # Configure the logger
         logger.setLevel(log_level)
 
-        # Only add handler if we don't already have one
         if not logger.handlers:
             handler = logging.StreamHandler()
             handler.setLevel(log_level)
@@ -137,7 +130,6 @@ def pytest_configure(config: pytest.Config) -> None:
 
         logger.info("Initializing API coverage plugin...")
 
-    # Register xdist plugin if available
     if config.pluginmanager.hasplugin("xdist"):
         config.pluginmanager.register(DeferXdistPlugin(), "defer_xdist_plugin")
 
@@ -157,11 +149,9 @@ def client(request: pytest.FixtureRequest) -> Any:
     """
     session = request.node.session
 
-    # Only proceed if API coverage is enabled
     if not session.config.getoption("--api-cov-report"):
         pytest.skip("API coverage not enabled. Use --api-cov-report flag.")
 
-    # Try to get app from existing fixture first
     app = None
     try:
         app = request.getfixturevalue("app")
@@ -170,13 +160,11 @@ def client(request: pytest.FixtureRequest) -> Any:
         logger.debug("> No 'app' fixture found, trying auto-discovery...")
         app = auto_discover_app()
 
-    # If still no app found, show helpful error
     if app is None:
         helpful_msg = get_helpful_error_message()
         print(helpful_msg)
         pytest.skip("No API app found. See error message above for setup guidance.")
 
-    # Validate the app is supported
     if not is_supported_framework(app):
         pytest.skip(f"Unsupported framework: {type(app).__name__}. pytest-api-coverage supports Flask and FastAPI.")
 
@@ -189,7 +177,6 @@ def client(request: pytest.FixtureRequest) -> Any:
     if coverage_data is None:
         pytest.skip("API coverage data not initialized. This should not happen.")
 
-    # Discover endpoints on the first run of this fixture.
     if not coverage_data.discovered_endpoints.endpoints:
         try:
             endpoints = adapter.get_endpoints()
@@ -215,15 +202,13 @@ def pytest_sessionfinish(session: pytest.Session) -> None:
 
         logger.debug(f"> pytest-api-coverage: Generating report for {len(coverage_data.recorder)} recorded endpoints.")
         if hasattr(session.config, "workeroutput"):
-            # Send data to master process in serializable format
             serializable_recorder = coverage_data.recorder.to_serializable()
             session.config.workeroutput["api_call_recorder"] = serializable_recorder
             session.config.workeroutput["discovered_endpoints"] = coverage_data.discovered_endpoints.endpoints
             logger.debug("> Sent API call data and discovered endpoints to master process")
         else:
             logger.debug("> No workeroutput found, generating report for master data.")
 
-            # Get worker data from config if available
             worker_recorder_data = getattr(session.config, "worker_api_call_recorder", {})
             worker_endpoints = getattr(session.config, "worker_discovered_endpoints", [])
 
@@ -247,7 +232,6 @@ def pytest_sessionfinish(session: pytest.Session) -> None:
         if hasattr(session, "api_coverage_data"):
             delattr(session, "api_coverage_data")
 
-        # Clear any module-level caches
         if hasattr(session.config, "worker_api_call_recorder"):
             delattr(session.config, "worker_api_call_recorder")
 
@@ -281,7 +265,6 @@ def pytest_testnodedown(self, node: Any) -> None:
             node.config.worker_api_call_recorder = current
             logger.debug(f"> Updated current data: {current}")
 
-        # Merge discovered endpoints (take the first non-empty list we get)
         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}")

src/pytest_api_cov/pytest_flags.py

@@ -1,4 +1,4 @@
-"""src/pytest_api_cov/pytest_flags.py"""
+"""pytest flag configuration for API coverage."""
 
 from typing import TYPE_CHECKING