import logging

import os

import pathlib

import shutil

import tempfile

import time

import uuid

from dataclasses import dataclass

from urllib.parse import urlparse

MINIO_BUCKET = os.environ.get("CI_MINIO_BUCKET", "ci-tests")

MINIO_ENDPOINT = os.environ.get(

"CI_MINIO_ENDPOINT", "http://minio.minio.svc.cluster.local:9000"

)

MINIO_REGION = os.environ.get("CI_MINIO_REGION", "us-east-1")

KAFKA_BOOTSTRAP = os.environ.get(

"KAFKA_BOOTSTRAP_SERVERS", "ci-kafka-bootstrap.kafka:9092"

)

def env_truthy(name: str) -> bool:

"""Return True when an environment variable is set to a truthy value."""

value = os.environ.get(name)

return value is not None and value.lower() not in {"", "0", "false", "no"}

def required_env(name: str) -> str:

"""Return a required environment variable or raise a descriptive error."""

value = os.environ.get(name)

if value:

return value

raise RuntimeError(f"required environment variable '{name}' is not set")

def runs_in_ci() -> bool:

"""Return True when the test suite is running under CI."""

return env_truthy("CI")

@dataclass

class DeltaTestLocation:

"""Describe where the Delta sink writes test data and how to read it back."""

uri: str

connector_config: dict[str, object]

root_path: str

local_dir: pathlib.Path | None = None

@classmethod

def create(

cls,

pipeline_name: str,

*,

mode: str = "truncate",

stable_subpath: str | None = None,

) -> "DeltaTestLocation":

"""Use the local filesystem for local runs and MinIO-backed S3 in CI.

:param mode: Value of the connector's ``mode`` field. Output

connectors use ``"truncate"`` (the default); input connectors

should pass ``"snapshot"``.

:param stable_subpath: When set, locates the table at a fixed

path so cached data persists across test runs. When unset,

a fresh random subpath is used (the original behavior). Use

a stable path only when the table contents are deterministic

and idempotent across runs.

"""

if runs_in_ci():

access_key = required_env("CI_K8S_MINIO_ACCESS_KEY_ID")

secret_key = required_env("CI_K8S_MINIO_SECRET_ACCESS_KEY")

tail = stable_subpath if stable_subpath is not None else uuid.uuid4().hex

prefix = f"{pipeline_name}/{tail}"

root_path = f"{MINIO_BUCKET}/{prefix}"

minio_endpoint = MINIO_ENDPOINT.rstrip("/")

parsed_endpoint = urlparse(minio_endpoint)

if (

parsed_endpoint.scheme not in {"http", "https"}

or not parsed_endpoint.netloc

):

raise ValueError(

"CI_MINIO_ENDPOINT must be a full URL, e.g. "

"'http://minio.minio.svc.cluster.local:9000'"

)

return cls(

uri=f"s3://{root_path}",

connector_config={

"uri": f"s3://{root_path}",

"mode": mode,

"aws_access_key_id": access_key,

"aws_secret_access_key": secret_key,

"aws_region": MINIO_REGION,

"aws_endpoint": minio_endpoint,

"aws_allow_http": str(parsed_endpoint.scheme == "http").lower(),

},

root_path=root_path,

)

if stable_subpath is not None:

local_dir = pathlib.Path("/tmp") / f"{pipeline_name}_{stable_subpath}"

local_dir.mkdir(parents=True, exist_ok=True)

else:

local_dir = pathlib.Path(

tempfile.mkdtemp(prefix=f"{pipeline_name}_delta_", dir="/tmp")

)

return cls(

uri=f"file://{local_dir}",

connector_config={

"uri": f"file://{local_dir}",

"mode": mode,

},

root_path=str(local_dir),

local_dir=local_dir,

)

def delta_storage_options(self) -> dict[str, str]:

"""Return `deltalake` storage_options derived from the connector config."""

return {

k: str(v)

for k, v in self.connector_config.items()

if k not in ("uri", "mode")

}

def _s3_filesystem(self):

"""Build a pyarrow ``S3FileSystem`` from the connector config.

Pyarrow imports are deferred to keep module-level test collection

cheap on hosts that don't read Delta tables.

"""

import pyarrow.fs as pafs

cfg = self.connector_config

endpoint = str(cfg["aws_endpoint"]).rstrip("/")

parsed_endpoint = urlparse(endpoint)

return pafs.S3FileSystem(

access_key=str(cfg["aws_access_key_id"]),

secret_key=str(cfg["aws_secret_access_key"]),

region=str(cfg["aws_region"]),

scheme=parsed_endpoint.scheme,

endpoint_override=parsed_endpoint.netloc,

)

def log_json_paths(self) -> list[str]:

"""List Delta transaction log JSON files in version order."""

if self.local_dir is not None:

return [

str(path)

for path in sorted((self.local_dir / "_delta_log").glob("*.json"))

]

import pyarrow.fs as pafs

fs = self._s3_filesystem()

infos = fs.get_file_info(

pafs.FileSelector(f"{self.root_path}/_delta_log", recursive=False)

)

return sorted(

info.path

for info in infos

if info.type == pafs.FileType.File and info.path.endswith(".json")

)

def _read_text(self, path: str) -> str:

"""Read a text file from the configured backend."""

if self.local_dir is not None:

return pathlib.Path(path).read_text(encoding="utf-8")

with self._s3_filesystem().open_input_file(path) as handle:

return handle.readall().decode("utf-8")

def _read_parquet(self, relative_path: str):

"""Read a Delta data file (relative to ``root_path``) as a pyarrow.Table."""

import pyarrow.parquet as pq

if self.local_dir is not None:

return pq.read_table(self.local_dir / relative_path)

return pq.read_table(

f"{self.root_path}/{relative_path}", filesystem=self._s3_filesystem()

)

def read_rows(self) -> list[dict]:

"""Read the active rows of the Delta table by replaying its log.

Walks ``_delta_log/*.json``, follows ``add``/``remove`` actions to

derive the current set of parquet files, reads them with pyarrow

and drops Feldera-internal ``__feldera_*`` columns. Works against

both local-filesystem and S3-backed (MinIO) tables; avoids the

``deltalake`` Python package, whose wheel aborts on aarch64 hosts.

"""

import json as _json

import pyarrow as pa

active: dict[str, None] = {}

for log_path in self.log_json_paths():

for line in self._read_text(log_path).splitlines():

action = _json.loads(line)

if (add := action.get("add")) is not None:

active[add["path"]] = None

if (remove := action.get("remove")) is not None:

active.pop(remove["path"], None)

if not active:

return []

tables = [self._read_parquet(rel) for rel in sorted(active)]

return [

{

key: value

for key, value in row.items()

if not key.startswith("__feldera_")

}

for row in pa.concat_tables(tables).to_pylist()

]

def row_count(self, missing_ok: bool = False) -> int:

"""Return the row count of the current Delta snapshot.

Uses the per-file `numRecords` stats recorded in the delta log, so

we never need to scan parquet (and never need pyarrow).

The `deltalake` import is deferred to here so that module-level

test collection does not crash on aarch64 hosts where the wheel

aborts on import; tests that need this method gate themselves

with `@skip_on_arm64`.

:param missing_ok: When True, return ``-1`` if the table does not

exist instead of raising. Useful when probing a cache.

"""

from deltalake import DeltaTable

from deltalake.exceptions import TableNotFoundError

try:

dt = DeltaTable(self.uri, storage_options=self.delta_storage_options())

except TableNotFoundError:

if missing_ok:

return -1

raise

return dt.count()

def cleanup(self) -> None:

"""Remove the local temp directory, if any.

No-op on the CI/MinIO path: the bucket is ephemeral and the uuid

prefix prevents collisions, so leaked keys are acceptable.

"""

if self.local_dir is not None:

shutil.rmtree(self.local_dir, ignore_errors=True)

self.local_dir = None

def wait_for_condition(

description: str,

predicate_func,

timeout_s: float | None,

poll_interval_s: float,

) -> None:

"""Poll ``predicate_func`` until it returns truthy or the timeout elapses.

:param description: Human-readable description used in timeout/errors.

:param predicate_func: Callable returning ``True`` when condition is met.

:param timeout_s: Maximum wait time in seconds. ``None`` means wait forever.

:param poll_interval_s: Poll interval in seconds.

:raises TimeoutError: If the condition is not met within ``timeout_s``.

"""

if timeout_s is not None and poll_interval_s > timeout_s:

raise ValueError(

f"poll interval ({poll_interval_s}s) cannot be larger than"

f" timeout ({timeout_s}s)"

)

timestamp_start_s = time.monotonic()

timestamp_deadline_s = (

timestamp_start_s + timeout_s if timeout_s is not None else float("inf")

)

attempt = 0

while True:

if time.monotonic() > timestamp_deadline_s:

raise TimeoutError(

f"timeout ({timeout_s:.1f}s) waiting for condition '{description}'"

)

attempt += 1

if predicate_func():

logging.debug(

f"condition '{description}' met after"

f" {time.monotonic() - timestamp_start_s:.1f}s"

)

return

time.sleep(poll_interval_s)