From 323b4b3ad3c2c023cb51a1785b30464a9c2626fd Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 14:07:58 +0200 Subject: [PATCH 01/14] feat: csv_source rules and CsvSourceReader port Signed-off-by: Simone Carolini --- continuo_python_runtime/csv_source.py | 66 +++++++++++++++++++++++++++ tests/test_csv_source.py | 52 +++++++++++++++++++++ 2 files changed, 118 insertions(+) create mode 100644 continuo_python_runtime/csv_source.py create mode 100644 tests/test_csv_source.py diff --git a/continuo_python_runtime/csv_source.py b/continuo_python_runtime/csv_source.py new file mode 100644 index 0000000..8b8bbed --- /dev/null +++ b/continuo_python_runtime/csv_source.py @@ -0,0 +1,66 @@ +"""CSV-source rules shared by the run harness and the validation runner: +the URI grammar, the header-conformance rule, and the reader port both +consumers depend on. Dependency-free — adapters that do I/O live in +continuo_python_runtime/csv_readers/ and implement CsvSourceReader. +""" +from abc import ABC, abstractmethod +from dataclasses import dataclass +from pathlib import Path + +HEADER_PROBE_BYTES = 65_536 # first ranged fetch when probing for the header line +MAX_HEADER_BYTES = 1_048_576 # a header line longer than this is a failure + + +@dataclass(frozen=True) +class CsvUri: + scheme: str # "s3" | "https" + raw: str + bucket: str = "" # s3 only + key: str = "" # s3 only + + +def parse_csv_uri(uri: str) -> CsvUri: + """Parse a csv source URI. Accepts exactly s3://bucket/key and https://... + + Raises ValueError for anything else (http:// included) so contract + validation fails at lint/parse time, never at run time. + """ + if uri.startswith("s3://"): + bucket, _, key = uri[len("s3://"):].partition("/") + if not bucket or not key: + raise ValueError(f"invalid s3 csv uri (missing bucket or key): {uri!r}") + return CsvUri(scheme="s3", raw=uri, bucket=bucket, key=key) + if uri.startswith("https://") and len(uri) > len("https://"): + return CsvUri(scheme="https", raw=uri) + raise ValueError( + f"invalid csv uri {uri!r}: must be s3://bucket/key or https://..." + ) + + +def check_header(header_cols: list[str], declared_cols: list[str]) -> set[str]: + """Presence-only header conformance: every declared column must appear in + the CSV header, in any order. Returns the set of header columns NOT + declared (extras) so callers can surface them as a warning. + + Raises ValueError naming every missing declared column. + """ + header = set(header_cols) + missing = [c for c in declared_cols if c not in header] + if missing: + raise ValueError(f"csv header missing declared column(s): {sorted(missing)}") + return header - set(declared_cols) + + +class CsvSourceReader(ABC): + """Port for reading a csv source. Implemented by csv_readers adapters; + consumed by the run harness (full fetch) and the validation runner + (header line only). The dependency arrow runs adapter -> this port.""" + + @abstractmethod + def fetch_header_line(self, uri: CsvUri) -> str: + """Return the CSV's first line (no trailing newline). Raises on an + unreachable source or a header longer than MAX_HEADER_BYTES.""" + + @abstractmethod + def fetch(self, uri: CsvUri, dest: Path) -> Path: + """Stream the full object to dest and return dest.""" diff --git a/tests/test_csv_source.py b/tests/test_csv_source.py new file mode 100644 index 0000000..4f8bd9c --- /dev/null +++ b/tests/test_csv_source.py @@ -0,0 +1,52 @@ +"""tests/test_csv_source.py""" +import pytest + +from continuo_python_runtime.csv_source import ( + CsvSourceReader, + check_header, + parse_csv_uri, +) + + +def test_parse_s3_uri(): + uri = parse_csv_uri("s3://drops/exports/orders.csv") + assert (uri.scheme, uri.bucket, uri.key) == ("s3", "drops", "exports/orders.csv") + assert uri.raw == "s3://drops/exports/orders.csv" + + +def test_parse_https_uri(): + uri = parse_csv_uri("https://example.com/x.csv") + assert uri.scheme == "https" + assert uri.raw == "https://example.com/x.csv" + + +@pytest.mark.parametrize("bad", [ + "http://example.com/x.csv", # http is not accepted + "s3://bucket-only", + "s3://bucket/", + "gs://bucket/key", + "orders.csv", + "", +]) +def test_parse_rejects_invalid(bad): + with pytest.raises(ValueError): + parse_csv_uri(bad) + + +def test_check_header_presence_only_any_order(): + extras = check_header(["b", "a", "c"], ["a", "b"]) + assert extras == {"c"} + + +def test_check_header_no_extras(): + assert check_header(["a", "b"], ["a", "b"]) == set() + + +def test_check_header_missing_declared_column_raises(): + with pytest.raises(ValueError, match="missing declared column"): + check_header(["a"], ["a", "b"]) + + +def test_reader_port_is_abstract(): + with pytest.raises(TypeError): + CsvSourceReader() # type: ignore[abstract] From c06345ba9a4639b0007eba18e721cdf69a187f80 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 14:12:17 +0200 Subject: [PATCH 02/14] fix: reject empty-host https csv uri Signed-off-by: Simone Carolini --- continuo_python_runtime/csv_source.py | 7 +++++-- tests/test_csv_source.py | 2 ++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/continuo_python_runtime/csv_source.py b/continuo_python_runtime/csv_source.py index 8b8bbed..a93751b 100644 --- a/continuo_python_runtime/csv_source.py +++ b/continuo_python_runtime/csv_source.py @@ -30,8 +30,11 @@ def parse_csv_uri(uri: str) -> CsvUri: if not bucket or not key: raise ValueError(f"invalid s3 csv uri (missing bucket or key): {uri!r}") return CsvUri(scheme="s3", raw=uri, bucket=bucket, key=key) - if uri.startswith("https://") and len(uri) > len("https://"): - return CsvUri(scheme="https", raw=uri) + if uri.startswith("https://"): + remainder = uri[len("https://"):] + host, _, _ = remainder.partition("/") + if host: + return CsvUri(scheme="https", raw=uri) raise ValueError( f"invalid csv uri {uri!r}: must be s3://bucket/key or https://..." ) diff --git a/tests/test_csv_source.py b/tests/test_csv_source.py index 4f8bd9c..7e7d2e3 100644 --- a/tests/test_csv_source.py +++ b/tests/test_csv_source.py @@ -25,6 +25,8 @@ def test_parse_https_uri(): "s3://bucket-only", "s3://bucket/", "gs://bucket/key", + "https:///x.csv", # empty host not accepted + "https://", # empty host not accepted "orders.csv", "", ]) From 1877d94414e3e47bd090e7e800cfb3afb12d9cb1 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 14:18:51 +0200 Subject: [PATCH 03/14] feat: S3 and HTTPS CsvSourceReader adapters, integration-tested against minio and a live HTTP server Signed-off-by: Simone Carolini --- .../csv_readers/__init__.py | 11 ++ continuo_python_runtime/csv_readers/https.py | 40 +++++++ continuo_python_runtime/csv_readers/s3.py | 44 +++++++ tests/conftest.py | 93 +++++++++++++++ tests/test_csv_readers_integration.py | 110 ++++++++++++++++++ 5 files changed, 298 insertions(+) create mode 100644 continuo_python_runtime/csv_readers/__init__.py create mode 100644 continuo_python_runtime/csv_readers/https.py create mode 100644 continuo_python_runtime/csv_readers/s3.py create mode 100644 tests/test_csv_readers_integration.py diff --git a/continuo_python_runtime/csv_readers/__init__.py b/continuo_python_runtime/csv_readers/__init__.py new file mode 100644 index 0000000..52e37a3 --- /dev/null +++ b/continuo_python_runtime/csv_readers/__init__.py @@ -0,0 +1,11 @@ +"""continuo_python_runtime/csv_readers/__init__.py""" +from continuo_python_runtime.csv_source import CsvSourceReader, CsvUri +from continuo_python_runtime.csv_readers.https import HttpsCsvSourceReader +from continuo_python_runtime.csv_readers.s3 import S3CsvSourceReader + + +def reader_for(uri: CsvUri) -> CsvSourceReader: + """Composition edge: pick the adapter for the parsed scheme.""" + if uri.scheme == "s3": + return S3CsvSourceReader() + return HttpsCsvSourceReader() diff --git a/continuo_python_runtime/csv_readers/https.py b/continuo_python_runtime/csv_readers/https.py new file mode 100644 index 0000000..6b2c533 --- /dev/null +++ b/continuo_python_runtime/csv_readers/https.py @@ -0,0 +1,40 @@ +"""continuo_python_runtime/csv_readers/https.py""" +import shutil +import urllib.request +from pathlib import Path + +from continuo_python_runtime.csv_source import ( + HEADER_PROBE_BYTES, + MAX_HEADER_BYTES, + CsvSourceReader, + CsvUri, +) + + +class HttpsCsvSourceReader(CsvSourceReader): + """Reads a csv source over HTTPS (public URLs; no auth in v1). A server + that ignores Range and answers 200 is handled by streaming and reading + only to the first newline before closing the connection.""" + + def fetch_header_line(self, uri: CsvUri) -> str: + req = urllib.request.Request( + uri.raw, headers={"Range": f"bytes=0-{HEADER_PROBE_BYTES - 1}"}) + buf = b"" + with urllib.request.urlopen(req) as resp: # noqa: S310 — scheme gated by parse_csv_uri + while b"\n" not in buf: + chunk = resp.read(HEADER_PROBE_BYTES) + if not chunk: + break + buf += chunk + if len(buf) > MAX_HEADER_BYTES: + raise ValueError( + f"csv header line exceeds {MAX_HEADER_BYTES} bytes: {uri.raw}") + return buf.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") + + def fetch(self, uri: CsvUri, dest: Path) -> Path: + with urllib.request.urlopen(uri.raw) as resp, open(dest, "wb") as f: # noqa: S310 + shutil.copyfileobj(resp, f) + return dest + + +assert issubclass(HttpsCsvSourceReader, CsvSourceReader) diff --git a/continuo_python_runtime/csv_readers/s3.py b/continuo_python_runtime/csv_readers/s3.py new file mode 100644 index 0000000..4a7ba3e --- /dev/null +++ b/continuo_python_runtime/csv_readers/s3.py @@ -0,0 +1,44 @@ +"""continuo_python_runtime/csv_readers/s3.py""" +from pathlib import Path + +from continuo_python_runtime.csv_source import ( + HEADER_PROBE_BYTES, + MAX_HEADER_BYTES, + CsvSourceReader, + CsvUri, +) +from continuo_python_runtime.validation.s3 import make_s3_client + + +class S3CsvSourceReader(CsvSourceReader): + """Reads a csv source from S3. Reuses make_s3_client so S3_ENDPOINT_URL + (minio, localstack) and boto3's own credential chain behave identically + to the validation runner's existing S3 access.""" + + def fetch_header_line(self, uri: CsvUri) -> str: + client = make_s3_client() + start = 0 + buf = b"" + while True: + end = start + HEADER_PROBE_BYTES - 1 + body = client.get_object( + Bucket=uri.bucket, Key=uri.key, Range=f"bytes={start}-{end}" + )["Body"].read() + buf += body + if b"\n" in buf: + return buf.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") + if len(body) < HEADER_PROBE_BYTES: # whole object read, no newline + return buf.rstrip(b"\r").decode("utf-8") + if len(buf) > MAX_HEADER_BYTES: + raise ValueError( + f"csv header line exceeds {MAX_HEADER_BYTES} bytes: {uri.raw}") + start += HEADER_PROBE_BYTES + + def fetch(self, uri: CsvUri, dest: Path) -> Path: + client = make_s3_client() + with open(dest, "wb") as f: + client.download_fileobj(uri.bucket, uri.key, f) + return dest + + +assert issubclass(S3CsvSourceReader, CsvSourceReader) diff --git a/tests/conftest.py b/tests/conftest.py index 0b8b9dc..238e779 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,12 @@ """Shared test fixtures.""" import os +import subprocess import sys +import time +import urllib.error +import urllib.request +import uuid from pathlib import Path import pytest @@ -141,3 +146,91 @@ def harness_repo(tmp_path): "output_columns": [{"name": "id", "type": "INTEGER", "nullable": False}], }]})) return tmp_path + + +@pytest.fixture(scope="session") +def minio_container(): + """Session-scoped real minio backend, started via a plain `docker run`. + + No testcontainers/pytest-docker dependency in this repo, so this drives + docker directly. Publishes minio's 9000 to an EPHEMERAL host port (colima + may already hold 9000 for another stack) and waits for minio's own + /minio/health/live endpoint to return 200 before yielding. Any failure — + docker missing, `docker run` erroring, the health check never turning + green — raises so the dependent tests error loudly instead of silently + skipping, per this suite's "real backends, no stubs, no silent skip" + integration-testing policy. + """ + name = f"csv-readers-minio-{uuid.uuid4().hex[:12]}" + try: + subprocess.run( + [ + "docker", "run", "-d", "--name", name, + "-p", "0:9000", + "-e", "MINIO_ROOT_USER=minioadmin", + "-e", "MINIO_ROOT_PASSWORD=minioadmin", + "minio/minio:latest", + "server", "/data", "--address", ":9000", + ], + check=True, capture_output=True, text=True, timeout=60, + ) + except FileNotFoundError as exc: + raise RuntimeError( + "docker is not available; the csv reader integration tests " + "require a live docker daemon" + ) from exc + except subprocess.CalledProcessError as exc: + raise RuntimeError(f"docker run for minio failed: {exc.stderr}") from exc + + try: + port_out = subprocess.run( + ["docker", "port", name, "9000/tcp"], + check=True, capture_output=True, text=True, timeout=10, + ).stdout.strip() + # e.g. "0.0.0.0:54321\n[::]:54321" -- take the first (ipv4) mapping. + host_port = port_out.splitlines()[0].rsplit(":", 1)[1] + endpoint = f"http://127.0.0.1:{host_port}" + + deadline = time.monotonic() + 30 + last_error: Exception | None = None + healthy = False + while time.monotonic() < deadline: + try: + with urllib.request.urlopen( + f"{endpoint}/minio/health/live", timeout=2 + ) as resp: + if resp.status == 200: + healthy = True + break + except (urllib.error.URLError, ConnectionError, TimeoutError) as exc: + last_error = exc + time.sleep(0.5) + if not healthy: + raise RuntimeError( + f"minio container {name} never became healthy at " + f"{endpoint}/minio/health/live: {last_error}" + ) + + # make_s3_client() forwards only S3_ENDPOINT_URL and otherwise leaves + # credentials to boto3's own chain, so the chain needs something to + # find. This is session-scoped (not monkeypatch, which is + # function-scoped) -- set directly and restore on teardown. + prior_key = os.environ.get("AWS_ACCESS_KEY_ID") + prior_secret = os.environ.get("AWS_SECRET_ACCESS_KEY") + os.environ["AWS_ACCESS_KEY_ID"] = "minioadmin" + os.environ["AWS_SECRET_ACCESS_KEY"] = "minioadmin" + try: + yield (endpoint, "minioadmin", "minioadmin") + finally: + if prior_key is None: + os.environ.pop("AWS_ACCESS_KEY_ID", None) + else: + os.environ["AWS_ACCESS_KEY_ID"] = prior_key + if prior_secret is None: + os.environ.pop("AWS_SECRET_ACCESS_KEY", None) + else: + os.environ["AWS_SECRET_ACCESS_KEY"] = prior_secret + finally: + subprocess.run( + ["docker", "rm", "-f", name], capture_output=True, text=True, timeout=30 + ) diff --git a/tests/test_csv_readers_integration.py b/tests/test_csv_readers_integration.py new file mode 100644 index 0000000..f2a4047 --- /dev/null +++ b/tests/test_csv_readers_integration.py @@ -0,0 +1,110 @@ +"""tests/test_csv_readers_integration.py + +Real-backend tests: minio for S3CsvSourceReader, a live stdlib HTTP server for +HttpsCsvSourceReader. HTTPS-the-scheme is terminated before our code in +production (urllib handles TLS); the adapter's Range/stream logic is what these +tests exercise, so the local server speaking plain HTTP to a loopback socket is +acceptable ONLY here — build the CsvUri directly rather than via parse_csv_uri. +""" +import http.server +import socketserver +import threading + +import boto3 +import pytest + +from continuo_python_runtime.csv_source import CsvUri, parse_csv_uri +from continuo_python_runtime.csv_readers import reader_for +from continuo_python_runtime.csv_readers.https import HttpsCsvSourceReader +from continuo_python_runtime.csv_readers.s3 import S3CsvSourceReader + +CSV_BODY = b"order_id,amount,extra\n1,10.5,x\n2,20.0,y\n" + + +@pytest.fixture(scope="session") +def minio(minio_container): # minio_container: session fixture starting minio via docker + endpoint, access, secret = minio_container + client = boto3.client( + "s3", endpoint_url=endpoint, + aws_access_key_id=access, aws_secret_access_key=secret, + ) + client.create_bucket(Bucket="drops") + client.put_object(Bucket="drops", Key="orders.csv", Body=CSV_BODY) + return endpoint + + +def test_s3_fetch_header_line(minio, monkeypatch): + monkeypatch.setenv("S3_ENDPOINT_URL", minio) + header = S3CsvSourceReader().fetch_header_line(parse_csv_uri("s3://drops/orders.csv")) + assert header == "order_id,amount,extra" + + +def test_s3_fetch_full_object(minio, monkeypatch, tmp_path): + monkeypatch.setenv("S3_ENDPOINT_URL", minio) + dest = S3CsvSourceReader().fetch( + parse_csv_uri("s3://drops/orders.csv"), tmp_path / "o.csv") + assert dest.read_bytes() == CSV_BODY + + +def test_s3_missing_object_raises(minio, monkeypatch): + monkeypatch.setenv("S3_ENDPOINT_URL", minio) + with pytest.raises(Exception): + S3CsvSourceReader().fetch_header_line(parse_csv_uri("s3://drops/nope.csv")) + + +class _RangeHandler(http.server.BaseHTTPRequestHandler): + honour_range = True + + def do_GET(self): + if self.path == "/orders.csv": + body = CSV_BODY + rng = self.headers.get("Range") + if rng and self.honour_range: + start, end = rng.removeprefix("bytes=").split("-") + body = body[int(start):int(end) + 1] + self.send_response(206) + else: + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + else: + self.send_response(404) + self.end_headers() + + def log_message(self, *args): # keep test output clean + pass + + +@pytest.fixture(params=[True, False], ids=["range-honoured", "range-ignored"]) +def http_csv_server(request): + handler = type("H", (_RangeHandler,), {"honour_range": request.param}) + with socketserver.TCPServer(("127.0.0.1", 0), handler) as srv: + t = threading.Thread(target=srv.serve_forever, daemon=True) + t.start() + yield f"127.0.0.1:{srv.server_address[1]}" + srv.shutdown() + + +def test_https_fetch_header_line_with_and_without_range(http_csv_server): + uri = CsvUri(scheme="https", raw=f"http://{http_csv_server}/orders.csv") + header = HttpsCsvSourceReader().fetch_header_line(uri) + assert header == "order_id,amount,extra" + + +def test_https_fetch_full(http_csv_server, tmp_path): + uri = CsvUri(scheme="https", raw=f"http://{http_csv_server}/orders.csv") + dest = HttpsCsvSourceReader().fetch(uri, tmp_path / "o.csv") + assert dest.read_bytes() == CSV_BODY + + +def test_https_404_raises(http_csv_server): + uri = CsvUri(scheme="https", raw=f"http://{http_csv_server}/nope.csv") + with pytest.raises(Exception): + HttpsCsvSourceReader().fetch_header_line(uri) + + +def test_reader_for_dispatches_on_scheme(): + assert isinstance(reader_for(parse_csv_uri("s3://b/k")), S3CsvSourceReader) + assert isinstance( + reader_for(parse_csv_uri("https://x/y.csv")), HttpsCsvSourceReader) From 170f2553467521b74be33fcfb2d30e56d3d8a8d3 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 14:32:24 +0200 Subject: [PATCH 04/14] fix(csv-readers): https header extend + overflow, integration marker, extend/overflow tests Signed-off-by: Simone Carolini --- .../csv_readers/__init__.py | 10 ++- continuo_python_runtime/csv_readers/https.py | 36 ++++---- tests/conftest.py | 4 + tests/test_csv_readers_integration.py | 89 ++++++++++++++++--- 4 files changed, 108 insertions(+), 31 deletions(-) diff --git a/continuo_python_runtime/csv_readers/__init__.py b/continuo_python_runtime/csv_readers/__init__.py index 52e37a3..db65ae5 100644 --- a/continuo_python_runtime/csv_readers/__init__.py +++ b/continuo_python_runtime/csv_readers/__init__.py @@ -5,7 +5,13 @@ def reader_for(uri: CsvUri) -> CsvSourceReader: - """Composition edge: pick the adapter for the parsed scheme.""" + """Composition edge: pick the adapter for the parsed scheme. + + parse_csv_uri already constrains uri.scheme to "s3" or "https", but the + dispatch stays explicit (rather than an s3/else fallback) so a scheme + added to the parser without a matching adapter fails loudly here too.""" if uri.scheme == "s3": return S3CsvSourceReader() - return HttpsCsvSourceReader() + if uri.scheme == "https": + return HttpsCsvSourceReader() + raise ValueError(f"unsupported csv scheme: {uri.scheme!r}") diff --git a/continuo_python_runtime/csv_readers/https.py b/continuo_python_runtime/csv_readers/https.py index 6b2c533..4af5e1b 100644 --- a/continuo_python_runtime/csv_readers/https.py +++ b/continuo_python_runtime/csv_readers/https.py @@ -12,24 +12,30 @@ class HttpsCsvSourceReader(CsvSourceReader): - """Reads a csv source over HTTPS (public URLs; no auth in v1). A server - that ignores Range and answers 200 is handled by streaming and reading - only to the first newline before closing the connection.""" + """Reads a csv source over HTTPS (public URLs; no auth in v1). Mirrors + S3CsvSourceReader's probe-and-extend strategy: each ranged request is + independent, so a server that ignores Range and answers 200 with the + full body is handled too -- its first (unbounded) response already + contains the whole object, so the newline is found on the first pass.""" def fetch_header_line(self, uri: CsvUri) -> str: - req = urllib.request.Request( - uri.raw, headers={"Range": f"bytes=0-{HEADER_PROBE_BYTES - 1}"}) + start = 0 buf = b"" - with urllib.request.urlopen(req) as resp: # noqa: S310 — scheme gated by parse_csv_uri - while b"\n" not in buf: - chunk = resp.read(HEADER_PROBE_BYTES) - if not chunk: - break - buf += chunk - if len(buf) > MAX_HEADER_BYTES: - raise ValueError( - f"csv header line exceeds {MAX_HEADER_BYTES} bytes: {uri.raw}") - return buf.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") + while True: + end = start + HEADER_PROBE_BYTES - 1 + req = urllib.request.Request( + uri.raw, headers={"Range": f"bytes={start}-{end}"}) + with urllib.request.urlopen(req) as resp: # noqa: S310 — scheme gated by parse_csv_uri + body = resp.read() + buf += body + if b"\n" in buf: + return buf.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") + if len(body) < HEADER_PROBE_BYTES: # whole object read, no newline + return buf.rstrip(b"\r").decode("utf-8") + if len(buf) > MAX_HEADER_BYTES: + raise ValueError( + f"csv header line exceeds {MAX_HEADER_BYTES} bytes: {uri.raw}") + start += HEADER_PROBE_BYTES def fetch(self, uri: CsvUri, dest: Path) -> Path: with urllib.request.urlopen(uri.raw) as resp, open(dest, "wb") as f: # noqa: S310 diff --git a/tests/conftest.py b/tests/conftest.py index 238e779..1c4da6e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -215,6 +215,10 @@ def minio_container(): # credentials to boto3's own chain, so the chain needs something to # find. This is session-scoped (not monkeypatch, which is # function-scoped) -- set directly and restore on teardown. + # WARNING: any other test that runs while this fixture is active and + # makes a REAL (non-mocked, non-S3_ENDPOINT_URL-redirected) AWS call + # would authenticate with these fake minioadmin credentials, not the + # caller's real ones. prior_key = os.environ.get("AWS_ACCESS_KEY_ID") prior_secret = os.environ.get("AWS_SECRET_ACCESS_KEY") os.environ["AWS_ACCESS_KEY_ID"] = "minioadmin" diff --git a/tests/test_csv_readers_integration.py b/tests/test_csv_readers_integration.py index f2a4047..8a60445 100644 --- a/tests/test_csv_readers_integration.py +++ b/tests/test_csv_readers_integration.py @@ -13,14 +13,42 @@ import boto3 import pytest -from continuo_python_runtime.csv_source import CsvUri, parse_csv_uri +from continuo_python_runtime.csv_source import ( + HEADER_PROBE_BYTES, + MAX_HEADER_BYTES, + CsvUri, + parse_csv_uri, +) from continuo_python_runtime.csv_readers import reader_for from continuo_python_runtime.csv_readers.https import HttpsCsvSourceReader from continuo_python_runtime.csv_readers.s3 import S3CsvSourceReader +pytestmark = pytest.mark.integration + CSV_BODY = b"order_id,amount,extra\n1,10.5,x\n2,20.0,y\n" +def _long_header_body(header_len: int) -> tuple[bytes, str]: + """A csv body whose header line is exactly `header_len` bytes (no + newline inside it), followed by one short data row. Returns + (full_body, expected_header_str) so a test can assert on the exact + text a multi-probe fetch_header_line should reassemble.""" + chunk = "colname_padding_" + header = (chunk * (header_len // len(chunk) + 1))[:header_len] + return header.encode() + b"\n1,2,3\n", header + + +# 2.5x HEADER_PROBE_BYTES: a Range-honouring reader needs 3 probes to see +# the newline, so this is exactly the case FIX 1 (silent truncation to the +# first ~64KB window) would have gotten wrong. +LONG_HEADER_LEN = HEADER_PROBE_BYTES * 5 // 2 +LONG_HEADER_BODY, LONG_HEADER_STR = _long_header_body(LONG_HEADER_LEN) + +# No newline anywhere, longer than MAX_HEADER_BYTES: must raise, never hang +# or return a truncated line. +OVERFLOW_BODY = b"z" * (MAX_HEADER_BYTES + 100_000) + + @pytest.fixture(scope="session") def minio(minio_container): # minio_container: session fixture starting minio via docker endpoint, access, secret = minio_container @@ -30,6 +58,8 @@ def minio(minio_container): # minio_container: session fixture starting minio v ) client.create_bucket(Bucket="drops") client.put_object(Bucket="drops", Key="orders.csv", Body=CSV_BODY) + client.put_object(Bucket="drops", Key="long_header.csv", Body=LONG_HEADER_BODY) + client.put_object(Bucket="drops", Key="overflow.csv", Body=OVERFLOW_BODY) return endpoint @@ -52,25 +82,44 @@ def test_s3_missing_object_raises(minio, monkeypatch): S3CsvSourceReader().fetch_header_line(parse_csv_uri("s3://drops/nope.csv")) +def test_s3_fetch_header_line_extends_across_probes(minio, monkeypatch): + monkeypatch.setenv("S3_ENDPOINT_URL", minio) + header = S3CsvSourceReader().fetch_header_line( + parse_csv_uri("s3://drops/long_header.csv")) + assert header == LONG_HEADER_STR + + +def test_s3_fetch_header_line_overflow_raises(minio, monkeypatch): + monkeypatch.setenv("S3_ENDPOINT_URL", minio) + with pytest.raises(Exception): + S3CsvSourceReader().fetch_header_line(parse_csv_uri("s3://drops/overflow.csv")) + + class _RangeHandler(http.server.BaseHTTPRequestHandler): honour_range = True + _BODIES = { + "/orders.csv": CSV_BODY, + "/long_header.csv": LONG_HEADER_BODY, + "/overflow.csv": OVERFLOW_BODY, + } + def do_GET(self): - if self.path == "/orders.csv": - body = CSV_BODY - rng = self.headers.get("Range") - if rng and self.honour_range: - start, end = rng.removeprefix("bytes=").split("-") - body = body[int(start):int(end) + 1] - self.send_response(206) - else: - self.send_response(200) - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - else: + body = self._BODIES.get(self.path) + if body is None: self.send_response(404) self.end_headers() + return + rng = self.headers.get("Range") + if rng and self.honour_range: + start, end = rng.removeprefix("bytes=").split("-") + body = body[int(start):int(end) + 1] + self.send_response(206) + else: + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) def log_message(self, *args): # keep test output clean pass @@ -104,6 +153,18 @@ def test_https_404_raises(http_csv_server): HttpsCsvSourceReader().fetch_header_line(uri) +def test_https_fetch_header_line_extends_across_probes(http_csv_server): + uri = CsvUri(scheme="https", raw=f"http://{http_csv_server}/long_header.csv") + header = HttpsCsvSourceReader().fetch_header_line(uri) + assert header == LONG_HEADER_STR + + +def test_https_fetch_header_line_overflow_raises(http_csv_server): + uri = CsvUri(scheme="https", raw=f"http://{http_csv_server}/overflow.csv") + with pytest.raises(Exception): + HttpsCsvSourceReader().fetch_header_line(uri) + + def test_reader_for_dispatches_on_scheme(): assert isinstance(reader_for(parse_csv_uri("s3://b/k")), S3CsvSourceReader) assert isinstance( From 52f9095b72b99e972f875cd63618a3bdbcb2cb92 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 14:40:28 +0200 Subject: [PATCH 05/14] feat(contract): kind field with per-kind rules; python-csv shape enforced Signed-off-by: Simone Carolini --- continuo_python_runtime/contract/loader.py | 97 +++++++++++++++------- continuo_python_runtime/contract/model.py | 2 + tests/contract/test_loader.py | 59 +++++++++++++ tests/contract/test_model.py | 16 +++- 4 files changed, 142 insertions(+), 32 deletions(-) diff --git a/continuo_python_runtime/contract/loader.py b/continuo_python_runtime/contract/loader.py index 50ec010..3a57df1 100644 --- a/continuo_python_runtime/contract/loader.py +++ b/continuo_python_runtime/contract/loader.py @@ -18,9 +18,11 @@ from continuo_python_runtime.contract.model import ( CRITICALITIES, EXTRA_COLUMNS_POLICIES, + KINDS, Column, Node, ) +from continuo_python_runtime.csv_source import parse_csv_uri from continuo_python_runtime.errors import ContractError from continuo_python_runtime.types import parse_sql_type @@ -31,6 +33,7 @@ "owner", "schedule", "criticality", + "kind", "script", "extra_columns", "reads", @@ -39,7 +42,7 @@ "content_hash", } -_REQUIRED_STRING_FIELDS = ("schema", "table", "owner", "schedule", "script") +_REQUIRED_STRING_FIELDS = ("schema", "table", "owner", "schedule") _ALLOWED_OUTPUT_COLUMN_KEYS = {"name", "type", "nullable"} @@ -170,6 +173,12 @@ def parse_node( if unknown: raise ContractError(f"{label}: unknown key(s) {sorted(unknown)}") + kind = raw.get("kind", "python-model") + if kind not in KINDS: + raise ContractError( + f"{label}: 'kind' must be one of {sorted(KINDS)}, got {kind!r}" + ) + for field in _REQUIRED_STRING_FIELDS: value = raw.get(field) if not isinstance(value, str) or not value.strip(): @@ -181,7 +190,20 @@ def parse_node( table = raw["table"] owner = raw["owner"] schedule = raw["schedule"] - script = raw["script"] + + if kind == "python-csv": + if "script" in raw: + raise ContractError( + f"{label}: 'script' is forbidden for kind python-csv " + "(csv nodes are contract-only)" + ) + script = "" + else: + script = raw.get("script") + if not isinstance(script, str) or not script.strip(): + raise ContractError( + f"{label}: required field 'script' must be a non-empty string" + ) criticality = raw.get("criticality") if not isinstance(criticality, str) or criticality not in CRITICALITIES: @@ -200,39 +222,51 @@ def parse_node( ) reads = raw.get("reads") - if not isinstance(reads, dict) or not reads: - raise ContractError( - f"{label}: 'reads' must be a non-empty mapping of name -> SQL" - ) - for name, sql in reads.items(): - if not isinstance(name, str) or not name.strip(): - raise ContractError( - f"{label}: 'reads' name {name!r} must be a non-empty string" - ) - if not isinstance(sql, str) or not sql.strip(): + if kind == "python-csv": + if not isinstance(reads, dict) or set(reads) != {"csv"}: raise ContractError( - f"{label}: 'reads.{name}' must be a non-empty SQL string" + f"{label}: a python-csv node's 'reads' must be exactly " + "{csv: }" ) - if not check_reads: - continue try: - ensure_single_read(sql, dialect) - except (ValueError, TokenError) as exc: - # ensure_single_read's own message is phrased for check_binds - # (its only other caller today), so it's wrapped rather than - # surfaced bare here. TokenError is also caught: an unterminated - # string literal or comment fails sqlglot's tokenizer with a - # TokenError, a SqlglotError sibling of ParseError and not a - # subclass of ValueError -- despite ensure_single_read's - # docstring promising every rejection is a ValueError. Only - # TokenError, not the broader SqlglotError, is caught here: by - # the time control reaches this point `dialect` has already been - # validated once in load_contract_dir, so any other SqlglotError - # a future sqlglot version might raise from this call should - # surface as itself, not get relabeled as a rejected read. + parse_csv_uri(reads["csv"]) + except (ValueError, TypeError) as exc: + raise ContractError(f"{label}: invalid csv uri: {exc}") from exc + else: + if not isinstance(reads, dict) or not reads: raise ContractError( - f"{label}: 'reads.{name}' must be a single read query ({exc})" - ) from exc + f"{label}: 'reads' must be a non-empty mapping of name -> SQL" + ) + for name, sql in reads.items(): + if not isinstance(name, str) or not name.strip(): + raise ContractError( + f"{label}: 'reads' name {name!r} must be a non-empty string" + ) + if not isinstance(sql, str) or not sql.strip(): + raise ContractError( + f"{label}: 'reads.{name}' must be a non-empty SQL string" + ) + if not check_reads: + continue + try: + ensure_single_read(sql, dialect) + except (ValueError, TokenError) as exc: + # ensure_single_read's own message is phrased for check_binds + # (its only other caller today), so it's wrapped rather than + # surfaced bare here. TokenError is also caught: an unterminated + # string literal or comment fails sqlglot's tokenizer with a + # TokenError, a SqlglotError sibling of ParseError and not a + # subclass of ValueError -- despite ensure_single_read's + # docstring promising every rejection is a ValueError. Only + # TokenError, not the broader SqlglotError, is caught here: by + # the time control reaches this point `dialect` has already + # been validated once in load_contract_dir, so any other + # SqlglotError a future sqlglot version might raise from this + # call should surface as itself, not get relabeled as a + # rejected read. + raise ContractError( + f"{label}: 'reads.{name}' must be a single read query ({exc})" + ) from exc raw_columns = raw.get("output_columns") if not isinstance(raw_columns, list) or not raw_columns: @@ -297,6 +331,7 @@ def parse_node( extra_columns=extra_columns, config=config, content_hash=content_hash, + kind=kind, ) diff --git a/continuo_python_runtime/contract/model.py b/continuo_python_runtime/contract/model.py index cd344d5..8a0eb27 100644 --- a/continuo_python_runtime/contract/model.py +++ b/continuo_python_runtime/contract/model.py @@ -6,6 +6,7 @@ # Module-level constants CRITICALITIES = frozenset({"REGULATORY", "CORE", "SECONDARY"}) EXTRA_COLUMNS_POLICIES = frozenset({"raise", "warn"}) +KINDS = frozenset({"python-model", "python-csv"}) CONTRACT_VERSION = 1 @@ -34,6 +35,7 @@ class Node: extra_columns: str = "raise" config: dict[str, Any] = field(default_factory=dict) content_hash: str | None = None + kind: str = "python-model" @property def relation(self) -> str: diff --git a/tests/contract/test_loader.py b/tests/contract/test_loader.py index 21e9e34..45a1cbb 100644 --- a/tests/contract/test_loader.py +++ b/tests/contract/test_loader.py @@ -137,6 +137,65 @@ def test_unknown_key_error_names_source_and_node(): assert "analytics.t" in str(exc) +def _model_entry(**over): + entry = dict(VALID) + entry.update(over) + return entry + + +def _csv_entry(**over): + entry = { + "schema": "analytics", "table": "orders_csv", "owner": "team", + "schedule": "daily", "criticality": "SECONDARY", + "kind": "python-csv", + "reads": {"csv": "s3://drops/orders.csv"}, + "output_columns": [{"name": "order_id", "type": "INTEGER", "nullable": False}], + } + entry.update(over) + return entry + + +def test_kind_defaults_to_python_model(): + node = parse_node(_model_entry(), "f.yml") + assert node.kind == "python-model" + + +def test_csv_node_parses(): + node = parse_node(_csv_entry(), "f.yml", check_reads=False) + assert node.kind == "python-csv" + assert node.reads == {"csv": "s3://drops/orders.csv"} + assert node.script == "" + + +def test_csv_node_with_script_is_a_contract_error(): + with pytest.raises(ContractError, match="'script' is forbidden"): + parse_node(_csv_entry(script="scripts/x.py"), "f.yml") + + +def test_csv_node_reads_must_be_exactly_csv_uri(): + with pytest.raises(ContractError, match="exactly"): + parse_node(_csv_entry(reads={"csv": "s3://b/k", "other": "select 1"}), "f.yml") + with pytest.raises(ContractError, match="exactly"): + parse_node(_csv_entry(reads={"dwh": "s3://b/k"}), "f.yml") + + +def test_csv_node_bad_uri_is_a_contract_error(): + with pytest.raises(ContractError, match="csv uri"): + parse_node(_csv_entry(reads={"csv": "http://insecure/x.csv"}), "f.yml") + + +def test_unknown_kind_rejected(): + with pytest.raises(ContractError, match="kind"): + parse_node(_csv_entry(kind="python-parquet"), "f.yml") + + +def test_model_node_still_requires_script(): + entry = _model_entry() + del entry["script"] + with pytest.raises(ContractError, match="script"): + parse_node(entry, "f.yml") + + def test_bad_criticality_and_policy_and_type(): with pytest.raises(ContractError, match="criticality"): parse_node({**VALID, "criticality": "HIGH"}, "f.yml") diff --git a/tests/contract/test_model.py b/tests/contract/test_model.py index 8039419..9cef35c 100644 --- a/tests/contract/test_model.py +++ b/tests/contract/test_model.py @@ -1,4 +1,4 @@ -from continuo_python_runtime.contract.model import Column, Node +from continuo_python_runtime.contract.model import KINDS, Column, Node def _node(**over): @@ -29,3 +29,17 @@ def test_defaults_and_relation(): def test_config_accepts_nested_mapping(): n = _node(config={"indexes": [{"columns": ["id"], "unique": True}]}) assert n.config == {"indexes": [{"columns": ["id"], "unique": True}]} + + +def test_kind_defaults_to_python_model(): + n = _node() + assert n.kind == "python-model" + + +def test_kind_accepts_python_csv(): + n = _node(kind="python-csv") + assert n.kind == "python-csv" + + +def test_kinds_frozenset_contains_both_kinds(): + assert KINDS == frozenset({"python-model", "python-csv"}) From 1695854deb715b6f3200c5d9a4d6e03dc4ac8065 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 14:46:15 +0200 Subject: [PATCH 06/14] =?UTF-8?q?feat(merge):=20csv=20wire=20entries=20?= =?UTF-8?q?=E2=80=94=20kind=20on=20the=20wire,=20uri-based=20source=5Fhash?= =?UTF-8?q?,=20no=20closure?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Simone Carolini --- continuo_python_runtime/contract/merge.py | 20 ++++++--- tests/contract/test_merge.py | 55 ++++++++++++++++++++++- 2 files changed, 68 insertions(+), 7 deletions(-) diff --git a/continuo_python_runtime/contract/merge.py b/continuo_python_runtime/contract/merge.py index 6fdc62a..420923e 100644 --- a/continuo_python_runtime/contract/merge.py +++ b/continuo_python_runtime/contract/merge.py @@ -27,6 +27,7 @@ def node_entry(node: Node) -> dict: "owner": node.owner, "schedule": node.schedule, "criticality": node.criticality, + "kind": node.kind, "script": node.script, "reads": node.reads, "output_columns": [ @@ -128,12 +129,19 @@ def build_wire_contract( for node in nodes: entry = node_entry(node) - script_path = resolve_script_path(node.script, repo_root, context=node.relation) - script_bytes = script_path.read_bytes() - closure = resolve_closure(script_path, repo_root) - member_bytes = [member.read_bytes() for member in closure] - _lint_node_closure(node, repo_root, script_path, script_bytes, closure, member_bytes) - entry.update(hash_parts(entry, script_bytes, member_bytes)) + if node.kind == "python-csv": + # A csv node has no script and no import closure: its source IS + # the uri -- new file content at the same uri is new data, not a + # new node version. + uri_bytes = node.reads["csv"].encode() + entry.update(hash_parts(entry, uri_bytes, [])) + else: + script_path = resolve_script_path(node.script, repo_root, context=node.relation) + script_bytes = script_path.read_bytes() + closure = resolve_closure(script_path, repo_root) + member_bytes = [member.read_bytes() for member in closure] + _lint_node_closure(node, repo_root, script_path, script_bytes, closure, member_bytes) + entry.update(hash_parts(entry, script_bytes, member_bytes)) wire_nodes.append(entry) diff --git a/tests/contract/test_merge.py b/tests/contract/test_merge.py index 9148af9..6bcbf3e 100644 --- a/tests/contract/test_merge.py +++ b/tests/contract/test_merge.py @@ -1,5 +1,7 @@ """Tests for contract v1 merger.""" +from hashlib import sha256 + import yaml import pytest @@ -10,7 +12,7 @@ WIRE_ENTRY_KEYS = { "schema", "table", "owner", "schedule", "criticality", "script", "reads", - "output_columns", "description", "extra_columns", "config", + "output_columns", "description", "extra_columns", "config", "kind", "source_hash", "shared_code_hash", "config_hash", "content_hash", } @@ -428,3 +430,54 @@ def test_write_wire_contract_creates_missing_out_dir(contract_repo, tmp_path): write_wire_contract(doc, out) assert out.exists() assert yaml.safe_load(out.read_text())["service"] == "s" + + +def test_wire_entry_carries_kind(contract_repo): + repo = contract_repo + doc = build_wire_contract(repo / "contracts", repo, "s") + assert doc["nodes"][0]["kind"] == "python-model" + + +def test_csv_wire_entry_hashes_the_uri(tmp_path): + uri = "s3://drops/orders.csv" + (tmp_path / "contracts").mkdir() + (tmp_path / "contracts" / "c.yml").write_text( + "nodes:\n" + " - schema: analytics\n" + " table: orders_csv\n" + " owner: team\n" + " schedule: daily\n" + " criticality: SECONDARY\n" + " kind: python-csv\n" + f" reads: {{csv: {uri}}}\n" + " output_columns:\n" + " - {name: order_id, type: INTEGER, nullable: false}\n" + ) + doc = build_wire_contract(tmp_path / "contracts", tmp_path, "svc") + entry = doc["nodes"][0] + assert entry["kind"] == "python-csv" + assert entry["script"] == "" + assert entry["source_hash"] == sha256(uri.encode()).hexdigest() + assert entry["shared_code_hash"] == "" + assert entry["content_hash"] == content_hash_fold( + entry["source_hash"], "", entry["config_hash"]) + + +def test_csv_node_needs_no_script_file_on_disk(tmp_path): + # same fixture as above: note NO scripts/ dir exists — must not raise + uri = "s3://drops/orders.csv" + (tmp_path / "contracts").mkdir() + (tmp_path / "contracts" / "c.yml").write_text( + "nodes:\n" + " - schema: analytics\n" + " table: orders_csv\n" + " owner: team\n" + " schedule: daily\n" + " criticality: SECONDARY\n" + " kind: python-csv\n" + f" reads: {{csv: {uri}}}\n" + " output_columns:\n" + " - {name: order_id, type: INTEGER, nullable: false}\n" + ) + doc = build_wire_contract(tmp_path / "contracts", tmp_path, "svc") + assert len(doc["nodes"]) == 1 From eb2e3f1d70ca52a5ce01c3ce643b019c6c89b9c0 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 14:55:54 +0200 Subject: [PATCH 07/14] feat(harness): python-csv producer; run_node dispatches on node.kind Signed-off-by: Simone Carolini --- adapters/postgres/tests/conftest.py | 114 ++++++++++++++++++ .../test_integration_runtime_postgres.py | 79 ++++++++++++ continuo_python_runtime/csv_loader.py | 40 ++++++ continuo_python_runtime/harness.py | 25 ++-- tests/test_csv_loader.py | 56 +++++++++ tests/test_harness.py | 54 +++++++++ 6 files changed, 361 insertions(+), 7 deletions(-) create mode 100644 adapters/postgres/tests/conftest.py create mode 100644 continuo_python_runtime/csv_loader.py create mode 100644 tests/test_csv_loader.py diff --git a/adapters/postgres/tests/conftest.py b/adapters/postgres/tests/conftest.py new file mode 100644 index 0000000..058080c --- /dev/null +++ b/adapters/postgres/tests/conftest.py @@ -0,0 +1,114 @@ +"""Shared fixtures for adapters/postgres/tests. + +``adapters/postgres/tests`` is its own top-level pytest package (it carries +``__init__.py``, per the root pyproject's import-mode note), so it cannot see +fixtures declared in the root ``tests/conftest.py`` — pytest only walks +conftest files up a test's own directory ancestry, and this directory is not +an ancestor of the root ``tests/`` package. ``minio_container`` is duplicated +here rather than imported, matching the root fixture byte-for-byte in +behavior (real minio via a plain ``docker run``, dynamic host port, health +wait, ``docker rm -f`` teardown) so the postgres-adapter integration suite +gets the same "real backends, no stubs, no silent skip" guarantee without a +cross-package import. +""" + +import os +import subprocess +import time +import urllib.error +import urllib.request +import uuid + +import pytest + + +@pytest.fixture(scope="session") +def minio_container(): + """Session-scoped real minio backend, started via a plain `docker run`. + + No testcontainers/pytest-docker dependency in this repo, so this drives + docker directly. Publishes minio's 9000 to an EPHEMERAL host port (colima + may already hold 9000 for another stack) and waits for minio's own + /minio/health/live endpoint to return 200 before yielding. Any failure — + docker missing, `docker run` erroring, the health check never turning + green — raises so the dependent tests error loudly instead of silently + skipping, per this suite's "real backends, no stubs, no silent skip" + integration-testing policy. + """ + name = f"postgres-adapter-minio-{uuid.uuid4().hex[:12]}" + try: + subprocess.run( + [ + "docker", "run", "-d", "--name", name, + "-p", "0:9000", + "-e", "MINIO_ROOT_USER=minioadmin", + "-e", "MINIO_ROOT_PASSWORD=minioadmin", + "minio/minio:latest", + "server", "/data", "--address", ":9000", + ], + check=True, capture_output=True, text=True, timeout=60, + ) + except FileNotFoundError as exc: + raise RuntimeError( + "docker is not available; the postgres adapter's csv integration " + "test requires a live docker daemon" + ) from exc + except subprocess.CalledProcessError as exc: + raise RuntimeError(f"docker run for minio failed: {exc.stderr}") from exc + + try: + port_out = subprocess.run( + ["docker", "port", name, "9000/tcp"], + check=True, capture_output=True, text=True, timeout=10, + ).stdout.strip() + # e.g. "0.0.0.0:54321\n[::]:54321" -- take the first (ipv4) mapping. + host_port = port_out.splitlines()[0].rsplit(":", 1)[1] + endpoint = f"http://127.0.0.1:{host_port}" + + deadline = time.monotonic() + 30 + last_error: Exception | None = None + healthy = False + while time.monotonic() < deadline: + try: + with urllib.request.urlopen( + f"{endpoint}/minio/health/live", timeout=2 + ) as resp: + if resp.status == 200: + healthy = True + break + except (urllib.error.URLError, ConnectionError, TimeoutError) as exc: + last_error = exc + time.sleep(0.5) + if not healthy: + raise RuntimeError( + f"minio container {name} never became healthy at " + f"{endpoint}/minio/health/live: {last_error}" + ) + + # make_s3_client() forwards only S3_ENDPOINT_URL and otherwise leaves + # credentials to boto3's own chain, so the chain needs something to + # find. This is session-scoped (not monkeypatch, which is + # function-scoped) -- set directly and restore on teardown. + # WARNING: any other test that runs while this fixture is active and + # makes a REAL (non-mocked, non-S3_ENDPOINT_URL-redirected) AWS call + # would authenticate with these fake minioadmin credentials, not the + # caller's real ones. + prior_key = os.environ.get("AWS_ACCESS_KEY_ID") + prior_secret = os.environ.get("AWS_SECRET_ACCESS_KEY") + os.environ["AWS_ACCESS_KEY_ID"] = "minioadmin" + os.environ["AWS_SECRET_ACCESS_KEY"] = "minioadmin" + try: + yield (endpoint, "minioadmin", "minioadmin") + finally: + if prior_key is None: + os.environ.pop("AWS_ACCESS_KEY_ID", None) + else: + os.environ["AWS_ACCESS_KEY_ID"] = prior_key + if prior_secret is None: + os.environ.pop("AWS_SECRET_ACCESS_KEY", None) + else: + os.environ["AWS_SECRET_ACCESS_KEY"] = prior_secret + finally: + subprocess.run( + ["docker", "rm", "-f", name], capture_output=True, text=True, timeout=30 + ) diff --git a/adapters/postgres/tests/test_integration_runtime_postgres.py b/adapters/postgres/tests/test_integration_runtime_postgres.py index a15d248..26fd03e 100644 --- a/adapters/postgres/tests/test_integration_runtime_postgres.py +++ b/adapters/postgres/tests/test_integration_runtime_postgres.py @@ -7,10 +7,13 @@ import os import uuid +import boto3 import psycopg2 import pyarrow as pa import pytest +import yaml +from continuo_python_runtime.harness import run_node from continuo_python_runtime_postgres.adapter import PostgresAdapter PG = dict( @@ -303,3 +306,79 @@ def test_ensure_schema_generic_failure_rolls_back_and_releases_advisory_lock(): second.commit() second.close() assert acquired is True + + +# --- python-csv node end-to-end: real minio -> run_node -> real postgres --- +# +# minio_container is declared in adapters/postgres/tests/conftest.py (this +# package cannot see the root tests/conftest.py fixture of the same name -- +# see that conftest's module docstring). + +CSV_BODY = b"order_id,amount\n1,10.5\n2,20.0\n3,5.25\n" + + +@pytest.fixture(scope="session") +def csv_minio(minio_container): + """Seed a real minio bucket with the csv this test's node reads.""" + endpoint, access, secret = minio_container + client = boto3.client( + "s3", endpoint_url=endpoint, + aws_access_key_id=access, aws_secret_access_key=secret, + ) + client.create_bucket(Bucket="pg-drops") + client.put_object(Bucket="pg-drops", Key="orders.csv", Body=CSV_BODY) + return endpoint + + +def _csv_contract_dir(tmp_path, schema): + """A contract dir with a single python-csv node reading s3://pg-drops/orders.csv.""" + (tmp_path / "contracts").mkdir() + (tmp_path / "contracts" / "t.yml").write_text(yaml.safe_dump({"nodes": [{ + "schema": schema, "table": "orders_csv", "owner": "m", "schedule": "daily", + "criticality": "SECONDARY", "kind": "python-csv", + "reads": {"csv": "s3://pg-drops/orders.csv"}, + "output_columns": [ + {"name": "order_id", "type": "INTEGER", "nullable": False}, + {"name": "amount", "type": "DOUBLE PRECISION"}, + ], + }]})) + return tmp_path + + +@pytest.mark.integration +def test_run_node_csv_kind_loads_minio_csv_into_postgres(clean_schema, csv_minio, monkeypatch, tmp_path): + """run_node on a python-csv node fetches from real minio and writes to real postgres. + + Exercises the full production path added in this task: harness.run_node + dispatches on node.kind to csv_loader.produce_csv (no reader/adapter test + doubles here -- the S3CsvSourceReader from csv_readers.reader_for talks + to the real minio container, and PostgresAdapter writes to the real + postgres stack), then conform()/ensure_table()/load() proceed exactly as + for a python-model node. + """ + monkeypatch.setenv("S3_ENDPOINT_URL", csv_minio) + repo = _csv_contract_dir(tmp_path, clean_schema) + env = { + "NODE_ID": f"python-csv.svc.{clean_schema}.orders_csv", + "TABLE_NAME": "orders_csv", + "TARGET_SCHEMA": clean_schema, + "CONTRACT_DIR": str(repo / "contracts"), + "APP_ROOT": str(repo), + } + a = _adapter() + + assert run_node(env, adapter=a) == 0 + + assert _columns(clean_schema, "orders_csv") == [ + ("order_id", "integer", "NO"), + ("amount", "double precision", "YES"), + ] + assert _count(clean_schema, "orders_csv") == 3 + conn = _conn() + with conn.cursor() as cur: + cur.execute( + f'SELECT order_id, amount FROM "{clean_schema}"."orders_csv" ORDER BY order_id' + ) + rows = cur.fetchall() + conn.close() + assert rows == [(1, 10.5), (2, 20.0), (3, 5.25)] diff --git a/continuo_python_runtime/csv_loader.py b/continuo_python_runtime/csv_loader.py new file mode 100644 index 0000000..d59409f --- /dev/null +++ b/continuo_python_runtime/csv_loader.py @@ -0,0 +1,40 @@ +"""continuo_python_runtime/csv_loader.py + +Producer for python-csv nodes: materialize the declared table from the csv +source alone. Everything from conform() down (type coercion, extra_columns +policy, ensure_table, transactional load) is the existing harness path — +this module only turns the contract entry into a pyarrow Table. +""" +import logging +import tempfile +from pathlib import Path + +import pyarrow.csv # type: ignore[import-untyped] + +from continuo_python_runtime.contract.model import Node +from continuo_python_runtime.csv_readers import reader_for +from continuo_python_runtime.csv_source import CsvSourceReader, parse_csv_uri +from continuo_python_runtime.errors import LoadError + +logger = logging.getLogger("continuo_python_runtime.csv_loader") + + +def produce_csv(node: Node, reader: CsvSourceReader | None = None) -> "pyarrow.Table": + """Fetch node.reads['csv'] and parse it (RFC4180 defaults) into a Table. + + The caller conforms the result to output_columns exactly as for a script + node, so declared types — not csv inference — decide the warehouse schema. + """ + uri = parse_csv_uri(node.reads["csv"]) + active_reader = reader if reader is not None else reader_for(uri) + try: + with tempfile.TemporaryDirectory() as tmp: + dest = active_reader.fetch(uri, Path(tmp) / "source.csv") + table = pyarrow.csv.read_csv(dest) + except LoadError: + raise + except Exception as exc: + raise LoadError(f"csv fetch failed for {node.relation}: {exc}") from exc + logger.info("csv source %s: %d rows, columns=%s", + uri.raw, table.num_rows, table.column_names) + return table diff --git a/continuo_python_runtime/harness.py b/continuo_python_runtime/harness.py index b35bdfb..a703dce 100644 --- a/continuo_python_runtime/harness.py +++ b/continuo_python_runtime/harness.py @@ -31,6 +31,8 @@ from continuo_python_runtime.contract.loader import load_contract_dir from continuo_python_runtime.contract.model import Node from continuo_python_runtime.contract.paths import resolve_script_path +from continuo_python_runtime.csv_loader import produce_csv +from continuo_python_runtime.csv_source import CsvSourceReader from continuo_python_runtime.errors import ContractError, HarnessError, LoadError, ScriptError logger = logging.getLogger("continuo_python_runtime.harness") @@ -206,9 +208,16 @@ def _validate_config_early(adapter: Any, node: Node) -> None: ) from exc -def run_node(env: Mapping[str, str], adapter: Any = None) -> int: +def run_node( + env: Mapping[str, str], adapter: Any = None, reader: CsvSourceReader | None = None +) -> int: """Run a single node end-to-end and print exactly one sentinel result block. + ``reader`` mirrors the ``adapter`` injection seam: when given, it is + threaded into :func:`produce_csv` for a python-csv node instead of + letting that function pick a reader via ``reader_for``. Ignored for a + python-model node. + Returns 0 on success, 1 on any :class:`HarnessError`. """ node_id = env.get("NODE_ID") or "" @@ -246,13 +255,15 @@ def run_node(env: Mapping[str, str], adapter: Any = None) -> int: _validate_config_early(active_adapter, node) - with contextlib.redirect_stdout(sys.stderr): - module = load_script(node, app_root) - - ctx = RunContext(node, active_adapter) - raw_result = _execute_script(module, ctx) + if node.kind == "python-csv": + table = produce_csv(node, reader=reader) + else: + with contextlib.redirect_stdout(sys.stderr): + module = load_script(node, app_root) + ctx = RunContext(node, active_adapter) + raw_result = _execute_script(module, ctx) + table = to_arrow(raw_result) - table = to_arrow(raw_result) conformed = conform(table, node.output_columns, node.extra_columns) columns = [ diff --git a/tests/test_csv_loader.py b/tests/test_csv_loader.py new file mode 100644 index 0000000..2bfeaf8 --- /dev/null +++ b/tests/test_csv_loader.py @@ -0,0 +1,56 @@ +"""tests/test_csv_loader.py — unit tier: the loader through a local-file test +double of the PORT (the port is ours; substituting a test implementation of +our own abstraction is not stubbing an external service).""" +import pyarrow as pa +import pytest + +from continuo_python_runtime.contract.model import Column, Node +from continuo_python_runtime.csv_loader import produce_csv +from continuo_python_runtime.csv_source import CsvSourceReader +from continuo_python_runtime.errors import LoadError + + +class LocalFileReader(CsvSourceReader): + def __init__(self, body: bytes): + self.body = body + + def fetch_header_line(self, uri): + return self.body.split(b"\n", 1)[0].decode() + + def fetch(self, uri, dest): + dest.write_bytes(self.body) + return dest + + +def _csv_node(**over): + kw = dict( + schema="analytics", table="orders_csv", owner="t", schedule="daily", + criticality="SECONDARY", script="", kind="python-csv", + reads={"csv": "s3://drops/orders.csv"}, + output_columns=( + Column(name="order_id", type="INTEGER", nullable=False), + Column(name="amount", type="DOUBLE PRECISION"), + ), + extra_columns="warn", + ) + kw.update(over) + return Node(**kw) + + +def test_produce_csv_returns_arrow_table(): + table = produce_csv(_csv_node(), reader=LocalFileReader( + b"order_id,amount\n1,10.5\n2,20.0\n")) + assert isinstance(table, pa.Table) + assert table.num_rows == 2 + + +def test_produce_csv_fetch_failure_maps_to_load_error(): + class Broken(CsvSourceReader): + def fetch_header_line(self, uri): + raise OSError("boom") + + def fetch(self, uri, dest): + raise OSError("boom") + + with pytest.raises(LoadError, match="csv fetch failed"): + produce_csv(_csv_node(), reader=Broken()) diff --git a/tests/test_harness.py b/tests/test_harness.py index 5ea3e16..c807f0f 100644 --- a/tests/test_harness.py +++ b/tests/test_harness.py @@ -597,6 +597,60 @@ def test_ensure_import_paths_is_idempotent(tmp_path): assert sys.path.count(str(script_dir.resolve())) == 1 +# --- python-csv nodes: run_node dispatches on node.kind, never touches load_script --- + + +def test_run_node_on_csv_node_never_calls_load_script(monkeypatch, tmp_path, capsys): + """A python-csv node must be produced via produce_csv, not load_script/_execute_script.""" + import continuo_python_runtime.harness as harness_mod + from continuo_python_runtime.csv_source import CsvSourceReader + + class LocalFileReader(CsvSourceReader): + def __init__(self, body: bytes): + self.body = body + + def fetch_header_line(self, uri): + return self.body.split(b"\n", 1)[0].decode() + + def fetch(self, uri, dest): + dest.write_bytes(self.body) + return dest + + def _boom_if_called(*args, **kwargs): + raise AssertionError("load_script must not be called for a python-csv node") + + monkeypatch.setattr(harness_mod, "load_script", _boom_if_called) + + (tmp_path / "contracts").mkdir() + (tmp_path / "contracts" / "t.yml").write_text(yaml.safe_dump({"nodes": [{ + "schema": "analytics", "table": "orders_csv", "owner": "m", "schedule": "daily", + "criticality": "SECONDARY", "kind": "python-csv", + "reads": {"csv": "s3://drops/orders.csv"}, + "output_columns": [ + {"name": "order_id", "type": "INTEGER", "nullable": False}, + {"name": "amount", "type": "DOUBLE PRECISION"}, + ], + }]})) + + env = { + "NODE_ID": "python-csv.svc.analytics.orders_csv", + "TABLE_NAME": "orders_csv", + "TARGET_SCHEMA": "analytics", + "CONTRACT_DIR": str(tmp_path / "contracts"), + "APP_ROOT": str(tmp_path), + } + ad = FakeWarehouseAdapter() + reader = LocalFileReader(b"order_id,amount\n1,10.5\n2,20.0\n") + + assert run_node(env, adapter=ad, reader=reader) == 0 + out = capsys.readouterr().out + assert out.count("===CONTINUO_VALIDATION_RESULT_BEGIN===") == 1 + assert ad.ensured is not None + assert ad.loaded is not None + assert ad.loaded[0:2] == ("analytics", "orders_csv") + assert ad.loaded[2].num_rows == 2 + + def test_adapter_construction_failure_emits_single_load_error_block(monkeypatch, harness_repo, capsys): import continuo_python_runtime.harness as harness_mod From 13f06e1eab17c53148a1d182ddafac9c02d1d898 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 15:05:12 +0200 Subject: [PATCH 08/14] feat(validation): csv_source header check with extra-columns warning in build_from_columns Signed-off-by: Simone Carolini --- continuo_python_runtime/validation/runner.py | 34 ++++- tests/test_validation_runner.py | 132 +++++++++++++++++++ 2 files changed, 162 insertions(+), 4 deletions(-) diff --git a/continuo_python_runtime/validation/runner.py b/continuo_python_runtime/validation/runner.py index f49ead9..b93978f 100644 --- a/continuo_python_runtime/validation/runner.py +++ b/continuo_python_runtime/validation/runner.py @@ -8,10 +8,16 @@ - ``build_from_columns``: for python nodes, which have no SELECT to shape their output from. Fetch the node's validation spec JSON from S3 (``CANDIDATE_SPEC_URI`` — ``{"reads": [sql, ...], "output_columns": [{"name","type","nullable"}, ...], - "config": {...}}``; ``config`` is optional and defaults to ``{}``), bind-check every - declared read against the candidate schema so an upstream that - dropped a column the script reads fails the release gate, then materialize the - output table empty from the declared typed columns and the declared physical layout. + "config": {...}, "csv_source": "s3://..." | "https://..."}``; ``config`` and + ``csv_source`` are both optional. When ``csv_source`` is set, its header row is + fetched (without downloading the full object) and checked against the declared + output columns: a declared column missing from the header fails the release gate, + while a header column not declared only logs a ``csv_header_extra_columns`` + warning (it is silently dropped at load time). ``config`` defaults to ``{}``. + Every declared read is bind-checked against the candidate schema so an upstream + that dropped a column the script reads fails the release gate, then the output + table is materialized empty from the declared typed columns and the declared + physical layout. The engine adapter is discovered from the single installed ``continuo_engine.adapters`` entry point — each runner image installs exactly one. @@ -19,6 +25,7 @@ printed as its last line; all diagnostics go to stderr via the ``logging`` module. A non-zero exit marks the node failed. """ +import csv import json import logging import os @@ -29,6 +36,8 @@ AdapterDiscoveryError, discover_adapter, ) +from continuo_python_runtime.csv_readers import reader_for +from continuo_python_runtime.csv_source import check_header, parse_csv_uri from continuo_python_runtime.validation import s3 logger = logging.getLogger("validation_runner") @@ -126,6 +135,7 @@ def main() -> None: prod_schema = None spec: dict | None = None config: dict = {} + csv_source: str = "" if op in _NODE_OPS: table = _require("TABLE_NAME") unique_id = _node_id() or f"model.{table}" @@ -175,6 +185,12 @@ def main() -> None: print(result.result_block("error", msg, unique_id=unique_id), flush=True) sys.exit(2) config = raw_config or {} + csv_source = spec.get("csv_source", "") + if csv_source and not isinstance(csv_source, str): + msg = f"candidate spec 'csv_source' must be a string, got {type(csv_source).__name__}" + logger.error("%s", msg) + print(result.result_block("error", msg, unique_id=unique_id), flush=True) + sys.exit(2) else: prod_schema = _require("PROD_SCHEMA") elif op in _SCHEMA_OPS: @@ -220,6 +236,16 @@ def main() -> None: adapter.build_empty_from_sql(schema, table, candidate_sql) elif op == "build_from_columns": assert spec is not None, "spec must be set for build_from_columns" + if csv_source: + csv_uri = parse_csv_uri(csv_source) + header_line = reader_for(csv_uri).fetch_header_line(csv_uri) + declared = [c["name"] for c in spec["output_columns"]] + extras = check_header(next(csv.reader([header_line])), declared) + if extras: + logger.warning( + "csv_header_extra_columns node=%s columns=%s — columns present in the " + "csv but not declared in output_columns; they will not be loaded", + unique_id, sorted(extras)) for read_sql in spec.get("reads", []): adapter.check_binds(read_sql) adapter.build_empty_from_columns(schema, table, spec["output_columns"], config) diff --git a/tests/test_validation_runner.py b/tests/test_validation_runner.py index a619dc3..f76de09 100644 --- a/tests/test_validation_runner.py +++ b/tests/test_validation_runner.py @@ -1,4 +1,5 @@ """Unit tests for the runner — hand-written fakes, no live DB or S3.""" +import boto3 import pytest from continuo_engine_contract import result @@ -648,6 +649,137 @@ def test_main_build_from_columns_adapter_config_rejection_fails_the_gate(monkeyp assert "sortkey" in out +# -------------------------------------------------------------------------- +# main — build_from_columns csv_source header check (python-csv-nodes A6) +# -------------------------------------------------------------------------- + +CSV_HEADER_WITH_EXTRA_BODY = b"order_id,amount,extra\n1,10.5,x\n2,20.0,y\n" +CSV_HEADER_MISSING_COLUMN_BODY = b"order_id,amount\n1,10.5\n2,20.0\n" + + +@pytest.fixture(scope="session") +def csv_source_bucket(minio_container): + """Real minio-backed csv fixtures for the csv_source header-check tests. + + Uploads two objects into a bucket dedicated to this module (kept separate + from test_csv_readers_integration.py's own ``drops`` bucket so the two + modules' session-scoped setup never race each other): a csv whose header + has an extra undeclared column, and one missing a declared column. + ``nope.csv`` is deliberately never uploaded, so a test can point + csv_source at it to exercise the unreachable-source path against the + real minio backend. + """ + endpoint, access, secret = minio_container + client = boto3.client( + "s3", endpoint_url=endpoint, + aws_access_key_id=access, aws_secret_access_key=secret, + ) + client.create_bucket(Bucket="csv-validation") + client.put_object(Bucket="csv-validation", Key="orders.csv", Body=CSV_HEADER_WITH_EXTRA_BODY) + client.put_object( + Bucket="csv-validation", Key="orders_missing_col.csv", + Body=CSV_HEADER_MISSING_COLUMN_BODY, + ) + return endpoint + + +@pytest.mark.integration +def test_main_build_from_columns_with_csv_source_checks_header( + monkeypatch, capsys, caplog, csv_source_bucket +): + """csv_source header conformance against a real minio object: declared + columns are all present, and the csv's extra undeclared column ("extra") + logs a structured warning but does not block the build.""" + _set_common_env(monkeypatch) + monkeypatch.setenv("VALIDATION_OP", "build_from_columns") + monkeypatch.setenv("S3_ENDPOINT_URL", csv_source_bucket) + fake = FakeWarehouseAdapter() + _install_fake_adapter(monkeypatch, fake) + columns = [ + {"name": "order_id", "type": "BIGINT", "nullable": False}, + {"name": "amount", "type": "DOUBLE PRECISION", "nullable": True}, + ] + spec = _spec(columns=columns) + spec["csv_source"] = "s3://csv-validation/orders.csv" + monkeypatch.setattr(runner, "load_candidate_spec", lambda: spec) + + with caplog.at_level("WARNING"): + runner.main() + + assert fake.column_builds == [("_candidate_relA", "orders", columns, {})] + out = capsys.readouterr().out + assert '"status":"success"' in out + assert "csv_header_extra_columns" in caplog.text + assert "extra" in caplog.text + + +@pytest.mark.integration +def test_main_build_from_columns_csv_header_missing_column_fails( + monkeypatch, capsys, csv_source_bucket +): + """A declared column absent from the real csv header fails the release + gate: exit 1, error block names the missing column.""" + _set_common_env(monkeypatch) + monkeypatch.setenv("VALIDATION_OP", "build_from_columns") + monkeypatch.setenv("S3_ENDPOINT_URL", csv_source_bucket) + fake = FakeWarehouseAdapter() + _install_fake_adapter(monkeypatch, fake) + columns = [ + {"name": "order_id", "type": "BIGINT", "nullable": False}, + {"name": "customer_id", "type": "BIGINT", "nullable": False}, + ] + spec = _spec(columns=columns) + spec["csv_source"] = "s3://csv-validation/orders_missing_col.csv" + monkeypatch.setattr(runner, "load_candidate_spec", lambda: spec) + + with pytest.raises(SystemExit) as exc: + runner.main() + + assert exc.value.code == 1 + assert fake.column_builds == [] # header check blocks the build + out = capsys.readouterr().out + assert '"status":"error"' in out + assert "missing declared column" in out + + +@pytest.mark.integration +def test_main_build_from_columns_unreachable_csv_fails(monkeypatch, capsys, csv_source_bucket): + """An unreachable csv_source blocks promotion: exit 1, error block emitted.""" + _set_common_env(monkeypatch) + monkeypatch.setenv("VALIDATION_OP", "build_from_columns") + monkeypatch.setenv("S3_ENDPOINT_URL", csv_source_bucket) + fake = FakeWarehouseAdapter() + _install_fake_adapter(monkeypatch, fake) + spec = _spec() + spec["csv_source"] = "s3://csv-validation/nope.csv" + monkeypatch.setattr(runner, "load_candidate_spec", lambda: spec) + + with pytest.raises(SystemExit) as exc: + runner.main() + + assert exc.value.code == 1 + assert fake.column_builds == [] + out = capsys.readouterr().out + assert '"status":"error"' in out + + +def test_main_build_from_columns_without_csv_source_unchanged(monkeypatch, capsys): + """No csv_source key in the spec: behavior is unchanged from before A6 — + no header fetch is attempted, and the build proceeds straight through.""" + _set_common_env(monkeypatch) + monkeypatch.setenv("VALIDATION_OP", "build_from_columns") + monkeypatch.setenv("CANDIDATE_SPEC_URI", "s3://continuo/candidate-spec/rel-1/svc.orders.json") + fake = FakeWarehouseAdapter() + _install_fake_adapter(monkeypatch, fake) + columns = [{"name": "id", "type": "BIGINT", "nullable": False}] + monkeypatch.setattr(runner, "load_candidate_spec", lambda: _spec([], columns)) + + runner.main() + + assert fake.column_builds == [("_candidate_relA", "orders", columns, {})] + assert '"status":"success"' in capsys.readouterr().out + + # -------------------------------------------------------------------------- # main — sentinel-block invariant across every block-emitting exit path # -------------------------------------------------------------------------- From 1e7498fa88f1ca04a17775b4046e1633d0b37a18 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 15:13:23 +0200 Subject: [PATCH 09/14] docs(template): flat csv contract example and author docs Signed-off-by: Simone Carolini --- README.md | 28 +++++++++++++++++++++- continuo_python_runtime/contract/loader.py | 5 ++-- template/contracts/example_csv.yml | 17 +++++++++++++ tests/test_template.py | 14 +++++++---- 4 files changed, 56 insertions(+), 8 deletions(-) create mode 100644 template/contracts/example_csv.yml diff --git a/README.md b/README.md index a70223f..f19e271 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,9 @@ the Go parser has not been taught is a production outage, not a refactor. login` step to `release.yml`. 5. Write a contract file under `contracts/` (see `template/contracts/example.yml`) and a script under `scripts/` that - implements `run(ctx)` (see `template/scripts/example.py`). + implements `run(ctx)` (see `template/scripts/example.py`). A node that + only needs to land a csv file needs no script at all — see + `template/contracts/example_csv.yml` and "Node kinds" below. 6. Push to `main`. The `release.yml` workflow lints the scripts, validates and merges the contracts, builds and pushes the image, uploads the merged contract to S3, and POSTs the release. @@ -121,6 +123,30 @@ The runtime image does not re-run this gate, so a read that passes here is not re-judged under a different grammar in production. See `docs/boundary-contract.md` §13.1. +## Node kinds + +A contract node's `kind:` field selects how the node produces its rows. +Every rule below (`extra_columns`, `output_columns`, "Conform rules") applies +to both kinds identically — `kind` only changes how the pre-conform table is +produced, never how it is checked or written. + +- **`python-model`** (the default; the field may be omitted) — a script node. + It requires `script:` and a `reads:` map of one or more named SQL queries, + as described in "The script API" below. +- **`python-csv`** — a contract-only node: it has no script and its `reads:` + map must be exactly `{csv: }`, where the uri is `s3://bucket/key` or + an `https://` url (`http://` is rejected at validate time, not run time). + The harness fetches the file, parses it with RFC 4180 defaults, and feeds + the result straight into `conform()` — declared `output_columns` types + decide the warehouse schema, not whatever pyarrow infers from the csv. + Because there is no script, `script:` is a forbidden key for this kind; + `continuo-runtime validate`/`merge`/`lint` reject one that sets it. The + csv's header row must contain every declared output column (checked again, + independently, at release time before promotion); columns present in the + header but not declared are governed by the same `extra_columns` policy as + a script node's output — `raise` (default) fails the run, `warn` drops + them and logs a warning. See `template/contracts/example_csv.yml`. + ## The script API A node script is a Python file with exactly one required entry point: diff --git a/continuo_python_runtime/contract/loader.py b/continuo_python_runtime/contract/loader.py index 3a57df1..81a07d4 100644 --- a/continuo_python_runtime/contract/loader.py +++ b/continuo_python_runtime/contract/loader.py @@ -199,11 +199,12 @@ def parse_node( ) script = "" else: - script = raw.get("script") - if not isinstance(script, str) or not script.strip(): + raw_script = raw.get("script") + if not isinstance(raw_script, str) or not raw_script.strip(): raise ContractError( f"{label}: required field 'script' must be a non-empty string" ) + script = raw_script criticality = raw.get("criticality") if not isinstance(criticality, str) or criticality not in CRITICALITIES: diff --git a/template/contracts/example_csv.yml b/template/contracts/example_csv.yml new file mode 100644 index 0000000..d69003e --- /dev/null +++ b/template/contracts/example_csv.yml @@ -0,0 +1,17 @@ +nodes: + - schema: analytics + table: example_csv + description: "Contract-only node: loads a csv from the declared uri" + owner: your-team + schedule: daily + criticality: SECONDARY + kind: python-csv + # A csv node has no script and exactly one read: the source uri. + # s3://bucket/key and https:// urls are accepted; the file's header row + # must contain every declared output column (extra columns are dropped + # per extra_columns and surfaced as a warning). + reads: + csv: s3://your-bucket/drops/example.csv + output_columns: + - {name: order_id, type: INTEGER, nullable: false} + - {name: amount, type: "DOUBLE PRECISION", nullable: true} diff --git a/tests/test_template.py b/tests/test_template.py index e40b2eb..f6e9c24 100644 --- a/tests/test_template.py +++ b/tests/test_template.py @@ -20,14 +20,18 @@ def test_template_passes_lint_validate_merge(tmp_path): def test_template_passes_hash(capsys): - """Template must also pass the hash subcommand, one tab-separated sha256 line.""" + """Template must also pass the hash subcommand: one tab-separated sha256 + line per contract node — the sql-node example and the csv-node example.""" assert main(["hash", str(TEMPLATE / "contracts"), "--repo-root", str(TEMPLATE)]) == 0 out = capsys.readouterr().out lines = [line for line in out.splitlines() if line] - assert len(lines) == 1 - relation, hash_value = lines[0].split("\t") - assert relation == "analytics.example" - assert hash_value.startswith("sha256:") + assert len(lines) == 2 + by_relation = {} + for line in lines: + relation, hash_value = line.split("\t") + assert hash_value.startswith("sha256:") + by_relation[relation] = hash_value + assert set(by_relation) == {"analytics.example", "analytics.example_csv"} def test_template_demonstrates_multiple_named_reads(): From 2c1946fef7079c0a3f8cb5bddef75ee5cc5f205a Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 15:35:57 +0200 Subject: [PATCH 10/14] fix(csv): range-ignored header loop, empty-source guard, structured run warning; docs(boundary): kind field Final fix wave from the Phase-A python-csv whole-branch review: - FIX 1: HttpsCsvSourceReader.fetch_header_line no longer loops forever re-appending an identical full body when a server ignores Range and answers 200. It now checks the response status: a 200 (not 206) is terminal on the first pass regardless of size or newline presence, fixing a false MAX_HEADER_BYTES overflow for a no-newline body sized between HEADER_PROBE_BYTES and MAX_HEADER_BYTES. - FIX 6: validation/runner.py raises a clear ValueError naming the csv_source when fetch_header_line returns an empty header line, instead of an opaque StopIteration from csv.reader. Discovered while writing the real-minio test that a 0-byte S3 object's first Range probe raises botocore InvalidRange rather than returning an empty body, so S3CsvSourceReader.fetch_header_line now also treats an InvalidRange error on the first probe as an empty object. - FIX 7: csv_loader.produce_csv now logs the same structured csv_header_extra_columns warning the validation runner emits, giving extra_columns: drop spec parity between the validation and run paths. - FIX 2: collapse the "reads must be exactly {csv: }" error message into one f-string; message text unchanged. - FIX 3: add a test covering the existing exit-2 guard for a non-string csv_source in the candidate spec. - docs(boundary-contract): document the additive `kind` wire field (default python-model; python-csv is a contract-only node whose reads is exactly {csv: } and whose source_hash is sha256 of that uri). Signed-off-by: Simone Carolini --- continuo_python_runtime/contract/loader.py | 3 +- continuo_python_runtime/csv_loader.py | 12 ++++ continuo_python_runtime/csv_readers/https.py | 23 ++++++- continuo_python_runtime/csv_readers/s3.py | 18 +++++- continuo_python_runtime/validation/runner.py | 5 ++ docs/boundary-contract.md | 25 +++++++- tests/test_csv_loader.py | 25 ++++++++ tests/test_csv_readers_integration.py | 20 +++++++ tests/test_validation_runner.py | 63 +++++++++++++++++--- 9 files changed, 176 insertions(+), 18 deletions(-) diff --git a/continuo_python_runtime/contract/loader.py b/continuo_python_runtime/contract/loader.py index 81a07d4..b3b1032 100644 --- a/continuo_python_runtime/contract/loader.py +++ b/continuo_python_runtime/contract/loader.py @@ -226,8 +226,7 @@ def parse_node( if kind == "python-csv": if not isinstance(reads, dict) or set(reads) != {"csv"}: raise ContractError( - f"{label}: a python-csv node's 'reads' must be exactly " - "{csv: }" + f"{label}: a python-csv node's 'reads' must be exactly {{csv: }}" ) try: parse_csv_uri(reads["csv"]) diff --git a/continuo_python_runtime/csv_loader.py b/continuo_python_runtime/csv_loader.py index d59409f..b581a42 100644 --- a/continuo_python_runtime/csv_loader.py +++ b/continuo_python_runtime/csv_loader.py @@ -35,6 +35,18 @@ def produce_csv(node: Node, reader: CsvSourceReader | None = None) -> "pyarrow.T raise except Exception as exc: raise LoadError(f"csv fetch failed for {node.relation}: {exc}") from exc + declared = {col.name for col in node.output_columns} + extras = set(table.column_names) - declared + if extras: + # Spec parity with the validation runner's csv_source header check + # (continuo_python_runtime/validation/runner.py): extra_columns: drop + # silently discards these at conform() time, so this structured + # warning is the only place the RUN path surfaces which columns were + # dropped. + logger.warning( + "csv_header_extra_columns node=%s columns=%s — columns present in the " + "csv but not declared in output_columns; they will not be loaded", + node.relation, sorted(extras)) logger.info("csv source %s: %d rows, columns=%s", uri.raw, table.num_rows, table.column_names) return table diff --git a/continuo_python_runtime/csv_readers/https.py b/continuo_python_runtime/csv_readers/https.py index 4af5e1b..5fafb44 100644 --- a/continuo_python_runtime/csv_readers/https.py +++ b/continuo_python_runtime/csv_readers/https.py @@ -14,9 +14,11 @@ class HttpsCsvSourceReader(CsvSourceReader): """Reads a csv source over HTTPS (public URLs; no auth in v1). Mirrors S3CsvSourceReader's probe-and-extend strategy: each ranged request is - independent, so a server that ignores Range and answers 200 with the - full body is handled too -- its first (unbounded) response already - contains the whole object, so the newline is found on the first pass.""" + independent, so a server that ignores Range and answers 200 (not 206) + is handled too -- its response is the whole object, so it is treated + as terminal on the first pass regardless of whether it contains a + newline or how large it is, rather than being re-fetched and + re-appended pass after pass.""" def fetch_header_line(self, uri: CsvUri) -> str: start = 0 @@ -27,6 +29,21 @@ def fetch_header_line(self, uri: CsvUri) -> str: uri.raw, headers={"Range": f"bytes={start}-{end}"}) with urllib.request.urlopen(req) as resp: # noqa: S310 — scheme gated by parse_csv_uri body = resp.read() + range_honoured = resp.status == 206 + if not range_honoured: + # The server ignored our Range header and returned the entire + # object (status 200), not just the requested window -- *body* + # is therefore the whole object and this response is terminal, + # regardless of its size relative to HEADER_PROBE_BYTES. Every + # retry would re-fetch the identical full body, so looping + # would only re-append it pass after pass and eventually trip + # a false MAX_HEADER_BYTES overflow. + if b"\n" in body: + return body.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") + if len(body) > MAX_HEADER_BYTES: + raise ValueError( + f"csv header line exceeds {MAX_HEADER_BYTES} bytes: {uri.raw}") + return body.rstrip(b"\r").decode("utf-8") buf += body if b"\n" in buf: return buf.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") diff --git a/continuo_python_runtime/csv_readers/s3.py b/continuo_python_runtime/csv_readers/s3.py index 4a7ba3e..2739848 100644 --- a/continuo_python_runtime/csv_readers/s3.py +++ b/continuo_python_runtime/csv_readers/s3.py @@ -1,6 +1,8 @@ """continuo_python_runtime/csv_readers/s3.py""" from pathlib import Path +from botocore.exceptions import ClientError # type: ignore[import-untyped] + from continuo_python_runtime.csv_source import ( HEADER_PROBE_BYTES, MAX_HEADER_BYTES, @@ -21,9 +23,19 @@ def fetch_header_line(self, uri: CsvUri) -> str: buf = b"" while True: end = start + HEADER_PROBE_BYTES - 1 - body = client.get_object( - Bucket=uri.bucket, Key=uri.key, Range=f"bytes={start}-{end}" - )["Body"].read() + try: + body = client.get_object( + Bucket=uri.bucket, Key=uri.key, Range=f"bytes={start}-{end}" + )["Body"].read() + except ClientError as exc: + if start == 0 and exc.response.get("Error", {}).get("Code") == "InvalidRange": + # A Range request on byte 0 is unsatisfiable only when the + # object itself is 0 bytes long: treat a 0-byte csv source + # as an empty header line rather than a hard failure -- + # the caller (validation/runner.py) raises a clear error + # for an empty header line. + return "" + raise buf += body if b"\n" in buf: return buf.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") diff --git a/continuo_python_runtime/validation/runner.py b/continuo_python_runtime/validation/runner.py index b93978f..37a1174 100644 --- a/continuo_python_runtime/validation/runner.py +++ b/continuo_python_runtime/validation/runner.py @@ -239,6 +239,11 @@ def main() -> None: if csv_source: csv_uri = parse_csv_uri(csv_source) header_line = reader_for(csv_uri).fetch_header_line(csv_uri) + if not header_line: + raise ValueError( + "csv source has no header line (empty or unreadable): " + f"{csv_source}" + ) declared = [c["name"] for c in spec["output_columns"]] extras = check_header(next(csv.reader([header_line])), declared) if extras: diff --git a/docs/boundary-contract.md b/docs/boundary-contract.md index 1f658d0..6868633 100644 --- a/docs/boundary-contract.md +++ b/docs/boundary-contract.md @@ -22,13 +22,36 @@ s3://///contract.yaml - Schema per §3: `contract_version: 1`, `service`, `nodes: [...]` — every node's wire entry carries `schema`, `table`, `owner`, `schedule`, - `criticality`, `script`, `reads`, `output_columns`, `description`, + `criticality`, `kind`, `script`, `reads`, `output_columns`, `description`, `extra_columns`, the four hash fields (`source_hash`, `shared_code_hash`, `config_hash`, `content_hash`), and `config` (physical layout — see below), which is present on every wire entry, defaulting to `{}` when the author's contract file didn't declare one. "Optional" describes the contract *file* an author writes, not the merged wire artifact, where every field above is always present. +- **`kind`** is additive: it defaults to `python-model` (a script-backed + node, everything described above and in §13.4). `python-csv` is the one + other value — a **contract-only** node with no script: it declares no + Python to run, and its `reads` must be exactly `{csv: }` (one key, + named `csv`, whose value is an `s3://bucket/key` or `https://...` URI — + anything else fails `ContractError` at merge time). Because there is no + script or import closure to hash, `source_hash` is instead + `sha256(uri_bytes)` over that same `csv` URI (`shared_code_hash` is `""`, + the empty-closure value), so a `python-csv` node's `content_hash` changes + when the URI itself changes, never when the data at that URI changes. The + runtime harness's dispatch on `run_node` reads `kind` to skip script + execution entirely for a `python-csv` node and instead fetch and parse + the csv (`csv_loader.produce_csv` in this repo), which the harness then + conforms to the declared `output_columns` exactly as it would a script + node's returned dataframe (§13.4's "Sole write sink" paragraph, unchanged + from that point on). The validation runner mirrors the same idea at + release-validation time: its `build_from_columns` op checks the csv's + header line — fetched via a `csv_source` key on the candidate spec, + without downloading the full object — against the declared columns + instead of bind-checking a SELECT, so a `python-csv` node still fails the + release gate on a missing declared column exactly as a script node fails + it on a missing bind. §13.4 below still applies unchanged to + `kind: python-model` nodes. - **Reads must be single-statement SELECTs with every table reference schema-qualified** (`analytics.table_a`, never `table_a`) — the resolver raises `UnqualifiedTableReference` and rejects the whole release otherwise diff --git a/tests/test_csv_loader.py b/tests/test_csv_loader.py index 2bfeaf8..06d7cff 100644 --- a/tests/test_csv_loader.py +++ b/tests/test_csv_loader.py @@ -54,3 +54,28 @@ def fetch(self, uri, dest): with pytest.raises(LoadError, match="csv fetch failed"): produce_csv(_csv_node(), reader=Broken()) + + +def test_produce_csv_logs_structured_warning_for_undeclared_columns(caplog): + """FIX 7 — spec parity: when the csv carries a column not declared in + output_columns, the RUN path logs the same structured + csv_header_extra_columns warning the validation runner emits, so + extra_columns: drop's silent-discard behavior is observable in both + places, not just at validation time.""" + node = _csv_node() # declares only order_id, amount + with caplog.at_level("WARNING"): + produce_csv(node, reader=LocalFileReader( + b"order_id,amount,extra\n1,10.5,x\n2,20.0,y\n")) + + assert "csv_header_extra_columns" in caplog.text + assert node.relation in caplog.text + assert "extra" in caplog.text + + +def test_produce_csv_no_warning_when_all_columns_declared(caplog): + """No extras: no csv_header_extra_columns warning is logged.""" + with caplog.at_level("WARNING"): + produce_csv(_csv_node(), reader=LocalFileReader( + b"order_id,amount\n1,10.5\n2,20.0\n")) + + assert "csv_header_extra_columns" not in caplog.text diff --git a/tests/test_csv_readers_integration.py b/tests/test_csv_readers_integration.py index 8a60445..cba05ab 100644 --- a/tests/test_csv_readers_integration.py +++ b/tests/test_csv_readers_integration.py @@ -48,6 +48,14 @@ def _long_header_body(header_len: int) -> tuple[bytes, str]: # or return a truncated line. OVERFLOW_BODY = b"z" * (MAX_HEADER_BYTES + 100_000) +# No newline anywhere, sized strictly between HEADER_PROBE_BYTES and +# MAX_HEADER_BYTES. A server that ignores Range and re-serves this whole +# body on every retry must recognize its first response as final (via the +# response status) rather than re-appending it pass after pass, which would +# otherwise double the buffer past MAX_HEADER_BYTES and raise a false +# overflow (FIX 1). +NO_NEWLINE_MID_BODY = b"a" * 100_000 + @pytest.fixture(scope="session") def minio(minio_container): # minio_container: session fixture starting minio via docker @@ -102,6 +110,7 @@ class _RangeHandler(http.server.BaseHTTPRequestHandler): "/orders.csv": CSV_BODY, "/long_header.csv": LONG_HEADER_BODY, "/overflow.csv": OVERFLOW_BODY, + "/no_newline.csv": NO_NEWLINE_MID_BODY, } def do_GET(self): @@ -165,6 +174,17 @@ def test_https_fetch_header_line_overflow_raises(http_csv_server): HttpsCsvSourceReader().fetch_header_line(uri) +def test_https_fetch_header_line_no_newline_mid_size_is_not_a_false_overflow(http_csv_server): + """A no-newline body sized between HEADER_PROBE_BYTES and MAX_HEADER_BYTES + must come back as the header line verbatim, on both a Range-honouring + server (multi-probe extend, whole object read on the short final probe) + and a Range-ignoring one (FIX 1: the 200 response is terminal on the + first pass, never re-appended).""" + uri = CsvUri(scheme="https", raw=f"http://{http_csv_server}/no_newline.csv") + header = HttpsCsvSourceReader().fetch_header_line(uri) + assert header == NO_NEWLINE_MID_BODY.decode("utf-8") + + def test_reader_for_dispatches_on_scheme(): assert isinstance(reader_for(parse_csv_uri("s3://b/k")), S3CsvSourceReader) assert isinstance( diff --git a/tests/test_validation_runner.py b/tests/test_validation_runner.py index f76de09..267679d 100644 --- a/tests/test_validation_runner.py +++ b/tests/test_validation_runner.py @@ -661,13 +661,13 @@ def test_main_build_from_columns_adapter_config_rejection_fails_the_gate(monkeyp def csv_source_bucket(minio_container): """Real minio-backed csv fixtures for the csv_source header-check tests. - Uploads two objects into a bucket dedicated to this module (kept separate + Uploads objects into a bucket dedicated to this module (kept separate from test_csv_readers_integration.py's own ``drops`` bucket so the two modules' session-scoped setup never race each other): a csv whose header - has an extra undeclared column, and one missing a declared column. - ``nope.csv`` is deliberately never uploaded, so a test can point - csv_source at it to exercise the unreachable-source path against the - real minio backend. + has an extra undeclared column, one missing a declared column, and a + 0-byte object (FIX 6 — empty/no-header source). ``nope.csv`` is + deliberately never uploaded, so a test can point csv_source at it to + exercise the unreachable-source path against the real minio backend. """ endpoint, access, secret = minio_container client = boto3.client( @@ -680,6 +680,7 @@ def csv_source_bucket(minio_container): Bucket="csv-validation", Key="orders_missing_col.csv", Body=CSV_HEADER_MISSING_COLUMN_BODY, ) + client.put_object(Bucket="csv-validation", Key="empty.csv", Body=b"") return endpoint @@ -763,6 +764,33 @@ def test_main_build_from_columns_unreachable_csv_fails(monkeypatch, capsys, csv_ assert '"status":"error"' in out +@pytest.mark.integration +def test_main_build_from_columns_empty_csv_source_gives_legible_error( + monkeypatch, capsys, csv_source_bucket +): + """A 0-byte csv_source (fetch_header_line returns "") must not surface as + an opaque StopIteration from csv.reader: exit 1, error block names the + empty/no-header source (FIX 6).""" + _set_common_env(monkeypatch) + monkeypatch.setenv("VALIDATION_OP", "build_from_columns") + monkeypatch.setenv("S3_ENDPOINT_URL", csv_source_bucket) + fake = FakeWarehouseAdapter() + _install_fake_adapter(monkeypatch, fake) + spec = _spec() + spec["csv_source"] = "s3://csv-validation/empty.csv" + monkeypatch.setattr(runner, "load_candidate_spec", lambda: spec) + + with pytest.raises(SystemExit) as exc: + runner.main() + + assert exc.value.code == 1 + assert fake.column_builds == [] # header check blocks the build + out = capsys.readouterr().out + assert '"status":"error"' in out + assert "no header line" in out + assert "s3://csv-validation/empty.csv" in out + + def test_main_build_from_columns_without_csv_source_unchanged(monkeypatch, capsys): """No csv_source key in the spec: behavior is unchanged from before A6 — no header fetch is attempted, and the build proceeds straight through.""" @@ -912,6 +940,17 @@ def _setup_build_from_columns_non_object_config(monkeypatch): monkeypatch.setattr(runner, "load_candidate_spec", lambda: _spec(config=["indexes"])) +def _setup_build_from_columns_non_string_csv_source(monkeypatch): + """Arrange a build_from_columns run whose spec csv_source is not a string (FIX 3).""" + _set_common_env(monkeypatch) + monkeypatch.setenv("VALIDATION_OP", "build_from_columns") + monkeypatch.setenv("CANDIDATE_SPEC_URI", "s3://continuo/candidate-spec/rel-1/svc.orders.json") + _install_fake_adapter(monkeypatch, FakeWarehouseAdapter()) + spec = _spec() + spec["csv_source"] = 123 + monkeypatch.setattr(runner, "load_candidate_spec", lambda: spec) + + def _setup_build_from_columns_invalid_json(monkeypatch): """Arrange a build_from_columns run whose spec body is not valid JSON.""" _set_common_env(monkeypatch) @@ -957,6 +996,11 @@ def _setup(monkeypatch): ("build_from_columns_missing_spec_uri", _setup_build_from_columns_missing_spec_uri, 2), ("build_from_columns_empty_output_columns", _setup_build_from_columns_empty_output_columns, 2), ("build_from_columns_non_object_config", _setup_build_from_columns_non_object_config, 2), + ( + "build_from_columns_non_string_csv_source", + _setup_build_from_columns_non_string_csv_source, + 2, + ), ("build_from_columns_invalid_json", _setup_build_from_columns_invalid_json, 2), ( "build_from_columns_non_object_json_list", @@ -986,13 +1030,14 @@ def test_main_emits_exactly_one_sentinel_block_as_last_stdout_line(monkeypatch, The contract (see ``result.py``) is: exactly ONE sentinel-framed block, as the terminal non-empty stdout line, on every outcome that emits one. Exercises all - twenty block-emitting paths through ``main()`` — success, ensure_schema, + twenty-one block-emitting paths through ``main()`` — success, ensure_schema, drop_schema, empty candidate SQL, S3-fetch error, unknown VALIDATION_OP, adapter discovery failure, missing required adapter env, missing DBT_TARGET_SCHEMA, - missing PROD_SCHEMA, and the ten build_from_columns paths (success, a failing + missing PROD_SCHEMA, and the eleven build_from_columns paths (success, a failing bind check, missing CANDIDATE_SPEC_URI, empty output_columns, a non-object config, - invalid spec JSON, and spec JSON that parses to a list/null/int/str instead of an - object) — each in its own isolated monkeypatch context so scenarios cannot leak + a non-string csv_source, invalid spec JSON, and spec JSON that parses to a + list/null/int/str instead of an object) — each in its own isolated monkeypatch + context so scenarios cannot leak patches into one another. """ for name, setup, expected_exit in _SENTINEL_SCENARIOS: From be30a0e84f57bc802d3248922b8e8414a6921ceb Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 16:19:10 +0200 Subject: [PATCH 11/14] fix(csv_loader): read declared columns as their target Arrow type pyarrow's default type inference in read_csv was the first thing to interpret a value, ahead of conform(): a VARCHAR column holding 00123 inferred as int64 and conform() wrote back "123", and a NUMERIC column holding a valid decimal like 10.50 inferred as float64, which conform()'s own lossy-cast guard then rejected outright. Pass output_columns as read_csv's ConvertOptions.column_types so every declared column is parsed once, directly as the type it is declared to be. Signed-off-by: Simone Carolini --- continuo_python_runtime/csv_loader.py | 18 +++++++++++++- tests/test_csv_loader.py | 35 +++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/continuo_python_runtime/csv_loader.py b/continuo_python_runtime/csv_loader.py index b581a42..e60fc85 100644 --- a/continuo_python_runtime/csv_loader.py +++ b/continuo_python_runtime/csv_loader.py @@ -15,6 +15,7 @@ from continuo_python_runtime.csv_readers import reader_for from continuo_python_runtime.csv_source import CsvSourceReader, parse_csv_uri from continuo_python_runtime.errors import LoadError +from continuo_python_runtime.types import arrow_type, parse_sql_type logger = logging.getLogger("continuo_python_runtime.csv_loader") @@ -24,13 +25,28 @@ def produce_csv(node: Node, reader: CsvSourceReader | None = None) -> "pyarrow.T The caller conforms the result to output_columns exactly as for a script node, so declared types — not csv inference — decide the warehouse schema. + + ``output_columns`` names/types are also passed to ``read_csv`` itself as + its convert schema (via ``ConvertOptions.column_types``): pyarrow's default + type inference is otherwise the *first* place a value gets interpreted, + and it can destroy the very lexical value ``conform()`` is supposed to + preserve -- a VARCHAR column holding ``00123`` infers as int64 and + conform() writes back ``"123"``, and a NUMERIC column holding a valid + decimal like ``10.50`` infers as float64, which conform()'s own + lossy-cast guard then rejects outright. Reading every declared column + directly as its target Arrow type sidesteps both: the value is parsed + once, as the type it is actually declared to be. """ uri = parse_csv_uri(node.reads["csv"]) active_reader = reader if reader is not None else reader_for(uri) + column_types = { + col.name: arrow_type(parse_sql_type(col.type)) for col in node.output_columns + } + convert_options = pyarrow.csv.ConvertOptions(column_types=column_types) try: with tempfile.TemporaryDirectory() as tmp: dest = active_reader.fetch(uri, Path(tmp) / "source.csv") - table = pyarrow.csv.read_csv(dest) + table = pyarrow.csv.read_csv(dest, convert_options=convert_options) except LoadError: raise except Exception as exc: diff --git a/tests/test_csv_loader.py b/tests/test_csv_loader.py index 06d7cff..a4d1921 100644 --- a/tests/test_csv_loader.py +++ b/tests/test_csv_loader.py @@ -1,6 +1,8 @@ """tests/test_csv_loader.py — unit tier: the loader through a local-file test double of the PORT (the port is ours; substituting a test implementation of our own abstraction is not stubbing an external service).""" +from decimal import Decimal + import pyarrow as pa import pytest @@ -79,3 +81,36 @@ def test_produce_csv_no_warning_when_all_columns_declared(caplog): b"order_id,amount\n1,10.5\n2,20.0\n")) assert "csv_header_extra_columns" not in caplog.text + + +def test_produce_csv_preserves_lexical_value_for_declared_varchar_column(): + """A declared VARCHAR column must be read as text -- pyarrow's default + type inference on `00123` would infer int64, and conform() would then + write back "123", silently dropping the leading zeros. Passing the + declared output_columns as read_csv's convert schema (spec: 'output_columns + names/types as the convert schema') prevents that: the column is parsed as + a string in the first place, so no lossy int64 round-trip ever happens.""" + node = _csv_node(output_columns=( + Column(name="order_id", type="VARCHAR(20)", nullable=False), + Column(name="amount", type="DOUBLE PRECISION"), + )) + table = produce_csv(node, reader=LocalFileReader( + b"order_id,amount\n00123,10.5\n00456,20.0\n")) + + assert table.column("order_id").to_pylist() == ["00123", "00456"] + + +def test_produce_csv_declared_decimal_column_not_corrupted_by_float_inference(): + """A declared NUMERIC column must not be inferred as float64 first -- + conform()'s own lossy-cast guard rejects float->decimal casts outright, + so without an explicit convert schema this would fail conform() even for + a value that is a perfectly valid decimal.""" + node = _csv_node(output_columns=( + Column(name="order_id", type="INTEGER", nullable=False), + Column(name="amount", type="NUMERIC(10,2)"), + )) + table = produce_csv(node, reader=LocalFileReader( + b"order_id,amount\n1,10.50\n2,20.00\n")) + + assert pa.types.is_decimal(table.schema.field("amount").type) + assert table.column("amount").to_pylist() == [Decimal("10.50"), Decimal("20.00")] From d1bf4a0f533e1695d39b640dfad8f91724bb8dbf Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 16:19:51 +0200 Subject: [PATCH 12/14] fix(csv_readers): https-only redirects, bounded reads, header ordering, BOM Four independent hardening fixes to the S3/HTTPS csv source readers: - https.py: an https:// source that redirected to a non-https target was followed silently (urllib doesn't care about scheme on redirect), downgrading both the header probe and the full fetch to plaintext. Install a redirect handler that refuses any target not itself https://. - https.py: when a server ignores our Range header and answers 200, the response is the whole object -- reading it with an unbounded resp.read() before even checking the status buffered a multi-gigabyte body in full just to inspect its header line. Bound that read to stop at the first newline or MAX_HEADER_BYTES. - s3.py/https.py: a newline that only arrived after the accumulated probe buffer had already grown past MAX_HEADER_BYTES was returned as a successful (oversized) header line, because the newline check ran before the length check. Extracted the shared extract_header_line() (in csv_source.py, the dependency-free port module) which measures the *resolved line itself* against the limit, and both readers now use it. - s3.py/https.py: a UTF-8 byte-order mark on the CSV's first byte survived a plain "utf-8" decode as a literal U+FEFF prepended to the first column name, so a declared column matching the visually-identical name failed check_header. Decode as utf-8-sig everywhere a header line is produced. Also adds an explicit finite timeout (30s) to both HTTPS urlopen calls: an https source that accepts the connection but stalls on headers or body would otherwise hang a validation or node run indefinitely. Tests: test_csv_readers_integration.py gains oversized-line-with-late- newline and BOM cases for both readers, plus a redirect-to-a-reachable- downgrade-target test proving the pre-fix code actually followed it (rather than merely erroring on an unreachable host). test_csv_readers_ https.py is a new unit tier covering the redirect handler and timeout wiring without a live server. Signed-off-by: Simone Carolini --- continuo_python_runtime/csv_readers/https.py | 91 ++++++++++++--- continuo_python_runtime/csv_readers/s3.py | 5 +- continuo_python_runtime/csv_source.py | 31 +++++- tests/test_csv_readers_https.py | 107 ++++++++++++++++++ tests/test_csv_readers_integration.py | 111 +++++++++++++++++++ tests/test_csv_source.py | 10 ++ 6 files changed, 333 insertions(+), 22 deletions(-) create mode 100644 tests/test_csv_readers_https.py diff --git a/continuo_python_runtime/csv_readers/https.py b/continuo_python_runtime/csv_readers/https.py index 5fafb44..c2bf470 100644 --- a/continuo_python_runtime/csv_readers/https.py +++ b/continuo_python_runtime/csv_readers/https.py @@ -1,5 +1,6 @@ """continuo_python_runtime/csv_readers/https.py""" import shutil +import urllib.error import urllib.request from pathlib import Path @@ -8,8 +9,61 @@ MAX_HEADER_BYTES, CsvSourceReader, CsvUri, + extract_header_line, ) +# Both urlopen calls below must never hang forever: a source that accepts the +# TCP connection but stalls on headers or body would otherwise wedge a +# validation run or a scheduled node run indefinitely. +_TIMEOUT_SECONDS = 30 + +# Chunk size for the bounded read used when a server ignores our Range header +# and answers 200: reading in chunks this small lets fetch_header_line stop +# at the first newline (or MAX_HEADER_BYTES) without ever buffering a +# multi-gigabyte body just to inspect its first line. +_READ_CHUNK_BYTES = 65_536 + + +class _HttpsOnlyRedirectHandler(urllib.request.HTTPRedirectHandler): + """Refuses to follow a redirect whose target is not itself https://. + + urllib follows redirects automatically, and by default does not care + what scheme the target uses -- an https:// source that redirects to + http:// (or any other scheme) would otherwise silently downgrade both + the header probe and the full fetch to plaintext, defeating + parse_csv_uri's https-only restriction. + """ + + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102 + if not newurl.lower().startswith("https://"): + raise urllib.error.HTTPError( + newurl, code, + f"refusing to follow redirect to non-https URL: {newurl}", + headers, fp, + ) + return super().redirect_request(req, fp, code, msg, headers, newurl) + + +_opener = urllib.request.build_opener(_HttpsOnlyRedirectHandler) + + +def _read_bounded(resp, limit: int) -> bytes: + """Read ``resp`` in small chunks, stopping at the first newline or once + more than ``limit`` bytes have been buffered, whichever comes first. + + Used for the range-ignored (status 200) path, where the server's + response is the whole object: an unbounded ``resp.read()`` there would + buffer a multi-gigabyte body in full just to read its header line. + """ + buf = b"" + while True: + chunk = resp.read(_READ_CHUNK_BYTES) + if not chunk: + return buf + buf += chunk + if b"\n" in buf or len(buf) > limit: + return buf + class HttpsCsvSourceReader(CsvSourceReader): """Reads a csv source over HTTPS (public URLs; no auth in v1). Mirrors @@ -27,35 +81,36 @@ def fetch_header_line(self, uri: CsvUri) -> str: end = start + HEADER_PROBE_BYTES - 1 req = urllib.request.Request( uri.raw, headers={"Range": f"bytes={start}-{end}"}) - with urllib.request.urlopen(req) as resp: # noqa: S310 — scheme gated by parse_csv_uri - body = resp.read() + with _opener.open(req, timeout=_TIMEOUT_SECONDS) as resp: # noqa: S310 — scheme gated by parse_csv_uri and _HttpsOnlyRedirectHandler range_honoured = resp.status == 206 + if range_honoured: + body = resp.read() + else: + # The server ignored our Range header and returned the + # entire object (status 200): bound the read itself so a + # multi-gigabyte body is never buffered in full, and + # this response is terminal regardless of its size -- + # every retry would re-fetch the identical full body, so + # looping would only re-append it pass after pass and + # eventually trip a false MAX_HEADER_BYTES overflow. + body = _read_bounded(resp, MAX_HEADER_BYTES) if not range_honoured: - # The server ignored our Range header and returned the entire - # object (status 200), not just the requested window -- *body* - # is therefore the whole object and this response is terminal, - # regardless of its size relative to HEADER_PROBE_BYTES. Every - # retry would re-fetch the identical full body, so looping - # would only re-append it pass after pass and eventually trip - # a false MAX_HEADER_BYTES overflow. - if b"\n" in body: - return body.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") - if len(body) > MAX_HEADER_BYTES: - raise ValueError( - f"csv header line exceeds {MAX_HEADER_BYTES} bytes: {uri.raw}") - return body.rstrip(b"\r").decode("utf-8") + return extract_header_line(body, uri.raw) buf += body if b"\n" in buf: - return buf.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") + return extract_header_line(buf, uri.raw) if len(body) < HEADER_PROBE_BYTES: # whole object read, no newline - return buf.rstrip(b"\r").decode("utf-8") + return buf.rstrip(b"\r").decode("utf-8-sig") if len(buf) > MAX_HEADER_BYTES: raise ValueError( f"csv header line exceeds {MAX_HEADER_BYTES} bytes: {uri.raw}") start += HEADER_PROBE_BYTES def fetch(self, uri: CsvUri, dest: Path) -> Path: - with urllib.request.urlopen(uri.raw) as resp, open(dest, "wb") as f: # noqa: S310 + with ( + _opener.open(uri.raw, timeout=_TIMEOUT_SECONDS) as resp, # noqa: S310 — scheme gated by parse_csv_uri and _HttpsOnlyRedirectHandler + open(dest, "wb") as f, + ): shutil.copyfileobj(resp, f) return dest diff --git a/continuo_python_runtime/csv_readers/s3.py b/continuo_python_runtime/csv_readers/s3.py index 2739848..dbfc291 100644 --- a/continuo_python_runtime/csv_readers/s3.py +++ b/continuo_python_runtime/csv_readers/s3.py @@ -8,6 +8,7 @@ MAX_HEADER_BYTES, CsvSourceReader, CsvUri, + extract_header_line, ) from continuo_python_runtime.validation.s3 import make_s3_client @@ -38,9 +39,9 @@ def fetch_header_line(self, uri: CsvUri) -> str: raise buf += body if b"\n" in buf: - return buf.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8") + return extract_header_line(buf, uri.raw) if len(body) < HEADER_PROBE_BYTES: # whole object read, no newline - return buf.rstrip(b"\r").decode("utf-8") + return buf.rstrip(b"\r").decode("utf-8-sig") if len(buf) > MAX_HEADER_BYTES: raise ValueError( f"csv header line exceeds {MAX_HEADER_BYTES} bytes: {uri.raw}") diff --git a/continuo_python_runtime/csv_source.py b/continuo_python_runtime/csv_source.py index a93751b..98e9c07 100644 --- a/continuo_python_runtime/csv_source.py +++ b/continuo_python_runtime/csv_source.py @@ -22,9 +22,14 @@ class CsvUri: def parse_csv_uri(uri: str) -> CsvUri: """Parse a csv source URI. Accepts exactly s3://bucket/key and https://... - Raises ValueError for anything else (http:// included) so contract - validation fails at lint/parse time, never at run time. + Raises ValueError for anything else (http:// included), and for a + non-string ``uri`` (e.g. a contract's `reads: {csv: 123}`) rather than + letting `.startswith()` raise a bare AttributeError/TypeError -- callers + (the contract loader) catch ValueError to turn this into a ContractError, + so a malformed uri fails at lint/parse time, never at run time. """ + if not isinstance(uri, str): + raise ValueError(f"csv uri must be a string, got {type(uri).__name__}: {uri!r}") if uri.startswith("s3://"): bucket, _, key = uri[len("s3://"):].partition("/") if not bucket or not key: @@ -40,6 +45,28 @@ def parse_csv_uri(uri: str) -> CsvUri: ) +def extract_header_line(data: bytes, source_desc: str) -> str: + """Return the first line of ``data`` (no trailing newline), decoded as + utf-8-sig so a UTF-8 byte-order mark on the CSV's first byte does not end + up prepended to the first column name. + + Shared by every :class:`CsvSourceReader` adapter's ``fetch_header_line``: + the check must run on the *resolved line itself* (the bytes up to, and + including the absence of, the first newline), not merely on an + intermediate buffer length -- a newline that only arrives after the + accumulated buffer has already grown past ``MAX_HEADER_BYTES`` must still + be rejected as oversized, not returned as a successful (if enormous) + header line. + + Raises: + ValueError: If the line exceeds ``MAX_HEADER_BYTES``. + """ + line = data.split(b"\n", 1)[0] if b"\n" in data else data + if len(line) > MAX_HEADER_BYTES: + raise ValueError(f"csv header line exceeds {MAX_HEADER_BYTES} bytes: {source_desc}") + return line.rstrip(b"\r").decode("utf-8-sig") + + def check_header(header_cols: list[str], declared_cols: list[str]) -> set[str]: """Presence-only header conformance: every declared column must appear in the CSV header, in any order. Returns the set of header columns NOT diff --git a/tests/test_csv_readers_https.py b/tests/test_csv_readers_https.py new file mode 100644 index 0000000..36ebd76 --- /dev/null +++ b/tests/test_csv_readers_https.py @@ -0,0 +1,107 @@ +"""tests/test_csv_readers_https.py — unit tier: HttpsCsvSourceReader mechanics +that don't need a live server (redirect-scheme enforcement, timeout wiring). +See test_csv_readers_integration.py for the live-server-backed Range/stream +behavior tests, including an end-to-end redirect-downgrade test against a +reachable target.""" +import urllib.error +import urllib.request + +import pytest + +from continuo_python_runtime.csv_readers import https as https_mod +from continuo_python_runtime.csv_readers.https import ( + _HttpsOnlyRedirectHandler, + _TIMEOUT_SECONDS, + HttpsCsvSourceReader, +) +from continuo_python_runtime.csv_source import CsvUri + + +@pytest.mark.parametrize("newurl", [ + "http://example.com/orders.csv", + "HTTP://example.com/orders.csv", + "ftp://example.com/orders.csv", +]) +def test_redirect_handler_rejects_non_https_target(newurl): + handler = _HttpsOnlyRedirectHandler() + req = urllib.request.Request("https://example.com/orders.csv") + with pytest.raises(urllib.error.HTTPError, match="non-https"): + handler.redirect_request(req, None, 302, "Found", {}, newurl) + + +def test_redirect_handler_allows_https_target(monkeypatch): + handler = _HttpsOnlyRedirectHandler() + req = urllib.request.Request("https://example.com/orders.csv") + captured = {} + + def fake_super_redirect(self, req, fp, code, msg, headers, newurl): + captured["newurl"] = newurl + return "a-request-object" + + monkeypatch.setattr( + urllib.request.HTTPRedirectHandler, "redirect_request", fake_super_redirect) + + result = handler.redirect_request( + req, None, 302, "Found", {}, "https://example.com/y.csv") + + assert result == "a-request-object" + assert captured["newurl"] == "https://example.com/y.csv" + + +def test_timeout_constant_is_finite_and_positive(): + assert 0 < _TIMEOUT_SECONDS < 300 + + +class _FakeHttpResponse: + """Just enough of urllib's response object for fetch_header_line/fetch: + a context manager with .status and a chunked .read(size).""" + + status = 200 + + def __init__(self, body: bytes): + self._remaining = body + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self, size: int = -1) -> bytes: + if size < 0: + chunk, self._remaining = self._remaining, b"" + else: + chunk, self._remaining = self._remaining[:size], self._remaining[size:] + return chunk + + +def test_fetch_header_line_passes_the_timeout_to_opener_open(monkeypatch): + captured = {} + + def fake_open(req, timeout=None): + captured["timeout"] = timeout + return _FakeHttpResponse(b"order_id,amount\n1,2\n") + + monkeypatch.setattr(https_mod._opener, "open", fake_open) + + uri = CsvUri(scheme="https", raw="https://example.com/orders.csv") + header = HttpsCsvSourceReader().fetch_header_line(uri) + + assert header == "order_id,amount" + assert captured["timeout"] == _TIMEOUT_SECONDS + + +def test_fetch_passes_the_timeout_to_opener_open(monkeypatch, tmp_path): + captured = {} + + def fake_open(url, timeout=None): + captured["timeout"] = timeout + return _FakeHttpResponse(b"order_id,amount\n1,2\n") + + monkeypatch.setattr(https_mod._opener, "open", fake_open) + + uri = CsvUri(scheme="https", raw="https://example.com/orders.csv") + dest = HttpsCsvSourceReader().fetch(uri, tmp_path / "o.csv") + + assert dest.read_bytes() == b"order_id,amount\n1,2\n" + assert captured["timeout"] == _TIMEOUT_SECONDS diff --git a/tests/test_csv_readers_integration.py b/tests/test_csv_readers_integration.py index cba05ab..c82e856 100644 --- a/tests/test_csv_readers_integration.py +++ b/tests/test_csv_readers_integration.py @@ -56,6 +56,20 @@ def _long_header_body(header_len: int) -> tuple[bytes, str]: # overflow (FIX 1). NO_NEWLINE_MID_BODY = b"a" * 100_000 +# A header line that DOES terminate in a newline, but only after exceeding +# MAX_HEADER_BYTES. Unlike OVERFLOW_BODY (no newline anywhere), this exercises +# the ordering bug where a newline arriving inside an already-oversized buffer +# was returned as a successful (if oversized) header instead of being +# rejected: the length check must run on the resolved line itself, not only +# on an intermediate buffer length reached before any newline was seen. +OVERSIZED_WITH_NEWLINE_LEN = MAX_HEADER_BYTES + 50_000 +OVERSIZED_WITH_NEWLINE_BODY, _ = _long_header_body(OVERSIZED_WITH_NEWLINE_LEN) + +# A UTF-8 byte-order mark prefixed onto an otherwise-ordinary header: the +# reader must strip it so the first declared column name still matches. +BOM_CSV_BODY = b"\xef\xbb\xbf" + CSV_BODY +BOM_HEADER_STR = "order_id,amount,extra" + @pytest.fixture(scope="session") def minio(minio_container): # minio_container: session fixture starting minio via docker @@ -68,6 +82,9 @@ def minio(minio_container): # minio_container: session fixture starting minio v client.put_object(Bucket="drops", Key="orders.csv", Body=CSV_BODY) client.put_object(Bucket="drops", Key="long_header.csv", Body=LONG_HEADER_BODY) client.put_object(Bucket="drops", Key="overflow.csv", Body=OVERFLOW_BODY) + client.put_object( + Bucket="drops", Key="oversized_with_newline.csv", Body=OVERSIZED_WITH_NEWLINE_BODY) + client.put_object(Bucket="drops", Key="bom.csv", Body=BOM_CSV_BODY) return endpoint @@ -103,6 +120,23 @@ def test_s3_fetch_header_line_overflow_raises(minio, monkeypatch): S3CsvSourceReader().fetch_header_line(parse_csv_uri("s3://drops/overflow.csv")) +def test_s3_fetch_header_line_oversized_line_with_late_newline_raises(minio, monkeypatch): + """A newline that only shows up after the accumulated buffer already + exceeds MAX_HEADER_BYTES must still raise -- not be returned as a + successful (oversized) header line.""" + monkeypatch.setenv("S3_ENDPOINT_URL", minio) + with pytest.raises(ValueError, match="exceeds"): + S3CsvSourceReader().fetch_header_line( + parse_csv_uri("s3://drops/oversized_with_newline.csv")) + + +def test_s3_fetch_header_line_strips_utf8_bom(minio, monkeypatch): + monkeypatch.setenv("S3_ENDPOINT_URL", minio) + header = S3CsvSourceReader().fetch_header_line(parse_csv_uri("s3://drops/bom.csv")) + assert header == BOM_HEADER_STR + assert header[0] != "" + + class _RangeHandler(http.server.BaseHTTPRequestHandler): honour_range = True @@ -111,6 +145,8 @@ class _RangeHandler(http.server.BaseHTTPRequestHandler): "/long_header.csv": LONG_HEADER_BODY, "/overflow.csv": OVERFLOW_BODY, "/no_newline.csv": NO_NEWLINE_MID_BODY, + "/oversized_with_newline.csv": OVERSIZED_WITH_NEWLINE_BODY, + "/bom.csv": BOM_CSV_BODY, } def do_GET(self): @@ -185,6 +221,81 @@ def test_https_fetch_header_line_no_newline_mid_size_is_not_a_false_overflow(htt assert header == NO_NEWLINE_MID_BODY.decode("utf-8") +def test_https_fetch_header_line_oversized_line_with_late_newline_raises(http_csv_server): + """A newline that only shows up after the accumulated buffer already + exceeds MAX_HEADER_BYTES must still raise -- not be returned as a + successful (oversized) header line. Exercises both the range-honoured + (multi-probe accumulation) and range-ignored (single bounded read) paths.""" + uri = CsvUri(scheme="https", raw=f"http://{http_csv_server}/oversized_with_newline.csv") + with pytest.raises(ValueError, match="exceeds"): + HttpsCsvSourceReader().fetch_header_line(uri) + + +def test_https_fetch_header_line_strips_utf8_bom(http_csv_server): + uri = CsvUri(scheme="https", raw=f"http://{http_csv_server}/bom.csv") + header = HttpsCsvSourceReader().fetch_header_line(uri) + assert header == BOM_HEADER_STR + assert header[0] != "" + + +class _DowngradeTargetHandler(http.server.BaseHTTPRequestHandler): + """A REACHABLE plain-HTTP server standing in for a downgrade target. Using + a reachable target (rather than a nonexistent host) makes the redirect + tests below meaningful: the vulnerable (pre-fix) code doesn't merely fail + to connect, it actually follows the redirect and successfully returns + this server's content -- proving the redirect really would have been + followed, not just that *some* exception happened to occur.""" + + def do_GET(self): + self.send_response(200) + self.send_header("Content-Length", str(len(CSV_BODY))) + self.end_headers() + self.wfile.write(CSV_BODY) + + def log_message(self, *args): + pass + + +@pytest.fixture +def downgrade_redirect_uri(): + """An https:// (test-convention) source whose every request 302s to a + second, reachable, genuinely plain-HTTP server.""" + with socketserver.TCPServer(("127.0.0.1", 0), _DowngradeTargetHandler) as target: + tt = threading.Thread(target=target.serve_forever, daemon=True) + tt.start() + target_port = target.server_address[1] + + redirector = type("Redirector", (http.server.BaseHTTPRequestHandler,), { + "do_GET": lambda self: ( + self.send_response(302), + self.send_header("Location", f"http://127.0.0.1:{target_port}/orders.csv"), + self.end_headers(), + ), + "log_message": lambda self, *a: None, + }) + with socketserver.TCPServer(("127.0.0.1", 0), redirector) as src: + ts = threading.Thread(target=src.serve_forever, daemon=True) + ts.start() + yield f"http://127.0.0.1:{src.server_address[1]}/redirect.csv" + src.shutdown() + target.shutdown() + + +def test_https_fetch_header_line_rejects_redirect_to_non_https(downgrade_redirect_uri): + """An https:// source that redirects to a REACHABLE non-https target must + not be followed -- the pre-fix code would have silently downgraded to + plaintext and returned the target's header line instead of raising.""" + uri = CsvUri(scheme="https", raw=downgrade_redirect_uri) + with pytest.raises(Exception): + HttpsCsvSourceReader().fetch_header_line(uri) + + +def test_https_fetch_rejects_redirect_to_non_https(downgrade_redirect_uri, tmp_path): + uri = CsvUri(scheme="https", raw=downgrade_redirect_uri) + with pytest.raises(Exception): + HttpsCsvSourceReader().fetch(uri, tmp_path / "o.csv") + + def test_reader_for_dispatches_on_scheme(): assert isinstance(reader_for(parse_csv_uri("s3://b/k")), S3CsvSourceReader) assert isinstance( diff --git a/tests/test_csv_source.py b/tests/test_csv_source.py index 7e7d2e3..87e5254 100644 --- a/tests/test_csv_source.py +++ b/tests/test_csv_source.py @@ -35,6 +35,16 @@ def test_parse_rejects_invalid(bad): parse_csv_uri(bad) +@pytest.mark.parametrize("bad", [123, 1.5, None, True, [], {}, b"s3://drops/x.csv"]) +def test_parse_rejects_non_string(bad): + """A non-string uri (e.g. `reads: {csv: 123}`) must raise ValueError, not + AttributeError from a bare `.startswith()` call -- callers (the contract + loader) catch ValueError/TypeError to turn this into a ContractError, and + an uncaught AttributeError would crash validation instead.""" + with pytest.raises(ValueError): + parse_csv_uri(bad) + + def test_check_header_presence_only_any_order(): extras = check_header(["b", "a", "c"], ["a", "b"]) assert extras == {"c"} From d4044ac0964b7959df28c4efcf09359ea32ed04b Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 16:20:03 +0200 Subject: [PATCH 13/14] fix(contract,validation): reject malformed non-string values, don't crash - contract/loader.py: kind: [python-csv] (or any unhashable kind) raised a bare TypeError from `kind not in KINDS` before the intended ContractError. Check isinstance(kind, str) first. - validation/runner.py: the build_from_columns csv_source guard was gated on truthiness (`if csv_source and not isinstance(...)`), so csv_source: 0 / false / [] / {} skipped both the type check and the header check below -- a malformed python-csv candidate could pass promotion silently. Gate on presence of the key instead. Signed-off-by: Simone Carolini --- continuo_python_runtime/contract/loader.py | 2 +- continuo_python_runtime/validation/runner.py | 6 +++- tests/test_contract_loader.py | 32 ++++++++++++++++++++ tests/test_validation_runner.py | 25 +++++++++++++++ 4 files changed, 63 insertions(+), 2 deletions(-) create mode 100644 tests/test_contract_loader.py diff --git a/continuo_python_runtime/contract/loader.py b/continuo_python_runtime/contract/loader.py index b3b1032..055d2fd 100644 --- a/continuo_python_runtime/contract/loader.py +++ b/continuo_python_runtime/contract/loader.py @@ -174,7 +174,7 @@ def parse_node( raise ContractError(f"{label}: unknown key(s) {sorted(unknown)}") kind = raw.get("kind", "python-model") - if kind not in KINDS: + if not isinstance(kind, str) or kind not in KINDS: raise ContractError( f"{label}: 'kind' must be one of {sorted(KINDS)}, got {kind!r}" ) diff --git a/continuo_python_runtime/validation/runner.py b/continuo_python_runtime/validation/runner.py index 37a1174..b650df3 100644 --- a/continuo_python_runtime/validation/runner.py +++ b/continuo_python_runtime/validation/runner.py @@ -186,7 +186,11 @@ def main() -> None: sys.exit(2) config = raw_config or {} csv_source = spec.get("csv_source", "") - if csv_source and not isinstance(csv_source, str): + if "csv_source" in spec and not isinstance(csv_source, str): + # Checked by presence, not truthiness: `csv_source: 0` / `false` / + # `[]` / `{}` are all non-string values a malformed spec could + # carry, and a truthiness guard would silently skip both this + # type check and the header check below for every one of them. msg = f"candidate spec 'csv_source' must be a string, got {type(csv_source).__name__}" logger.error("%s", msg) print(result.result_block("error", msg, unique_id=unique_id), flush=True) diff --git a/tests/test_contract_loader.py b/tests/test_contract_loader.py new file mode 100644 index 0000000..53c9d29 --- /dev/null +++ b/tests/test_contract_loader.py @@ -0,0 +1,32 @@ +"""tests/test_contract_loader.py — unit tests for parse_node's own validation +rules that are cheapest to exercise directly against a raw dict, rather than +through a full contract-directory fixture (see test_harness.py for those).""" +import pytest + +from continuo_python_runtime.contract.loader import parse_node +from continuo_python_runtime.errors import ContractError + + +def _base_node(**over): + node = { + "schema": "analytics", "table": "t", "owner": "m", "schedule": "daily", + "criticality": "SECONDARY", "script": "scripts/t.py", + "reads": {"ids": "select id from analytics.a"}, + "output_columns": [{"name": "id", "type": "INTEGER", "nullable": False}], + } + node.update(over) + return node + + +@pytest.mark.parametrize("bad_kind", [["python-csv"], {"k": "python-csv"}, 3, True, None]) +def test_non_string_kind_raises_contracterror_not_typeerror(bad_kind): + """A malformed `kind: [python-csv]` must fail membership-testing against + the frozenset KINDS with a ContractError, not an unhandled bare TypeError + from `kind not in KINDS` on an unhashable value (list/dict).""" + with pytest.raises(ContractError, match="'kind' must be one of"): + parse_node(_base_node(kind=bad_kind), "t.yml") + + +def test_valid_string_kind_still_accepted(): + node = parse_node(_base_node(kind="python-model"), "t.yml") + assert node.kind == "python-model" diff --git a/tests/test_validation_runner.py b/tests/test_validation_runner.py index 267679d..3773d08 100644 --- a/tests/test_validation_runner.py +++ b/tests/test_validation_runner.py @@ -808,6 +808,31 @@ def test_main_build_from_columns_without_csv_source_unchanged(monkeypatch, capsy assert '"status":"success"' in capsys.readouterr().out +@pytest.mark.parametrize("bad", [0, False, [], {}], ids=["zero", "false", "list", "dict"]) +def test_main_build_from_columns_falsey_non_string_csv_source_exits_2(monkeypatch, capsys, bad): + """A csv_source key that IS present but falsey and non-string (0, false, + [], {}) must still be rejected as a type error: exit 2, never silently + treated as 'no csv_source' (which would skip the header check entirely + and let a malformed python-csv candidate pass promotion).""" + _set_common_env(monkeypatch) + monkeypatch.setenv("VALIDATION_OP", "build_from_columns") + monkeypatch.setenv("CANDIDATE_SPEC_URI", "s3://continuo/candidate-spec/rel-1/svc.orders.json") + fake = FakeWarehouseAdapter() + _install_fake_adapter(monkeypatch, fake) + spec = _spec() + spec["csv_source"] = bad + monkeypatch.setattr(runner, "load_candidate_spec", lambda: spec) + + with pytest.raises(SystemExit) as exc: + runner.main() + + assert exc.value.code == 2 + assert fake.column_builds == [] # never reached the adapter + out = capsys.readouterr().out + assert '"status":"error"' in out + assert "csv_source" in out + + # -------------------------------------------------------------------------- # main — sentinel-block invariant across every block-emitting exit path # -------------------------------------------------------------------------- From 0a1d2151907cd87e7d2634e1d339a6dcec443048 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 21 Aug 2026 16:20:11 +0200 Subject: [PATCH 14/14] ci: exclude integration tests from the default runtime suite The documented and CI "Tests (runtime)" command only deselected `image`, so the new minio-backed csv-reader/validation-runner integration tests (marked `integration`) stayed selected -- the standard pre-PR command from CONTRIBUTING.md now required Docker despite that doc saying Docker is only needed for the Postgres/Trino integration tests. Deselect `integration` there too, and add a dedicated step that runs them explicitly on the same runner (docker is available; minio self-provisions via the minio_container fixture's own `docker run`), mirroring how the Postgres/Trino adapters split their integration tier into a separate job. Nothing is silently skipped: the new step names both test files. Signed-off-by: Simone Carolini --- .github/workflows/ci.yml | 10 ++++++++-- CONTRIBUTING.md | 11 +++++++---- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1b3b630..88fabaf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,9 +24,15 @@ jobs: run: uv run --package continuo-python-runtime-trino mypy adapters/trino/continuo_python_runtime_trino # `-m "not image"` deselects tests/test_image_smoke_validation.py, which # needs a built engine image and the env naming it. Those tests run in - # images.yml's smoke jobs, where an image actually exists. + # images.yml's smoke jobs, where an image actually exists. `and not + # integration` deselects the csv-reader/validation-runner tests that + # need a real minio backend (started via `docker run` by the + # `minio_container` fixture, not docker-compose) -- those run in the + # dedicated step below, on the same runner, where docker is available. - name: Tests (runtime) - run: uv run pytest --cov=continuo_python_runtime -m "not image" -v + run: uv run pytest --cov=continuo_python_runtime -m "not image and not integration" -v + - name: Tests (runtime, integration) + run: uv run pytest tests/test_csv_readers_integration.py tests/test_validation_runner.py -m integration -v - name: Tests (contract) run: uv run pytest contract/tests -v - name: Tests (adapter units) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 0494873..1b82927 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -39,7 +39,8 @@ repository; please do not add headers to new files. ## Development setup Prerequisites: Python 3.14+, [uv](https://docs.astral.sh/uv/), and Docker (only needed -for the Postgres/Trino integration tests). +for the Postgres/Trino integration tests and the csv-reader/validation-runner +integration tests, which start a real minio backend via `docker run`). ```bash uv sync --all-packages --all-groups @@ -58,14 +59,16 @@ uv run mypy continuo_python_runtime uv run mypy contract/continuo_engine_contract uv run --package continuo-python-runtime-postgres mypy adapters/postgres/continuo_python_runtime_postgres uv run --package continuo-python-runtime-trino mypy adapters/trino/continuo_python_runtime_trino -uv run pytest --cov=continuo_python_runtime -m "not image" -v +uv run pytest --cov=continuo_python_runtime -m "not image and not integration" -v +uv run pytest tests/test_csv_readers_integration.py tests/test_validation_runner.py -m integration -v uv run pytest contract/tests -v uv run pytest adapters/postgres/tests adapters/trino/tests -m "not integration" -v ``` These are exactly what `.github/workflows/ci.yml` runs. Integration tests against a real -Postgres/Trino stack need Docker and are not required for most changes — see -`.github/workflows/ci.yml` for how CI stands them up if you want to run them locally. +Postgres/Trino stack, or against the csv-reader/validation-runner minio backend, need +Docker and are not required for most changes — see `.github/workflows/ci.yml` for how CI +stands them up if you want to run them locally. Also run the security scan before opening a pull request that touches dependencies or anything that could carry a credential: