Api Test Suite Builder

#!/usr/bin/env python3
"""Validate API response samples against JSON Schema contracts.

Takes an OpenAPI/Swagger spec and a directory of response sample JSON files,
then validates each sample against its corresponding schema definition.
Reports schema violations, missing required fields, type mismatches, and
extra properties (if additionalProperties is false).

Usage:
    python contract_validator.py spec.json samples/
    python contract_validator.py spec.json samples/ --strict
    python contract_validator.py spec.json samples/ --json
"""

import argparse
import json
import os
import re
import sys
from pathlib import Path


# ---------- JSON Schema Validator (standard-library only) ----------

class SchemaValidationError:
    """A single validation error."""
    def __init__(self, path: str, message: str, schema_path: str = ""):
        self.path = path
        self.message = message
        self.schema_path = schema_path

    def to_dict(self) -> dict:
        return {"path": self.path, "message": self.message, "schema_path": self.schema_path}

    def __repr__(self):
        return f"{self.path}: {self.message}"


def validate_value(value, schema: dict, root_schema: dict, path: str = "$",
                   strict: bool = False) -> list:
    """Validate a value against a JSON Schema. Returns list of SchemaValidationError."""
    errors = []

    if not schema:
        return errors

    # Resolve $ref
    if "$ref" in schema:
        schema = _resolve_ref(root_schema, schema["$ref"])
        if schema is None:
            errors.append(SchemaValidationError(path, "unresolvable $ref"))
            return errors

    # Handle allOf
    if "allOf" in schema:
        for sub in schema["allOf"]:
            errors.extend(validate_value(value, sub, root_schema, path, strict))
        return errors

    # Handle oneOf
    if "oneOf" in schema:
        matches = 0
        for sub in schema["oneOf"]:
            sub_errors = validate_value(value, sub, root_schema, path, strict)
            if not sub_errors:
                matches += 1
        if matches == 0:
            errors.append(SchemaValidationError(path, "does not match any oneOf schema"))
        elif matches > 1:
            errors.append(SchemaValidationError(path, f"matches {matches} oneOf schemas, expected exactly 1"))
        return errors

    # Handle anyOf
    if "anyOf" in schema:
        for sub in schema["anyOf"]:
            sub_errors = validate_value(value, sub, root_schema, path, strict)
            if not sub_errors:
                return []
        errors.append(SchemaValidationError(path, "does not match any anyOf schema"))
        return errors

    # Nullable
    nullable = schema.get("nullable", False)
    if value is None:
        if not nullable and "type" in schema:
            errors.append(SchemaValidationError(path, f"expected {schema['type']}, got null"))
        return errors

    # Type checking
    expected_type = schema.get("type")
    if expected_type:
        type_ok = _check_type(value, expected_type)
        if not type_ok:
            actual = type(value).__name__
            errors.append(SchemaValidationError(path, f"expected type '{expected_type}', got '{actual}'"))
            return errors  # No point checking further if type is wrong

    # Enum
    if "enum" in schema:
        if value not in schema["enum"]:
            errors.append(SchemaValidationError(
                path, f"value '{value}' not in enum {schema['enum']}"
            ))

    # String constraints
    if expected_type == "string" and isinstance(value, str):
        if "minLength" in schema and len(value) < schema["minLength"]:
            errors.append(SchemaValidationError(
                path, f"string length {len(value)} below minLength {schema['minLength']}"
            ))
        if "maxLength" in schema and len(value) > schema["maxLength"]:
            errors.append(SchemaValidationError(
                path, f"string length {len(value)} above maxLength {schema['maxLength']}"
            ))
        if "pattern" in schema:
            if not re.search(schema["pattern"], value):
                errors.append(SchemaValidationError(
                    path, f"string does not match pattern '{schema['pattern']}'"
                ))
        if "format" in schema:
            fmt_error = _check_format(value, schema["format"])
            if fmt_error:
                errors.append(SchemaValidationError(path, fmt_error))

    # Number constraints
    if expected_type in ("integer", "number") and isinstance(value, (int, float)):
        if "minimum" in schema and value < schema["minimum"]:
            errors.append(SchemaValidationError(
                path, f"value {value} below minimum {schema['minimum']}"
            ))
        if "maximum" in schema and value > schema["maximum"]:
            errors.append(SchemaValidationError(
                path, f"value {value} above maximum {schema['maximum']}"
            ))
        if "exclusiveMinimum" in schema and value <= schema["exclusiveMinimum"]:
            errors.append(SchemaValidationError(
                path, f"value {value} not above exclusiveMinimum {schema['exclusiveMinimum']}"
            ))
        if "exclusiveMaximum" in schema and value >= schema["exclusiveMaximum"]:
            errors.append(SchemaValidationError(
                path, f"value {value} not below exclusiveMaximum {schema['exclusiveMaximum']}"
            ))

    # Object validation
    if expected_type == "object" and isinstance(value, dict):
        properties = schema.get("properties", {})
        required = schema.get("required", [])

        # Check required fields
        for req_field in required:
            if req_field not in value:
                errors.append(SchemaValidationError(
                    f"{path}.{req_field}", f"required field '{req_field}' is missing"
                ))

        # Validate known properties
        for prop_name, prop_schema in properties.items():
            if prop_name in value:
                errors.extend(validate_value(
                    value[prop_name], prop_schema, root_schema,
                    f"{path}.{prop_name}", strict
                ))

        # Check additional properties
        additional = schema.get("additionalProperties", True)
        if strict or additional is False:
            known_props = set(properties.keys())
            extra = set(value.keys()) - known_props
            if extra and additional is False:
                for prop in sorted(extra):
                    errors.append(SchemaValidationError(
                        f"{path}.{prop}", f"additional property '{prop}' not allowed"
                    ))
            elif extra and strict and additional is not True:
                for prop in sorted(extra):
                    errors.append(SchemaValidationError(
                        f"{path}.{prop}", f"unexpected property '{prop}' (strict mode)"
                    ))

        # Validate patternProperties
        pattern_props = schema.get("patternProperties", {})
        for pattern, p_schema in pattern_props.items():
            regex = re.compile(pattern)
            for key in value:
                if regex.search(key):
                    errors.extend(validate_value(
                        value[key], p_schema, root_schema,
                        f"{path}.{key}", strict
                    ))

    # Array validation
    if expected_type == "array" and isinstance(value, list):
        items_schema = schema.get("items", {})
        if "minItems" in schema and len(value) < schema["minItems"]:
            errors.append(SchemaValidationError(
                path, f"array length {len(value)} below minItems {schema['minItems']}"
            ))
        if "maxItems" in schema and len(value) > schema["maxItems"]:
            errors.append(SchemaValidationError(
                path, f"array length {len(value)} above maxItems {schema['maxItems']}"
            ))
        if "uniqueItems" in schema and schema["uniqueItems"]:
            seen = []
            for item in value:
                serialized = json.dumps(item, sort_keys=True)
                if serialized in seen:
                    errors.append(SchemaValidationError(path, "array contains duplicate items"))
                    break
                seen.append(serialized)

        for i, item in enumerate(value):
            errors.extend(validate_value(
                item, items_schema, root_schema,
                f"{path}[{i}]", strict
            ))

    return errors


def _check_type(value, expected: str) -> bool:
    """Check if value matches the expected JSON Schema type."""
    if expected == "string":
        return isinstance(value, str)
    elif expected == "integer":
        return isinstance(value, int) and not isinstance(value, bool)
    elif expected == "number":
        return isinstance(value, (int, float)) and not isinstance(value, bool)
    elif expected == "boolean":
        return isinstance(value, bool)
    elif expected == "array":
        return isinstance(value, list)
    elif expected == "object":
        return isinstance(value, dict)
    elif expected == "null":
        return value is None
    return True


def _check_format(value: str, fmt: str) -> str:
    """Basic format validation for common formats. Returns error message or empty string."""
    if fmt == "email":
        if "@" not in value or "." not in value.split("@")[-1]:
            return f"'{value}' is not a valid email format"
    elif fmt == "uri" or fmt == "url":
        if not re.match(r'^https?://', value):
            return f"'{value}' is not a valid URI format"
    elif fmt == "uuid":
        if not re.match(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', value, re.I):
            return f"'{value}' is not a valid UUID format"
    elif fmt in ("date-time", "datetime"):
        if not re.match(r'^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}', value):
            return f"'{value}' is not a valid date-time format"
    elif fmt == "date":
        if not re.match(r'^\d{4}-\d{2}-\d{2}$', value):
            return f"'{value}' is not a valid date format"
    return ""


def _resolve_ref(root_schema: dict, ref: str) -> dict:
    """Resolve a $ref path within the root schema."""
    if not ref.startswith("#/"):
        return None
    parts = ref[2:].split("/")
    current = root_schema
    for part in parts:
        part = part.replace("~1", "/").replace("~0", "~")
        if isinstance(current, dict) and part in current:
            current = current[part]
        else:
            return None
    return current


# ---------- Spec + Sample Loading ----------

def load_spec(spec_path: str) -> dict:
    """Load and validate an OpenAPI/Swagger spec file."""
    path = Path(spec_path)
    if not path.exists():
        print(f"Error: spec file not found: {spec_path}", file=sys.stderr)
        sys.exit(1)
    try:
        with open(path, "r", encoding="utf-8") as f:
            spec = json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error: invalid JSON in spec file: {e}", file=sys.stderr)
        sys.exit(1)
    if "openapi" not in spec and "swagger" not in spec:
        print("Error: not an OpenAPI/Swagger spec", file=sys.stderr)
        sys.exit(1)
    return spec


def extract_response_schemas(spec: dict) -> dict:
    """Extract response schemas keyed by 'METHOD /path STATUS'.

    Returns dict like:
        {"GET /users 200": <schema>, "POST /users 201": <schema>}
    """
    schemas = {}
    for path, path_item in spec.get("paths", {}).items():
        for method in ("get", "post", "put", "patch", "delete"):
            if method not in path_item:
                continue
            operation = path_item[method]
            for status, resp in operation.get("responses", {}).items():
                content = resp.get("content", {})
                json_content = content.get("application/json", {})
                schema = json_content.get("schema")
                if schema:
                    key = f"{method.upper()} {path} {status}"
                    schemas[key] = schema
    return schemas


def load_samples(samples_dir: str) -> list:
    """Load JSON sample files from the samples directory.

    Expected naming convention:
        METHOD_path_STATUS.json  (e.g., GET_users_200.json)
        or any .json file with a _meta key inside specifying method/path/status
    """
    samples = []
    samples_path = Path(samples_dir)
    if not samples_path.exists():
        print(f"Error: samples directory not found: {samples_dir}", file=sys.stderr)
        sys.exit(1)

    for json_file in sorted(samples_path.rglob("*.json")):
        try:
            with open(json_file, "r", encoding="utf-8") as f:
                data = json.load(f)
        except json.JSONDecodeError as e:
            samples.append({
                "file": str(json_file),
                "error": f"invalid JSON: {e}",
                "data": None,
                "meta": None,
            })
            continue

        # Try to extract meta from file content
        meta = None
        body = data
        if isinstance(data, dict) and "_meta" in data:
            meta = data.pop("_meta")
            body = data

        # Try to infer from filename: GET_api_v1_users_200.json
        if meta is None:
            meta = _infer_meta_from_filename(json_file.stem)

        samples.append({
            "file": str(json_file),
            "data": body,
            "meta": meta,
            "error": None,
        })

    return samples


def _infer_meta_from_filename(stem: str) -> dict:
    """Try to parse METHOD_path_STATUS from filename stem."""
    # Pattern: GET_api_v1_users_200
    match = re.match(r'^(GET|POST|PUT|PATCH|DELETE)_(.+)_(\d{3})$', stem, re.IGNORECASE)
    if match:
        method = match.group(1).upper()
        path_raw = match.group(2)
        status = match.group(3)
        # Convert underscores back to slashes
        path = "/" + path_raw.replace("_", "/")
        return {"method": method, "path": path, "status": status}

    # Fallback: try just METHOD_path
    match = re.match(r'^(GET|POST|PUT|PATCH|DELETE)_(.+)$', stem, re.IGNORECASE)
    if match:
        method = match.group(1).upper()
        path_raw = match.group(2)
        path = "/" + path_raw.replace("_", "/")
        return {"method": method, "path": path, "status": "200"}

    return None


def validate_samples(spec: dict, samples: list, strict: bool = False) -> list:
    """Validate each sample against its matching response schema."""
    response_schemas = extract_response_schemas(spec)
    results = []

    for sample in samples:
        result = {
            "file": sample["file"],
            "status": "skip",
            "errors": [],
            "meta": sample["meta"],
        }

        if sample["error"]:
            result["status"] = "error"
            result["errors"] = [{"path": "$", "message": sample["error"]}]
            results.append(result)
            continue

        if sample["meta"] is None:
            result["status"] = "skip"
            result["errors"] = [{"path": "$", "message": "could not determine endpoint from filename or _meta"}]
            results.append(result)
            continue

        meta = sample["meta"]
        schema_key = f"{meta['method']} {meta['path']} {meta['status']}"

        # Try exact match first, then try path parameter variants
        schema = response_schemas.get(schema_key)
        if schema is None:
            # Try matching with path params
            for key, s in response_schemas.items():
                key_parts = key.split(" ")
                if len(key_parts) == 3:
                    k_method, k_path, k_status = key_parts
                    if k_method == meta["method"] and k_status == meta["status"]:
                        if _paths_match(k_path, meta["path"]):
                            schema = s
                            break

        if schema is None:
            result["status"] = "skip"
            result["errors"] = [{"path": "$", "message": f"no schema found for {schema_key}"}]
            results.append(result)
            continue

        errors = validate_value(sample["data"], schema, spec, "$", strict)
        result["status"] = "pass" if not errors else "fail"
        result["errors"] = [e.to_dict() for e in errors]
        results.append(result)

    return results


def _paths_match(spec_path: str, sample_path: str) -> bool:
    """Check if a spec path (with {params}) matches a concrete sample path."""
    spec_parts = spec_path.strip("/").split("/")
    sample_parts = sample_path.strip("/").split("/")
    if len(spec_parts) != len(sample_parts):
        return False
    for sp, sa in zip(spec_parts, sample_parts):
        if sp.startswith("{") and sp.endswith("}"):
            continue
        if sp != sa:
            return False
    return True


def main():
    parser = argparse.ArgumentParser(
        description="Validate API response samples against OpenAPI schema contracts.",
        epilog="Example: python contract_validator.py openapi.json samples/ --strict",
    )
    parser.add_argument("spec", help="Path to OpenAPI/Swagger JSON spec file")
    parser.add_argument("samples_dir", help="Directory containing JSON response samples")
    parser.add_argument(
        "--strict",
        action="store_true",
        help="Enable strict mode: reject unexpected properties even if additionalProperties is not false",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON",
    )
    args = parser.parse_args()

    spec = load_spec(args.spec)
    samples = load_samples(args.samples_dir)

    if not samples:
        print("Warning: no JSON sample files found.", file=sys.stderr)
        sys.exit(1)

    results = validate_samples(spec, samples, strict=args.strict)

    passed = sum(1 for r in results if r["status"] == "pass")
    failed = sum(1 for r in results if r["status"] == "fail")
    errored = sum(1 for r in results if r["status"] == "error")
    skipped = sum(1 for r in results if r["status"] == "skip")

    if args.json_output:
        output = {
            "summary": {
                "total": len(results),
                "passed": passed,
                "failed": failed,
                "errored": errored,
                "skipped": skipped,
            },
            "strict_mode": args.strict,
            "results": results,
        }
        print(json.dumps(output, indent=2))
    else:
        print("=== Contract Validation Report ===")
        print(f"Strict mode: {'ON' if args.strict else 'OFF'}")
        print()
        print(f"Total samples: {len(results)}")
        print(f"  Passed:  {passed}")
        print(f"  Failed:  {failed}")
        print(f"  Errors:  {errored}")
        print(f"  Skipped: {skipped}")
        print()

        # Show failures and errors
        for r in results:
            if r["status"] == "pass":
                symbol = "[PASS]"
            elif r["status"] == "fail":
                symbol = "[FAIL]"
            elif r["status"] == "error":
                symbol = "[ERR ]"
            else:
                symbol = "[SKIP]"

            filename = Path(r["file"]).name
            meta_str = ""
            if r["meta"]:
                meta_str = f" ({r['meta']['method']} {r['meta']['path']} {r['meta']['status']})"

            print(f"  {symbol} {filename}{meta_str}")

            if r["status"] in ("fail", "error"):
                for err in r["errors"]:
                    print(f"         {err['path']}: {err['message']}")

        print()

    if failed > 0 or errored > 0:
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Analyze API test coverage by comparing spec endpoints vs existing test files.

Parses an OpenAPI/Swagger JSON spec to extract all defined endpoints, then
scans test files in a given directory to detect which endpoints have test
coverage. Reports missing coverage, coverage percentage, and gaps by
HTTP method and tag.

Usage:
    python coverage_analyzer.py spec.json tests/
    python coverage_analyzer.py spec.json tests/ --json
    python coverage_analyzer.py spec.json tests/ --threshold 90
"""

import argparse
import json
import os
import re
import sys
from pathlib import Path


def load_spec(spec_path: str) -> dict:
    """Load and validate an OpenAPI/Swagger spec file."""
    path = Path(spec_path)
    if not path.exists():
        print(f"Error: spec file not found: {spec_path}", file=sys.stderr)
        sys.exit(1)
    try:
        with open(path, "r", encoding="utf-8") as f:
            spec = json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error: invalid JSON in spec file: {e}", file=sys.stderr)
        sys.exit(1)
    if "openapi" not in spec and "swagger" not in spec:
        print("Error: file does not appear to be an OpenAPI or Swagger spec", file=sys.stderr)
        sys.exit(1)
    return spec


def extract_spec_endpoints(spec: dict) -> list:
    """Extract all endpoint definitions from the spec."""
    endpoints = []
    paths = spec.get("paths", {})
    for path, path_item in paths.items():
        for method in ("get", "post", "put", "patch", "delete", "head", "options"):
            if method not in path_item:
                continue
            operation = path_item[method]
            endpoints.append({
                "path": path,
                "method": method.upper(),
                "operation_id": operation.get("operationId", ""),
                "tags": operation.get("tags", []),
                "summary": operation.get("summary", ""),
                "deprecated": operation.get("deprecated", False),
            })
    return endpoints


def scan_test_files(test_dir: str) -> list:
    """Recursively find test files in the given directory."""
    test_files = []
    test_dir_path = Path(test_dir)
    if not test_dir_path.exists():
        print(f"Error: test directory not found: {test_dir}", file=sys.stderr)
        sys.exit(1)

    # Common test file patterns
    patterns = ["test_*.py", "*_test.py", "*.test.ts", "*.test.js", "*.spec.ts", "*.spec.js"]
    for pattern in patterns:
        test_files.extend(test_dir_path.rglob(pattern))
    return sorted(set(test_files))


def extract_tested_endpoints(test_files: list) -> list:
    """Parse test files to find which endpoints are being tested."""
    tested = []

    # Patterns to match endpoint references in test code
    patterns = [
        # Python httpx/requests: httpx.get(f"{BASE_URL}/api/v1/users")
        re.compile(r'httpx\.(get|post|put|patch|delete)\(\s*f?"[^"]*(/[^\s"{}]+)', re.IGNORECASE),
        # Python httpx/requests with variable: client.get("/api/v1/users")
        re.compile(r'(?:client|requests?)\.(get|post|put|patch|delete)\(\s*[f]?["\']([^"\']+)', re.IGNORECASE),
        # Supertest: request(app).get('/api/v1/users')
        re.compile(r'\.(?:get|post|put|patch|delete)\(\s*["\']([^"\']+)', re.IGNORECASE),
        # Generic URL path pattern in test strings: 'GET /api/v1/users'
        re.compile(r'["\'](?:GET|POST|PUT|PATCH|DELETE)\s+(/[^\s"\']+)', re.IGNORECASE),
        # describe block: describe('GET /api/v1/users'
        re.compile(r'describe\(\s*["\'](?:GET|POST|PUT|PATCH|DELETE)\s+(/[^\s"\']+)', re.IGNORECASE),
    ]

    method_patterns = [
        re.compile(r'httpx\.(get|post|put|patch|delete)', re.IGNORECASE),
        re.compile(r'(?:client|requests?)\.(get|post|put|patch|delete)', re.IGNORECASE),
        re.compile(r'\.(get|post|put|patch|delete)\(\s*["\']/', re.IGNORECASE),
        re.compile(r'["\'](?P<method>GET|POST|PUT|PATCH|DELETE)\s+/', re.IGNORECASE),
    ]

    for test_file in test_files:
        try:
            content = test_file.read_text(encoding="utf-8", errors="replace")
        except OSError:
            continue

        found_paths = set()

        # Extract path references
        for pattern in patterns:
            for match in pattern.finditer(content):
                groups = match.groups()
                # The path is the last group that starts with /
                path = None
                for g in reversed(groups):
                    if g and g.startswith("/"):
                        path = g
                        break
                if path:
                    # Normalize: strip query params, replace specific IDs with {id}
                    path = re.sub(r'\?.*$', '', path)
                    path = re.sub(r'/[0-9a-f]{8,}', '/{id}', path)
                    path = re.sub(r'/\d+', '/{id}', path)
                    found_paths.add(path)

        # Extract methods used in the file
        methods_found = set()
        for mp in method_patterns:
            for match in mp.finditer(content):
                m = match.group(1) if match.lastindex else match.group("method")
                methods_found.add(m.upper())

        for path in found_paths:
            if methods_found:
                for method in methods_found:
                    tested.append({
                        "path": path,
                        "method": method,
                        "test_file": str(test_file),
                    })
            else:
                tested.append({
                    "path": path,
                    "method": "UNKNOWN",
                    "test_file": str(test_file),
                })

    return tested


def _normalize_path(path: str) -> str:
    """Normalize an API path for comparison.

    Converts path parameters like {userId}, :userId to a canonical {param} form.
    """
    # OpenAPI style: /users/{userId} -> /users/{param}
    normalized = re.sub(r'\{[^}]+\}', '{param}', path)
    # Express style: /users/:userId -> /users/{param}
    normalized = re.sub(r':[a-zA-Z_]+', '{param}', normalized)
    return normalized.rstrip("/").lower()


def compute_coverage(spec_endpoints: list, tested_endpoints: list) -> dict:
    """Compare spec endpoints against tested endpoints to find gaps."""
    # Build a set of tested (normalized_path, method) pairs
    tested_set = set()
    tested_files_map = {}
    for te in tested_endpoints:
        key = (_normalize_path(te["path"]), te["method"])
        tested_set.add(key)
        tested_files_map.setdefault(key, []).append(te["test_file"])

    covered = []
    uncovered = []
    deprecated_skipped = 0

    for ep in spec_endpoints:
        if ep["deprecated"]:
            deprecated_skipped += 1
            continue
        norm_path = _normalize_path(ep["path"])
        key = (norm_path, ep["method"])

        if key in tested_set:
            covered.append({
                **ep,
                "test_files": list(set(tested_files_map.get(key, []))),
            })
        else:
            uncovered.append(ep)

    active_total = len(spec_endpoints) - deprecated_skipped
    coverage_pct = (len(covered) / active_total * 100) if active_total > 0 else 0.0

    # Coverage by method
    method_stats = {}
    for ep in spec_endpoints:
        if ep["deprecated"]:
            continue
        m = ep["method"]
        method_stats.setdefault(m, {"total": 0, "covered": 0})
        method_stats[m]["total"] += 1
    for ep in covered:
        m = ep["method"]
        method_stats[m]["covered"] += 1

    # Coverage by tag
    tag_stats = {}
    for ep in spec_endpoints:
        if ep["deprecated"]:
            continue
        tags = ep["tags"] or ["untagged"]
        for tag in tags:
            tag_stats.setdefault(tag, {"total": 0, "covered": 0})
            tag_stats[tag]["total"] += 1
    for ep in covered:
        tags = ep["tags"] or ["untagged"]
        for tag in tags:
            tag_stats[tag]["covered"] += 1

    return {
        "total_endpoints": len(spec_endpoints),
        "deprecated_skipped": deprecated_skipped,
        "active_endpoints": active_total,
        "covered": len(covered),
        "uncovered_count": len(uncovered),
        "coverage_percent": round(coverage_pct, 1),
        "covered_endpoints": covered,
        "uncovered_endpoints": uncovered,
        "by_method": method_stats,
        "by_tag": tag_stats,
        "test_files_scanned": len(set(te["test_file"] for te in tested_endpoints)),
    }


def main():
    parser = argparse.ArgumentParser(
        description="Analyze API test coverage by comparing spec endpoints vs test files.",
        epilog="Example: python coverage_analyzer.py openapi.json tests/ --threshold 90",
    )
    parser.add_argument("spec", help="Path to OpenAPI/Swagger JSON spec file")
    parser.add_argument("test_dir", help="Directory containing test files to scan")
    parser.add_argument(
        "--threshold",
        type=float,
        default=0.0,
        help="Minimum coverage %%. Exit code 1 if below threshold (default: 0 = no threshold)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON",
    )
    parser.add_argument(
        "--show-covered",
        action="store_true",
        help="Also list covered endpoints (default: only show gaps)",
    )
    args = parser.parse_args()

    spec = load_spec(args.spec)
    spec_endpoints = extract_spec_endpoints(spec)

    if not spec_endpoints:
        print("Warning: no endpoints found in spec.", file=sys.stderr)
        sys.exit(1)

    test_files = scan_test_files(args.test_dir)
    tested_endpoints = extract_tested_endpoints(test_files)
    report = compute_coverage(spec_endpoints, tested_endpoints)

    if args.json_output:
        output = {
            "coverage_percent": report["coverage_percent"],
            "total_endpoints": report["active_endpoints"],
            "covered": report["covered"],
            "uncovered_count": report["uncovered_count"],
            "deprecated_skipped": report["deprecated_skipped"],
            "test_files_scanned": report["test_files_scanned"],
            "by_method": report["by_method"],
            "by_tag": report["by_tag"],
            "uncovered_endpoints": [
                {"method": ep["method"], "path": ep["path"], "summary": ep["summary"]}
                for ep in report["uncovered_endpoints"]
            ],
        }
        if args.show_covered:
            output["covered_endpoints"] = [
                {"method": ep["method"], "path": ep["path"], "test_files": ep["test_files"]}
                for ep in report["covered_endpoints"]
            ]
        print(json.dumps(output, indent=2))
    else:
        pct = report["coverage_percent"]
        bar_len = 40
        filled = int(bar_len * pct / 100)
        bar = "#" * filled + "-" * (bar_len - filled)

        print("=== API Test Coverage Report ===")
        print()
        print(f"Coverage: [{bar}] {pct}%")
        print(f"  {report['covered']}/{report['active_endpoints']} endpoints covered")
        if report["deprecated_skipped"]:
            print(f"  {report['deprecated_skipped']} deprecated endpoints skipped")
        print(f"  {report['test_files_scanned']} test files scanned")
        print()

        # By method
        print("Coverage by HTTP method:")
        for method in sorted(report["by_method"].keys()):
            stats = report["by_method"][method]
            m_pct = (stats["covered"] / stats["total"] * 100) if stats["total"] else 0
            print(f"  {method:7s} {stats['covered']}/{stats['total']} ({m_pct:.0f}%)")
        print()

        # By tag
        if report["by_tag"]:
            print("Coverage by tag:")
            for tag in sorted(report["by_tag"].keys()):
                stats = report["by_tag"][tag]
                t_pct = (stats["covered"] / stats["total"] * 100) if stats["total"] else 0
                print(f"  {tag:20s} {stats['covered']}/{stats['total']} ({t_pct:.0f}%)")
            print()

        # Uncovered endpoints
        if report["uncovered_endpoints"]:
            print("UNCOVERED endpoints:")
            for ep in report["uncovered_endpoints"]:
                summary = f" - {ep['summary']}" if ep["summary"] else ""
                print(f"  {ep['method']:7s} {ep['path']}{summary}")
            print()

        # Covered endpoints (optional)
        if args.show_covered and report["covered_endpoints"]:
            print("COVERED endpoints:")
            for ep in report["covered_endpoints"]:
                files = ", ".join(ep["test_files"][:3])
                more = f" (+{len(ep['test_files'])-3} more)" if len(ep["test_files"]) > 3 else ""
                print(f"  {ep['method']:7s} {ep['path']}  <- {files}{more}")
            print()

    # Threshold check
    if args.threshold > 0 and report["coverage_percent"] < args.threshold:
        msg = f"Coverage {report['coverage_percent']}% is below threshold {args.threshold}%"
        if not args.json_output:
            print(f"FAIL: {msg}")
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Generate API test skeletons from OpenAPI/Swagger spec files.

Parses OpenAPI 3.x or Swagger 2.0 JSON spec files and generates test
code scaffolds covering auth, validation, happy path, and error cases
for each endpoint. Outputs pytest+httpx (Python) or vitest+supertest
(TypeScript) test files.

Usage:
    python test_generator.py spec.json --framework pytest
    python test_generator.py spec.json --framework vitest --output tests/
    python test_generator.py spec.json --json
"""

import argparse
import json
import os
import sys
import textwrap
from pathlib import Path


def load_spec(spec_path: str) -> dict:
    """Load and validate an OpenAPI/Swagger spec file."""
    path = Path(spec_path)
    if not path.exists():
        print(f"Error: spec file not found: {spec_path}", file=sys.stderr)
        sys.exit(1)
    if not path.suffix.lower() == ".json":
        print("Error: only JSON spec files are supported", file=sys.stderr)
        sys.exit(1)
    try:
        with open(path, "r", encoding="utf-8") as f:
            spec = json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error: invalid JSON in spec file: {e}", file=sys.stderr)
        sys.exit(1)
    # Basic validation
    if "openapi" not in spec and "swagger" not in spec:
        print("Error: file does not appear to be an OpenAPI or Swagger spec", file=sys.stderr)
        sys.exit(1)
    return spec


def extract_endpoints(spec: dict) -> list:
    """Extract endpoint definitions from spec paths."""
    endpoints = []
    paths = spec.get("paths", {})
    for path, path_item in paths.items():
        for method in ("get", "post", "put", "patch", "delete"):
            if method not in path_item:
                continue
            operation = path_item[method]
            endpoint = {
                "path": path,
                "method": method.upper(),
                "operation_id": operation.get("operationId", ""),
                "summary": operation.get("summary", ""),
                "tags": operation.get("tags", []),
                "parameters": operation.get("parameters", []),
                "request_body": None,
                "responses": {},
                "security": operation.get("security", path_item.get("security", spec.get("security", []))),
            }
            # Extract request body schema
            req_body = operation.get("requestBody", {})
            if req_body:
                content = req_body.get("content", {})
                json_content = content.get("application/json", {})
                schema = json_content.get("schema", {})
                endpoint["request_body"] = _resolve_schema(spec, schema)
            # Extract response status codes
            for status, resp in operation.get("responses", {}).items():
                endpoint["responses"][status] = resp.get("description", "")
            endpoints.append(endpoint)
    return endpoints


def _resolve_schema(spec: dict, schema: dict) -> dict:
    """Resolve a $ref in a schema to its definition."""
    if "$ref" in schema:
        ref_path = schema["$ref"]
        parts = ref_path.lstrip("#/").split("/")
        resolved = spec
        for part in parts:
            resolved = resolved.get(part, {})
        return resolved
    return schema


def _extract_required_fields(schema: dict) -> list:
    """Get required field names from a schema."""
    return schema.get("required", [])


def _extract_properties(schema: dict) -> dict:
    """Get properties with their types from a schema."""
    props = {}
    for name, prop in schema.get("properties", {}).items():
        props[name] = {
            "type": prop.get("type", "string"),
            "format": prop.get("format"),
            "enum": prop.get("enum"),
            "max_length": prop.get("maxLength"),
            "min_length": prop.get("minLength"),
            "minimum": prop.get("minimum"),
            "maximum": prop.get("maximum"),
        }
    return props


def _has_auth(endpoint: dict) -> bool:
    """Check if endpoint requires authentication."""
    return bool(endpoint.get("security"))


def _make_test_name(endpoint: dict) -> str:
    """Create a descriptive test function/describe name."""
    op_id = endpoint.get("operation_id")
    if op_id:
        return op_id
    method = endpoint["method"].lower()
    path_slug = endpoint["path"].strip("/").replace("/", "_").replace("{", "").replace("}", "")
    return f"{method}_{path_slug}"


# --- Pytest (Python) Generator ---

def generate_pytest(endpoints: list, spec: dict) -> dict:
    """Generate pytest+httpx test files. Returns {filename: content}."""
    files = {}
    # Group endpoints by first tag or path segment
    groups = {}
    for ep in endpoints:
        tag = ep["tags"][0] if ep["tags"] else ep["path"].strip("/").split("/")[0]
        tag = tag.lower().replace(" ", "_").replace("-", "_")
        groups.setdefault(tag, []).append(ep)

    for group_name, group_eps in groups.items():
        lines = [
            '"""Auto-generated API tests for {group}."""'.format(group=group_name),
            "import httpx",
            "import pytest",
            "",
            "",
            'BASE_URL = "http://localhost:8000"',
            "",
            "",
            "def auth_headers(token: str = \"test-token\") -> dict:",
            '    return {"Authorization": f"Bearer {token}"}',
            "",
        ]
        for ep in group_eps:
            test_name = _make_test_name(ep)
            method_lower = ep["method"].lower()
            path = ep["path"]
            summary = ep["summary"] or f'{ep["method"]} {ep["path"]}'

            # Happy path test
            lines.append("")
            lines.append(f"class Test{test_name.title().replace('_', '')}:")
            lines.append(f'    """{summary}"""')
            lines.append("")

            # Auth tests
            if _has_auth(ep):
                lines.append(f"    def test_returns_401_without_auth(self):")
                lines.append(f'        """Missing Authorization header returns 401."""')
                lines.append(f"        response = httpx.{method_lower}(")
                lines.append(f'            f"{{BASE_URL}}{path}"')
                lines.append(f"        )")
                lines.append(f"        assert response.status_code == 401")
                lines.append("")
                lines.append(f"    def test_returns_401_with_expired_token(self):")
                lines.append(f'        """Expired token returns 401."""')
                lines.append(f"        response = httpx.{method_lower}(")
                lines.append(f'            f"{{BASE_URL}}{path}",')
                lines.append(f'            headers=auth_headers("expired-token")')
                lines.append(f"        )")
                lines.append(f"        assert response.status_code == 401")
                lines.append("")

            # Validation tests for POST/PUT/PATCH with request body
            if ep["request_body"] and ep["method"] in ("POST", "PUT", "PATCH"):
                schema = ep["request_body"]
                required = _extract_required_fields(schema)
                properties = _extract_properties(schema)

                if required:
                    lines.append(f"    def test_returns_422_with_empty_body(self):")
                    lines.append(f'        """Empty request body returns 422."""')
                    lines.append(f"        response = httpx.{method_lower}(")
                    lines.append(f'            f"{{BASE_URL}}{path}",')
                    lines.append(f"            json={{}},")
                    if _has_auth(ep):
                        lines.append(f"            headers=auth_headers()")
                    lines.append(f"        )")
                    lines.append(f"        assert response.status_code == 422")
                    lines.append("")

                for field_name in required:
                    lines.append(f"    def test_returns_422_when_{field_name}_missing(self):")
                    lines.append(f'        """Missing required field {field_name} returns 422."""')
                    sample = _build_sample_payload(properties, required, exclude=field_name)
                    lines.append(f"        payload = {json.dumps(sample, indent=8)}")
                    lines.append(f"        response = httpx.{method_lower}(")
                    lines.append(f'            f"{{BASE_URL}}{path}",')
                    lines.append(f"            json=payload,")
                    if _has_auth(ep):
                        lines.append(f"            headers=auth_headers()")
                    lines.append(f"        )")
                    lines.append(f"        assert response.status_code == 422")
                    lines.append("")

                # Type mismatch tests
                for field_name, field_info in properties.items():
                    if field_info["type"] == "string":
                        lines.append(f"    def test_returns_422_when_{field_name}_wrong_type(self):")
                        lines.append(f'        """Wrong type for {field_name} returns 422."""')
                        sample = _build_sample_payload(properties, required)
                        sample[field_name] = 99999
                        lines.append(f"        payload = {json.dumps(sample, indent=8)}")
                        lines.append(f"        response = httpx.{method_lower}(")
                        lines.append(f'            f"{{BASE_URL}}{path}",')
                        lines.append(f"            json=payload,")
                        if _has_auth(ep):
                            lines.append(f"            headers=auth_headers()")
                        lines.append(f"        )")
                        lines.append(f"        assert response.status_code == 422")
                        lines.append("")

            # Happy path
            happy_status = "200"
            if ep["method"] == "POST":
                happy_status = "201" if "201" in ep["responses"] else "200"
            elif ep["method"] == "DELETE":
                happy_status = "204" if "204" in ep["responses"] else "200"

            lines.append(f"    def test_success(self):")
            lines.append(f'        """Successful {ep["method"]} returns {happy_status}."""')
            if ep["request_body"] and ep["method"] in ("POST", "PUT", "PATCH"):
                schema = ep["request_body"]
                properties = _extract_properties(schema)
                required = _extract_required_fields(schema)
                sample = _build_sample_payload(properties, required)
                lines.append(f"        payload = {json.dumps(sample, indent=8)}")
                lines.append(f"        response = httpx.{method_lower}(")
                lines.append(f'            f"{{BASE_URL}}{path}",')
                lines.append(f"            json=payload,")
            else:
                lines.append(f"        response = httpx.{method_lower}(")
                lines.append(f'            f"{{BASE_URL}}{path}",')
            if _has_auth(ep):
                lines.append(f"            headers=auth_headers()")
            lines.append(f"        )")
            lines.append(f"        assert response.status_code == {happy_status}")
            lines.append("")

            # Error code tests from responses
            for status_code, desc in ep["responses"].items():
                if status_code in ("200", "201", "204", "default"):
                    continue
                lines.append(f"    def test_returns_{status_code}(self):")
                lines.append(f'        """Trigger {status_code}: {desc}"""')
                lines.append(f"        # TODO: set up conditions to trigger {status_code}")
                lines.append(f"        pass")
                lines.append("")

        filename = f"test_{group_name}.py"
        files[filename] = "\n".join(lines) + "\n"
    return files


# --- Vitest (TypeScript) Generator ---

def generate_vitest(endpoints: list, spec: dict) -> dict:
    """Generate vitest+supertest test files. Returns {filename: content}."""
    files = {}
    groups = {}
    for ep in endpoints:
        tag = ep["tags"][0] if ep["tags"] else ep["path"].strip("/").split("/")[0]
        tag = tag.lower().replace(" ", "-")
        groups.setdefault(tag, []).append(ep)

    for group_name, group_eps in groups.items():
        lines = [
            "import { describe, it, expect } from 'vitest'",
            "import request from 'supertest'",
            "",
            "const BASE_URL = process.env.API_URL || 'http://localhost:3000'",
            "const AUTH_TOKEN = process.env.AUTH_TOKEN || 'test-token'",
            "",
            "function authHeaders() {",
            "  return { Authorization: `Bearer ${AUTH_TOKEN}` }",
            "}",
            "",
        ]
        for ep in group_eps:
            summary = ep["summary"] or f'{ep["method"]} {ep["path"]}'
            method_lower = ep["method"].lower()
            path = ep["path"]

            lines.append(f"describe('{ep['method']} {path}', () => {{")

            if _has_auth(ep):
                lines.append(f"  it('returns 401 without auth header', async () => {{")
                lines.append(f"    const res = await request(BASE_URL).{method_lower}('{path}')")
                lines.append(f"    expect(res.status).toBe(401)")
                lines.append(f"  }})")
                lines.append("")

            if ep["request_body"] and ep["method"] in ("POST", "PUT", "PATCH"):
                schema = ep["request_body"]
                required = _extract_required_fields(schema)
                if required:
                    lines.append(f"  it('returns 422 with empty body', async () => {{")
                    lines.append(f"    const res = await request(BASE_URL)")
                    lines.append(f"      .{method_lower}('{path}')")
                    lines.append(f"      .set(authHeaders())")
                    lines.append(f"      .send({{}})")
                    lines.append(f"    expect(res.status).toBe(422)")
                    lines.append(f"  }})")
                    lines.append("")

            happy_status = "200"
            if ep["method"] == "POST":
                happy_status = "201" if "201" in ep["responses"] else "200"

            lines.append(f"  it('returns {happy_status} on success', async () => {{")
            if ep["request_body"] and ep["method"] in ("POST", "PUT", "PATCH"):
                schema = ep["request_body"]
                properties = _extract_properties(schema)
                required = _extract_required_fields(schema)
                sample = _build_sample_payload(properties, required)
                payload_str = json.dumps(sample, indent=4)
                lines.append(f"    const payload = {payload_str}")
                lines.append(f"    const res = await request(BASE_URL)")
                lines.append(f"      .{method_lower}('{path}')")
                lines.append(f"      .set(authHeaders())")
                lines.append(f"      .send(payload)")
            else:
                lines.append(f"    const res = await request(BASE_URL)")
                lines.append(f"      .{method_lower}('{path}')")
                lines.append(f"      .set(authHeaders())")
            lines.append(f"    expect(res.status).toBe({happy_status})")
            lines.append(f"  }})")

            lines.append(f"}})")
            lines.append("")

        filename = f"{group_name}.test.ts"
        files[filename] = "\n".join(lines) + "\n"
    return files


def _build_sample_payload(properties: dict, required: list, exclude: str = None) -> dict:
    """Build a sample JSON payload from schema properties."""
    payload = {}
    for name, info in properties.items():
        if name == exclude:
            continue
        if info.get("enum"):
            payload[name] = info["enum"][0]
        elif info["type"] == "string":
            payload[name] = f"test-{name}"
        elif info["type"] == "integer":
            payload[name] = 1
        elif info["type"] == "number":
            payload[name] = 1.0
        elif info["type"] == "boolean":
            payload[name] = True
        elif info["type"] == "array":
            payload[name] = []
        elif info["type"] == "object":
            payload[name] = {}
        else:
            payload[name] = f"test-{name}"
    return payload


def generate_report(endpoints: list, files: dict, framework: str) -> dict:
    """Build a summary report of generated tests."""
    total_tests = 0
    for content in files.values():
        if framework == "pytest":
            total_tests += content.count("    def test_")
        else:
            total_tests += content.count("  it(")

    auth_endpoints = sum(1 for ep in endpoints if _has_auth(ep))
    mutation_endpoints = sum(1 for ep in endpoints if ep["method"] in ("POST", "PUT", "PATCH"))

    return {
        "total_endpoints": len(endpoints),
        "auth_endpoints": auth_endpoints,
        "mutation_endpoints": mutation_endpoints,
        "files_generated": len(files),
        "total_test_cases": total_tests,
        "framework": framework,
        "filenames": sorted(files.keys()),
    }


def main():
    parser = argparse.ArgumentParser(
        description="Generate API test skeletons from OpenAPI/Swagger spec files.",
        epilog="Example: python test_generator.py openapi.json --framework pytest --output tests/",
    )
    parser.add_argument("spec", help="Path to OpenAPI/Swagger JSON spec file")
    parser.add_argument(
        "--framework",
        choices=["pytest", "vitest"],
        default="pytest",
        help="Test framework to generate for (default: pytest)",
    )
    parser.add_argument(
        "--output",
        default=None,
        help="Output directory for generated test files (default: stdout preview)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output generation report as JSON",
    )
    args = parser.parse_args()

    spec = load_spec(args.spec)
    endpoints = extract_endpoints(spec)

    if not endpoints:
        print("Warning: no endpoints found in spec.", file=sys.stderr)
        sys.exit(1)

    if args.framework == "pytest":
        files = generate_pytest(endpoints, spec)
    else:
        files = generate_vitest(endpoints, spec)

    report = generate_report(endpoints, files, args.framework)

    # Write files or preview
    if args.output:
        out_dir = Path(args.output)
        out_dir.mkdir(parents=True, exist_ok=True)
        for filename, content in files.items():
            filepath = out_dir / filename
            filepath.write_text(content, encoding="utf-8")

    if args.json_output:
        print(json.dumps(report, indent=2))
    else:
        print(f"=== API Test Generator Report ===")
        print(f"Spec:               {args.spec}")
        print(f"Framework:          {args.framework}")
        print(f"Endpoints found:    {report['total_endpoints']}")
        print(f"  Authenticated:    {report['auth_endpoints']}")
        print(f"  Mutation (POST+): {report['mutation_endpoints']}")
        print(f"Files generated:    {report['files_generated']}")
        print(f"Total test cases:   {report['total_test_cases']}")
        print()
        if args.output:
            print(f"Output directory: {args.output}")
            for fn in report["filenames"]:
                print(f"  {fn}")
        else:
            print("--- Preview (use --output to write files) ---")
            for filename, content in files.items():
                print(f"\n{'='*60}")
                print(f"FILE: {filename}")
                print(f"{'='*60}")
                print(content)


if __name__ == "__main__":
    main()