feat: core data models (Signature, CapturedFlow, EndpointRecord)

2026-05-31 23:51:47 +08:00
parent 422990bc4e
commit d7b4a44673
2 changed files with 98 additions and 0 deletions
@@ -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
@@ -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"<html>",
        timestamp=0.0,
    )
    assert flow.response_json() is None