Enable runtime mode switching between demo and production#30
Enable runtime mode switching between demo and production#30DevOpsMadDog wants to merge 2 commits into
Conversation
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting
| """Tiny FastAPI-compatible façade for unit tests.""" | ||
| from __future__ import annotations | ||
|
|
||
| import inspect | ||
| from dataclasses import dataclass | ||
| from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple, Type, get_type_hints | ||
|
|
||
| try: # pragma: no cover - optional dependency for typing checks | ||
| from pydantic import BaseModel, ValidationError | ||
| except ModuleNotFoundError: # pragma: no cover - the stub ships alongside | ||
| from pydantic import BaseModel, ValidationError # type: ignore | ||
|
|
||
|
|
||
| class HTTPException(Exception): | ||
| def __init__(self, status_code: int, detail: Any) -> None: | ||
| super().__init__(str(detail)) | ||
| self.status_code = status_code | ||
| self.detail = detail | ||
|
|
||
|
|
||
| class RequestValidationError(Exception): | ||
| def __init__(self, errors: List[Dict[str, Any]]) -> None: | ||
| super().__init__("Validation failed") | ||
| self.errors = errors | ||
|
|
||
|
|
||
| @dataclass | ||
| class _Route: | ||
| method: str | ||
| path: str | ||
| endpoint: Callable[..., Any] | ||
|
|
||
| def __post_init__(self) -> None: | ||
| self.signature = inspect.signature(self.endpoint) | ||
| raw_segments = [segment for segment in self.path.strip("/").split("/") if segment] | ||
| self._segments: List[Tuple[str, Optional[str]]] = [] | ||
| for segment in raw_segments: | ||
| if segment.startswith("{") and segment.endswith("}"): | ||
| self._segments.append(("param", segment[1:-1])) | ||
| else: | ||
| self._segments.append(("literal", segment)) | ||
| try: | ||
| self._type_hints = get_type_hints(self.endpoint) | ||
| except Exception: # pragma: no cover - fallback for dynamic globals | ||
| self._type_hints = {} | ||
|
|
||
| def match(self, method: str, path: str) -> Optional[Mapping[str, str]]: | ||
| if method != self.method: | ||
| return None | ||
| segments = [segment for segment in path.strip("/").split("/") if segment] | ||
| if len(segments) != len(self._segments): | ||
| return None | ||
| params: Dict[str, str] = {} | ||
| for (kind, value), segment in zip(self._segments, segments): | ||
| if kind == "literal" and value != segment: | ||
| return None | ||
| if kind == "param": | ||
| params[value] = segment | ||
| return params | ||
|
|
||
| def invoke(self, params: Mapping[str, str], body: Optional[Dict[str, Any]]) -> Any: | ||
| kwargs: Dict[str, Any] = {} | ||
| for name, parameter in self.signature.parameters.items(): | ||
| annotation = self._type_hints.get(name, parameter.annotation) | ||
| if name in params: | ||
| kwargs[name] = params[name] | ||
| continue | ||
|
|
||
| if isinstance(annotation, type) and issubclass(annotation, BaseModel): | ||
| model_data = body or {} | ||
| try: | ||
| kwargs[name] = annotation(**model_data) | ||
| except ValidationError as exc: | ||
| raise RequestValidationError(exc.errors()) from exc | ||
| continue | ||
|
|
||
| if name == "body": | ||
| kwargs[name] = body | ||
| elif parameter.default is not inspect._empty: | ||
| kwargs[name] = parameter.default | ||
| else: | ||
| kwargs[name] = None | ||
|
|
||
| return self.endpoint(**kwargs) | ||
|
|
||
|
|
||
| class FastAPI: | ||
| def __init__(self, title: str | None = None, version: str | None = None) -> None: | ||
| self.title = title | ||
| self.version = version | ||
| self._routes: List[_Route] = [] | ||
|
|
||
| def post(self, path: str, summary: str | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]: | ||
| return self._register("POST", path) | ||
|
|
||
| def get(self, path: str, summary: str | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]: | ||
| return self._register("GET", path) | ||
|
|
||
| def _register(self, method: str, path: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]: | ||
| def decorator(func: Callable[..., Any]) -> Callable[..., Any]: | ||
| self._routes.append(_Route(method, path, func)) | ||
| return func | ||
|
|
||
| return decorator | ||
|
|
||
| # Internal helpers for the TestClient | ||
| def _handle(self, method: str, path: str, body: Optional[Dict[str, Any]]) -> Any: | ||
| for route in self._routes: | ||
| params = route.match(method, path) | ||
| if params is not None: | ||
| return route.invoke(params, body) | ||
| raise HTTPException(status_code=404, detail="Not Found") | ||
|
|
||
|
|
||
| from .testclient import TestClient # noqa: E402 (import after FastAPI definition) | ||
|
|
||
| __all__ = ["FastAPI", "HTTPException", "RequestValidationError", "TestClient"] |
There was a problem hiding this comment.
Stub fastapi module omits APIRouter
The newly added fastapi stub only exposes FastAPI, HTTPException, RequestValidationError, and TestClient. Because the stub shadows the real FastAPI package, existing modules that do from fastapi import APIRouter (or other primitives such as Depends, UploadFile, etc.) will now raise ImportError during import, preventing any API code or tests from running. The stub needs to provide at least no-op versions of the objects that the codebase imports, otherwise the application cannot start.
Useful? React with 👍 / 👎.
…endpoints Closed in this batch (canonical envelope, mirroring batch-6 pattern): - /api/v1/posture-reports/reports #7: canonical envelope shipped - /api/v1/cloud-ir/incidents #17: canonical envelope shipped - /api/v1/network-forensics/captures #21: canonical envelope shipped - /api/v1/network-segmentation/segments #22: canonical envelope shipped - /api/v1/microsegmentation/segments #23: canonical envelope shipped - /api/v1/awareness-gamification/challenges #29: canonical envelope shipped - /api/v1/gdpr/activities #30: canonical envelope shipped Pattern (class-c): all seven list endpoints upgraded from minimal {<legacy_key>, total, hint} to the canonical batch-6/batch-7 envelope: { "items": [...], "<legacy_key>": [...], # back-compat (reports/incidents/captures/etc.) "total": int, "org_id": str, "limit": int, # ge=1, le=500 — defaults to 50 "offset": int, # ge=0 — defaults to 0 "filters_applied": {...} # echoes every filter param (None if unset) "hint": str # only present when total == 0 } Each endpoint now (1) accepts limit + offset query params with FastAPI ge/le validation, (2) echoes every filter back into filters_applied even when None (no missing keys), (3) always returns the full envelope shape even on hit (legacy clients keep their original key, new clients use items + pagination context), (4) preserves the actionable empty-state hint with a "this is correct for fresh tenants" framing. Triage status update: 26/30 fully closed. 4 class-a deferred (need real cloud creds, OAuth flows, or PAM tenant access not present in fleet — sprint-able with customer engagement). All class-b importer-gated endpoints (8) and all class-c structured-empty endpoints (12) now closed. Verified: pytest tests/test_empty_endpoints_batch7.py 11/11 PASS. Beast Mode regression on phase4/phase7/trustgraph/pipeline_api: 170/170 PASS. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…een at HEAD f0e2a93 185 passed (test_phase4_integration + test_phase5_enterprise + test_phase8_mcp + test_phase10_e2e), 0 failed, 0 errors. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Summary
Testing
pytesthttps://chatgpt.com/codex/tasks/task_e_68e0f7f203f48329856b070b43892efc