diff --git a/src/auto_reverse/models.py b/src/auto_reverse/models.py new file mode 100644 index 0000000..f432cda --- /dev/null +++ b/src/auto_reverse/models.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from typing import Any + + +def status_class(status: int) -> str: + return f"{status // 100}xx" + + +@dataclass(frozen=True) +class Signature: + method: str + host: str + path_template: str + status_class: str + + +@dataclass +class CapturedFlow: + method: str + host: str + path: str + query: dict[str, list[str]] + req_headers: dict[str, str] + req_body: bytes | None + status: int + resp_headers: dict[str, str] + resp_body: bytes | None + timestamp: float + + def _json(self, body: bytes | None, headers: dict[str, str]) -> Any | None: + if body is None: + return None + ctype = headers.get("content-type", "").lower() + if "json" not in ctype: + return None + try: + return json.loads(body) + except (ValueError, UnicodeDecodeError): + return None + + def request_json(self) -> Any | None: + return self._json(self.req_body, self.req_headers) + + def response_json(self) -> Any | None: + return self._json(self.resp_body, self.resp_headers) + + +@dataclass +class EndpointRecord: + signature: Signature + sample_count: int = 0 + query_params: set[str] = field(default_factory=set[str]) + request_schema: dict[str, Any] | None = None + response_schema: dict[str, Any] | None = None + # LLM-enriched fields (filled by the doc engine): + summary: str = "" + description: str = "" + tag: str = "" + documented: bool = False diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..8fa6c5d --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,36 @@ +from auto_reverse.models import CapturedFlow, Signature, status_class + + +def test_status_class_buckets(): + assert status_class(200) == "2xx" + assert status_class(201) == "2xx" + assert status_class(404) == "4xx" + assert status_class(503) == "5xx" + + +def test_signature_is_hashable_and_equal(): + a = Signature("GET", "ex.com", "/api/users/{id}", "2xx") + b = Signature("GET", "ex.com", "/api/users/{id}", "2xx") + assert a == b + assert {a, b} == {a} + + +def test_captured_flow_json_body_parsing(): + flow = CapturedFlow( + method="POST", host="ex.com", path="/api/x", query={}, + req_headers={"content-type": "application/json"}, req_body=b'{"a": 1}', + status=201, resp_headers={"content-type": "application/json"}, + resp_body=b'{"ok": true}', timestamp=0.0, + ) + assert flow.request_json() == {"a": 1} + assert flow.response_json() == {"ok": True} + + +def test_captured_flow_non_json_body_returns_none(): + flow = CapturedFlow( + method="GET", host="ex.com", path="/x", query={}, + req_headers={}, req_body=None, status=200, + resp_headers={"content-type": "text/html"}, resp_body=b"", + timestamp=0.0, + ) + assert flow.response_json() is None