feat(connectors): add the http connector and wire the network gate
TDD step 7: read_http executes the optional http extension point behind an injectable transport seam (urllib_get is the only socket path; the suite runs socket-free against a mock). credential_ref resolves from the environment at run time, fail-fast before any transport call; the secret rides in the Authorization header only. Explicit single-slash URL join, CRLF-to-LF normalization, LF line-count cap (missing trailing newline still counts), and code-fence-marker rejection. materialize_bundle renders the body as a verbatim fenced block; the §8 gate test asserts zero transport calls without the opt-in. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QeqhJpYQyghASjiJo5EhGg
This commit is contained in:
parent
f12fdc1dd0
commit
ab1fa4d999
3 changed files with 242 additions and 6 deletions
|
|
@ -9,13 +9,22 @@ from __future__ import annotations
|
|||
import csv
|
||||
import os
|
||||
import sqlite3
|
||||
from collections.abc import Callable
|
||||
from contextlib import closing
|
||||
from pathlib import Path
|
||||
from urllib.error import URLError
|
||||
from urllib.parse import quote
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from .errors import SourceError
|
||||
from .render import sql_value_to_text
|
||||
|
||||
#: The http transport seam: ``(url, credential) -> response body text``.
|
||||
#: Injecting a canned implementation in tests keeps the suite socket-free
|
||||
#: (§11); the default ``urllib_get`` is the ONLY path that opens a socket,
|
||||
#: and it is reached only inside the gated http branch of materialization.
|
||||
HttpGet = Callable[[str, "str | None"], str]
|
||||
|
||||
|
||||
def safe_resolve(root: Path, relative: str) -> Path:
|
||||
"""Resolve `relative` against `root`, fail-closed (the OKF path rule).
|
||||
|
|
@ -116,3 +125,73 @@ def read_sql(
|
|||
except (sqlite3.Error, sqlite3.Warning) as exc:
|
||||
raise SourceError(f"sql extraction failed for query {query!r}: {exc}") from exc
|
||||
return header, rows
|
||||
|
||||
|
||||
def urllib_get(url: str, credential: str | None) -> str:
|
||||
"""The stdlib GET — the only socket path in this library.
|
||||
|
||||
Attaches `Authorization: Bearer {credential}` iff a credential was
|
||||
resolved. Transport/decode failures wrap as SourceError WITHOUT the
|
||||
secret — the credential rides in the header, never in the URL
|
||||
(embedded credentials are already rejected at schema validation).
|
||||
"""
|
||||
headers = {"Authorization": f"Bearer {credential}"} if credential is not None else {}
|
||||
request = Request(url, headers=headers, method="GET")
|
||||
try:
|
||||
with urlopen(request) as response:
|
||||
raw: bytes = response.read()
|
||||
return raw.decode("utf-8")
|
||||
except (URLError, OSError, UnicodeDecodeError) as exc:
|
||||
raise SourceError(f"http GET failed for {url!r}: {exc}") from exc
|
||||
|
||||
|
||||
def read_http(
|
||||
base_url: str,
|
||||
query: str,
|
||||
*,
|
||||
max_rows: int,
|
||||
credential_ref: str | None = None,
|
||||
get: HttpGet = urllib_get,
|
||||
) -> str:
|
||||
"""Execute an `http`-source extraction: GET `base_url` joined with `query`
|
||||
via the injectable `get` seam, returning the body for verbatim fenced
|
||||
rendering (§4/§5).
|
||||
|
||||
`credential_ref` names an environment variable holding the secret,
|
||||
resolved at run time (present-but-unset is fail-fast, before any
|
||||
transport call; None means no auth) — never read from the manifest,
|
||||
never logged, never stamped. URL join is explicit (one slash), not
|
||||
urljoin. The body is CRLF→LF normalized and capped on its LF line count
|
||||
(a missing trailing newline still counts the last line); exceeding
|
||||
max_rows is an ERROR (§8). A body line that is a code-fence marker is
|
||||
rejected — it cannot be safely embedded in a fenced block; fail-fast,
|
||||
never silent corruption.
|
||||
"""
|
||||
if credential_ref is None:
|
||||
credential: str | None = None
|
||||
else:
|
||||
credential = os.environ.get(credential_ref)
|
||||
if not credential:
|
||||
raise SourceError(
|
||||
f"http source credential_ref {credential_ref!r} is not set in the environment "
|
||||
"(credentials resolve at run time, never from the manifest — §4)"
|
||||
)
|
||||
url = base_url.rstrip("/") + "/" + query.lstrip("/")
|
||||
normalized = get(url, credential).replace("\r\n", "\n").replace("\r", "\n")
|
||||
for line in normalized.split("\n"):
|
||||
if line.lstrip(" ").startswith("```"):
|
||||
raise SourceError(
|
||||
f"http extraction {query!r} contains a code-fence marker line (```) — "
|
||||
"cannot be safely embedded in a fenced code block; fail-fast, never "
|
||||
"silent corruption"
|
||||
)
|
||||
newline_count = normalized.count("\n")
|
||||
line_count = (
|
||||
newline_count if (normalized == "" or normalized.endswith("\n")) else newline_count + 1
|
||||
)
|
||||
if line_count > max_rows:
|
||||
raise SourceError(
|
||||
f"http extraction {query!r} exceeds max_rows={max_rows} "
|
||||
"(error, never silent truncation — spec §8)"
|
||||
)
|
||||
return normalized
|
||||
|
|
|
|||
|
|
@ -15,8 +15,8 @@ import re
|
|||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from .connectors import read_csv, read_sql, safe_resolve
|
||||
from .errors import ManifestError, MaterializationError, NetworkGateError, SourceError
|
||||
from .connectors import HttpGet, read_csv, read_http, read_sql, safe_resolve, urllib_get
|
||||
from .errors import ManifestError, MaterializationError, NetworkGateError
|
||||
from .manifest import (
|
||||
Extraction,
|
||||
FileSource,
|
||||
|
|
@ -26,7 +26,7 @@ from .manifest import (
|
|||
generated_filename,
|
||||
load_manifest_bytes,
|
||||
)
|
||||
from .render import render_table
|
||||
from .render import render_fenced_block, render_table
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -148,6 +148,7 @@ def materialize_bundle(
|
|||
ingested_at: str,
|
||||
*,
|
||||
allow_network: bool = False,
|
||||
http_get: HttpGet | None = None,
|
||||
) -> IngestResult:
|
||||
"""Materialize a manifest's extractions into an OKF bundle (§5).
|
||||
|
||||
|
|
@ -155,7 +156,9 @@ def materialize_bundle(
|
|||
verbatim) — no wall-clock default; this is what makes golden extractions
|
||||
bit-deterministic. `allow_network` is the §8 per-run network opt-in: an
|
||||
`http` source is refused fail-fast unless it is set — the manifest itself
|
||||
cannot grant network access. Source calls are logged per §8 (which
|
||||
cannot grant network access. `http_get` optionally injects the transport
|
||||
seam (default urllib_get, the only socket path; ignored for `file`/`sql`)
|
||||
so tests run socket-free (§11). Source calls are logged per §8 (which
|
||||
source, when, row count) — never cell contents, never secrets.
|
||||
"""
|
||||
if not _INGESTED_AT_RE.match(ingested_at):
|
||||
|
|
@ -206,8 +209,17 @@ def materialize_bundle(
|
|||
)
|
||||
body = render_table(header, rows)
|
||||
row_count = len(rows)
|
||||
else: # HttpSource — transport lands with the http connector step.
|
||||
raise SourceError("http transport is not implemented yet")
|
||||
else: # HttpSource — the gate above guarantees allow_network here.
|
||||
get = http_get if http_get is not None else urllib_get
|
||||
text = read_http(
|
||||
source.base_url,
|
||||
extraction.query,
|
||||
max_rows=extraction.max_rows,
|
||||
credential_ref=source.credential_ref,
|
||||
get=get,
|
||||
)
|
||||
body = render_fenced_block(text) # verbatim, NOT table-escaped
|
||||
row_count = len(text.splitlines())
|
||||
_LOGGER.info(
|
||||
"source call: source=%s ingested_at=%s rows=%d", source.id, ingested_at, row_count
|
||||
)
|
||||
|
|
|
|||
145
tests/test_http_connector.py
Normal file
145
tests/test_http_connector.py
Normal file
|
|
@ -0,0 +1,145 @@
|
|||
"""The `http` connector and the §8 network gate (ingest-spec §4, §8).
|
||||
|
||||
The optional extension point, exercised ONLY against an injected mock
|
||||
transport — the test suite never opens a socket (§11). credential_ref names
|
||||
an environment variable resolved at run time; the secret rides in the
|
||||
Authorization header, never in the URL and never in any output.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from llm_ingestion_okf.connectors import read_http
|
||||
from llm_ingestion_okf.errors import NetworkGateError, SourceError
|
||||
from llm_ingestion_okf.materialize import materialize_bundle
|
||||
|
||||
INGESTED_AT = "2026-07-16T12:00:00Z"
|
||||
|
||||
|
||||
class MockGet:
|
||||
"""A canned transport: records calls, returns a fixed body."""
|
||||
|
||||
def __init__(self, body: str = "hello\n") -> None:
|
||||
self.body = body
|
||||
self.calls: list[tuple[str, str | None]] = []
|
||||
|
||||
def __call__(self, url: str, credential: str | None) -> str:
|
||||
self.calls.append((url, credential))
|
||||
return self.body
|
||||
|
||||
|
||||
# --- URL join and credential resolution ---
|
||||
|
||||
|
||||
def test_url_join_is_explicit_single_slash() -> None:
|
||||
get = MockGet()
|
||||
read_http("https://api.test/", "/orders", max_rows=10, get=get)
|
||||
assert get.calls == [("https://api.test/orders", None)]
|
||||
|
||||
|
||||
def test_no_credential_ref_means_no_credential() -> None:
|
||||
get = MockGet()
|
||||
read_http("https://api.test", "orders", max_rows=10, get=get)
|
||||
assert get.calls[0][1] is None
|
||||
|
||||
|
||||
def test_credential_resolved_from_environment(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("OKF_TEST_TOKEN", "s3cret")
|
||||
get = MockGet()
|
||||
read_http("https://api.test", "orders", max_rows=10, credential_ref="OKF_TEST_TOKEN", get=get)
|
||||
url, credential = get.calls[0]
|
||||
assert credential == "s3cret"
|
||||
assert "s3cret" not in url
|
||||
|
||||
|
||||
def test_unset_credential_ref_fails_before_transport(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("OKF_TEST_TOKEN", raising=False)
|
||||
get = MockGet()
|
||||
with pytest.raises(SourceError):
|
||||
read_http(
|
||||
"https://api.test", "orders", max_rows=10, credential_ref="OKF_TEST_TOKEN", get=get
|
||||
)
|
||||
assert get.calls == []
|
||||
|
||||
|
||||
# --- body handling ---
|
||||
|
||||
|
||||
def test_crlf_normalized_to_lf() -> None:
|
||||
assert read_http("https://x", "q", max_rows=10, get=MockGet("a\r\nb\r")) == "a\nb\n"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("body", ["```\n", "before\n```fence\n", " ```indented\n"])
|
||||
def test_code_fence_line_in_body_rejected(body: str) -> None:
|
||||
"""A body that cannot be safely embedded in a fenced block is an error —
|
||||
never silently-corrupted markdown."""
|
||||
with pytest.raises(SourceError):
|
||||
read_http("https://x", "q", max_rows=10, get=MockGet(body))
|
||||
|
||||
|
||||
def test_line_count_at_cap_is_allowed() -> None:
|
||||
assert read_http("https://x", "q", max_rows=2, get=MockGet("a\nb\n")) == "a\nb\n"
|
||||
|
||||
|
||||
def test_missing_trailing_newline_still_counts_last_line() -> None:
|
||||
with pytest.raises(SourceError):
|
||||
read_http("https://x", "q", max_rows=2, get=MockGet("a\nb\nc"))
|
||||
|
||||
|
||||
def test_exceeding_cap_is_an_error() -> None:
|
||||
with pytest.raises(SourceError):
|
||||
read_http("https://x", "q", max_rows=1, get=MockGet("a\nb\n"))
|
||||
|
||||
|
||||
# --- materialization end-to-end against the mock transport ---
|
||||
|
||||
|
||||
def http_manifest(tmp_path: Path) -> Path:
|
||||
data: dict[str, Any] = {
|
||||
"manifest_version": 1,
|
||||
"source": {"type": "http", "id": "api-1", "base_url": "https://api.test"},
|
||||
"bundle_summary": "A test bundle.",
|
||||
"extractions": [
|
||||
{
|
||||
"id": "orders",
|
||||
"title": "Orders",
|
||||
"query": "/orders",
|
||||
"okf_type": "dataset",
|
||||
"max_rows": 10,
|
||||
}
|
||||
],
|
||||
}
|
||||
src = tmp_path / "src"
|
||||
src.mkdir()
|
||||
path = src / "manifest.json"
|
||||
path.write_text(json.dumps(data), encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def test_http_materialization_renders_verbatim_fenced_body(tmp_path: Path) -> None:
|
||||
manifest_path = http_manifest(tmp_path)
|
||||
bundle = tmp_path / "bundle"
|
||||
get = MockGet('{"orders": [1, 2]}\n')
|
||||
materialize_bundle(manifest_path, bundle, INGESTED_AT, allow_network=True, http_get=get)
|
||||
content = (bundle / "ingest-orders.md").read_text(encoding="utf-8")
|
||||
assert content.endswith('```\n{"orders": [1, 2]}\n```\n')
|
||||
assert get.calls == [("https://api.test/orders", None)]
|
||||
index = (bundle / "index.md").read_text(encoding="utf-8")
|
||||
assert "- [Orders](ingest-orders.md)\n" in index
|
||||
|
||||
|
||||
def test_network_gate_blocks_before_any_transport_call(tmp_path: Path) -> None:
|
||||
"""Load-bearing (spec §11): without the per-run opt-in the connector
|
||||
refuses fail-fast — zero transport calls, zero disk mutation."""
|
||||
manifest_path = http_manifest(tmp_path)
|
||||
bundle = tmp_path / "bundle"
|
||||
get = MockGet()
|
||||
with pytest.raises(NetworkGateError):
|
||||
materialize_bundle(manifest_path, bundle, INGESTED_AT, http_get=get)
|
||||
assert get.calls == []
|
||||
assert not bundle.exists()
|
||||
Loading…
Add table
Add a link
Reference in a new issue