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: 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/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/contract/loader.py b/continuo_python_runtime/contract/loader.py index 50ec010..055d2fd 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 not isinstance(kind, str) or 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,21 @@ 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: + 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: @@ -200,39 +223,50 @@ 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/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/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/continuo_python_runtime/csv_loader.py b/continuo_python_runtime/csv_loader.py new file mode 100644 index 0000000..e60fc85 --- /dev/null +++ b/continuo_python_runtime/csv_loader.py @@ -0,0 +1,68 @@ +"""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 +from continuo_python_runtime.types import arrow_type, parse_sql_type + +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. + + ``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, convert_options=convert_options) + except LoadError: + 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/__init__.py b/continuo_python_runtime/csv_readers/__init__.py new file mode 100644 index 0000000..db65ae5 --- /dev/null +++ b/continuo_python_runtime/csv_readers/__init__.py @@ -0,0 +1,17 @@ +"""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. + + 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() + 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 new file mode 100644 index 0000000..c2bf470 --- /dev/null +++ b/continuo_python_runtime/csv_readers/https.py @@ -0,0 +1,118 @@ +"""continuo_python_runtime/csv_readers/https.py""" +import shutil +import urllib.error +import urllib.request +from pathlib import Path + +from continuo_python_runtime.csv_source import ( + HEADER_PROBE_BYTES, + 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 + S3CsvSourceReader's probe-and-extend strategy: each ranged request is + 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 + buf = b"" + while True: + end = start + HEADER_PROBE_BYTES - 1 + req = urllib.request.Request( + uri.raw, headers={"Range": f"bytes={start}-{end}"}) + 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: + return extract_header_line(body, uri.raw) + buf += body + if b"\n" in buf: + 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-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 ( + _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 + + +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..dbfc291 --- /dev/null +++ b/continuo_python_runtime/csv_readers/s3.py @@ -0,0 +1,57 @@ +"""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, + CsvSourceReader, + CsvUri, + extract_header_line, +) +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 + 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 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-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: + 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/continuo_python_runtime/csv_source.py b/continuo_python_runtime/csv_source.py new file mode 100644 index 0000000..98e9c07 --- /dev/null +++ b/continuo_python_runtime/csv_source.py @@ -0,0 +1,96 @@ +"""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), 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: + 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://"): + 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://..." + ) + + +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 + 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/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/continuo_python_runtime/validation/runner.py b/continuo_python_runtime/validation/runner.py index f49ead9..b650df3 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,16 @@ 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" 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) + sys.exit(2) else: prod_schema = _require("PROD_SCHEMA") elif op in _SCHEMA_OPS: @@ -220,6 +240,21 @@ 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) + 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: + 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/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/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/conftest.py b/tests/conftest.py index 0b8b9dc..1c4da6e 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,95 @@ 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. + # 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/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_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 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"}) 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_csv_loader.py b/tests/test_csv_loader.py new file mode 100644 index 0000000..a4d1921 --- /dev/null +++ b/tests/test_csv_loader.py @@ -0,0 +1,116 @@ +"""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 + +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()) + + +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 + + +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")] 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 new file mode 100644 index 0000000..c82e856 --- /dev/null +++ b/tests/test_csv_readers_integration.py @@ -0,0 +1,302 @@ +"""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 ( + 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) + +# 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 + +# 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 + 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) + 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 + + +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")) + + +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")) + + +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 + + _BODIES = { + "/orders.csv": CSV_BODY, + "/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): + 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 + + +@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_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_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_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( + reader_for(parse_csv_uri("https://x/y.csv")), HttpsCsvSourceReader) diff --git a/tests/test_csv_source.py b/tests/test_csv_source.py new file mode 100644 index 0000000..87e5254 --- /dev/null +++ b/tests/test_csv_source.py @@ -0,0 +1,64 @@ +"""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", + "https:///x.csv", # empty host not accepted + "https://", # empty host not accepted + "orders.csv", + "", +]) +def test_parse_rejects_invalid(bad): + with pytest.raises(ValueError): + 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"} + + +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] 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 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(): diff --git a/tests/test_validation_runner.py b/tests/test_validation_runner.py index a619dc3..3773d08 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,190 @@ 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 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, 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( + "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, + ) + client.put_object(Bucket="csv-validation", Key="empty.csv", Body=b"") + 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 + + +@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.""" + _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 + + +@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 # -------------------------------------------------------------------------- @@ -780,6 +965,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) @@ -825,6 +1021,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", @@ -854,13 +1055,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: