Workflow run tests

CLAUDE.md

@@ -0,0 +1,116 @@
+# CLAUDE.md
+
+Guidance for working in this repository. Rules with stricter, active requirements live in `.claude/rules/` (coding-style, testing, heresy) and apply to all changes.
+
+## Project
+
+`cdm-data-loaders`: data loaders and wranglers for the KBase CDM. Python >= 3.13, managed and executed with `uv`. Package source is under `src/cdm_data_loaders/` (~115 files, 8 top-level packages). Tests mirror the source layout under `tests/cdm_data_loaders/` (including pipeline end-to-end suites under `tests/cdm_data_loaders/pipelines/`), with multi-phase NCBI FTP integration tests in `tests/integration/ncbi_ftp/` and vcrpy cassettes in `tests/cassettes/`.
+
+Autouse fixtures in `tests/conftest.py` (`_isolated_cli_and_env` and `_isolated_dlt_config` from `tests/dlt_config_isolation.py`) reset `sys.argv = ["pytest"]`, strip `CDL_*` environment variables, and snapshot/restore `dlt.config` between tests. Pipeline end-to-end tests use `PipelineTestRunner` (`tests/cdm_data_loaders/pipelines/framework.py`) to run pipelines across `ExecutionMode.CORE`, `ExecutionMode.CLI`, and `ExecutionMode.DOCKER` (Docker tests are marked `requires_docker` and require `PIPELINE_DOCKER_IMAGE` to be set).
+
+Useful invocations:
+
+- Standard local test run (skips Spark, Ceph, and unset Docker image tests): `uv run pytest -m "not requires_spark and not requires_ceph"`
+- Full pipelines suite, excluding the bulk downloader: `uv run pytest tests/cdm_data_loaders/pipelines --ignore=tests/cdm_data_loaders/pipelines/all_the_bacteria`
+- Unit vs. container test script: `./scripts/run_tests.sh [unit|container|all]`
+- Lint: `uv run ruff check` (line length 120, target py313, full rule selection in `pyproject.toml`).
+
+## Architecture
+
+Domain logic lives in the top-level packages and is orchestration-free; orchestration lives in `pipelines/`. Data flows source -> reader -> intermediate representation (IR) -> emitter -> target:
+
+- Converters: reader (source schema -> IR) + emitter (IR -> target format) + thin facade composing them. See `docs/converters_ir.md` for the full IR contract and package layout.
+- IR is a frozen, slotted node tree (`converters/core/ir.py`) that preserves source facts without choosing target policy; emitters make all target-specific decisions.
+- `converters/core/` imports no dlt, PyIceberg, or PySpark modules. The only cross-package import is the `x-xsv-config` contract reusing `X_XSV_CONFIG_SCHEMA` from `cdm_data_loaders.readers.jsonschema_xsv.xsv_validator.custom_metaschema`.
+
+Settings hierarchy: pipeline settings in `pipelines/` extend the hierarchy in `core/settings.py`: `CdmDataLoadersBase` -> `LoggerSettings` -> `LocalInputOutputSettings` -> `CtsSettings` -> `FileFormatIngestSettings`. Settings use `env_prefix="CDL_"`, `cli_kebab_case=True`, and `default_settings_with_shortcuts(...)`. `CdmDataLoadersBase.__pydantic_init_subclass__` calls `check_aliases()` at class creation time to verify that CLI argument names, aliases, and `cli_shortcuts` do not collide or claim reserved flags (`h`, `help`). CLI parsing is invoked explicitly via `pydantic_settings.CliApp.run` in `pipelines/core.py` (`PipelineCli` / `run_cli`), not implicitly on model instantiation. Every pipeline registers a `cli` entry point in `pyproject.toml` under `[project.scripts]` (and in `scripts/entrypoint.sh` for Docker execution): `all_the_bacteria`, `jsonl_extract`, `jsonl_jsonschema`, `jsonl_pydantic`, `ncbi_ftp_promote`, `ncbi_ftp_sync`, `ncbi_rest_api`, `pdb_manifest`, `uniprot`, `uniref`, `xmltodict`, `xml2db`, `xsv`.
+
+## Package Map
+
+All paths below are relative to `src/cdm_data_loaders/`.
+
+### `core/`
+
+Framework-level foundations:
+
+- `fields.py`: common annotated field types (`BufferSize`, `DatasetName`, `FileGlob`, `LoaderFileFormat`, `TableName`, `WriteDisposition`, etc.), `LoaderFileFormatEnum` (`parquet`, `jsonl`), and `DEFAULTS`.
+- `path_fields.py`: `LocalInputDir` and `LocalOutputDir` path types (`file://` or plain local directory paths) and URL/path validation helpers (`normalize_local_path`, `validate_directory`, `validate_url_protocol`).
+- `settings.py`: `pydantic-settings` base class hierarchy (`CdmDataLoadersBase` -> `LoggerSettings` -> `LocalInputOutputSettings` -> `CtsSettings` -> `FileFormatIngestSettings`), `CLI_SHORTCUTS`, `default_settings_with_shortcuts()`, and CLI flag/shortcut collision checking (`check_aliases()`).
+
+### `converters/`
+
+Schema-format translation. See `docs/converters_ir.md` for the full contract.
+
+- `core/`: shared, target-independent foundation
+  - `ir.py` (NodeType, Field, SchemaDocument, Reader/Emitter protocols), `ir_values.py` (`freeze_value` and friends), `errors.py` (ConversionError), `extensions.py` (ExtensionRegistry, DEFAULT_EXTENSIONS), `guards.py` (input guards), `inference.py` (type inference, decimal scale), `metadata.py` (draft-specific JSON Schema keyword selection), `io.py` (schema loading), `paths.py` (`set_nested`).
+- `readers/`: source -> IR
+  - `json_schema.py` (JsonSchemaReader), `dlt.py` (DltReader over dlt's normalized table forest), `iceberg.py` (IcebergReader).
+- `emitters/`: IR -> target
+  - `json_schema.py`, `dlt.py`, `pyspark.py`, `pyarrow.py`, `linkml.py`.
+- Facades: `jsonschema_to_dlt.py`, `jsonschema_to_pyspark.py`, `jsonschema_to_pyarrow.py`, `dlt_to_jsonschema.py`, `pyiceberg_to_jsonschema.py`, `pyiceberg_to_linkml.py`, plus CLI dumps `pyiceberg_catalog_to_jsonschema.py` and `pyiceberg_catalog_to_linkml.py`.
+
+### `readers/`
+
+File-format readers feeding pipelines:
+
+- `jsonlines.py`: JSONL line parsing (`parse_jsonl_line`), paged streaming (`stream_jsonl_lines`), and table routing (`route_jsonl_lines`).
+- `xml.py`: shared XML streaming and `xmltodict` helpers (`parse_head_matter`, `stream_xml_file`, `process_xml_file_to_dict`, `process_xml_file`, `process_xml_file_batches`, `process_xml_file_items`, and `build_xml_file_resource` over dlt's `filesystem` source).
+- `xml2db_doc.py`: `xml2db` model building (`build_xml2db_model`), bounded-memory chunk fragment streaming (`iter_xml2db_chunk_fragments`), primary/foreign key rewriting (content-addressed keys for reused tables, file/chunk-namespaced keys otherwise), document flattening (`flatten_xml2db_document`), and `process_xml_file_with_xml2db`.
+- `xsd.py`: XSD schema loading (`load_schema`) and parent/child element cardinality classification (`find_list_and_single_child_paths`).
+- `jsonschema_xsv/`: `xsv_reader.py` (casts validated CSV/TSV rows to JSON Schema property types and resolves `x-xsv-config` parsing parameters) plus `xsv_validator/` (`qsv.py` wraps the external `qsv` binary for two-pass validation, normalization, and null replacement; `custom_metaschema.py` defines the `https://json-schema.kbase.us/draft/2020-12-xsv/schema` metaschema and `X_XSV_CONFIG_SCHEMA`, also imported by `converters`; `schema_utils.py` and `helpers.py`).
+
+### `parsers/`
+
+Parsers for third-party outputs and formats:
+
+- `uniprot/`: `uniprot_kb.py` (UniProt XML), `uniref.py` (UniRef XML), `metalink.py`, `relnotes.py`, `idmapping.schema.json`; shared XML helpers in `utils/xml_utils.py`.
+- `ncbi/datasets_api/openapi_pydantic.py`: pydantic models for NCBI Datasets API responses (bundled `AnnotationReportResponse.schema.json` and `DatasetReportResponse.schema.json` alongside).
+- `ncbi/ftp_mod_times/`: `ftp_mod_times.schema.json` for NCBI FTP modification-time files.
+- `bbmap_stats.py` (bbmap `stats.sh` output), `checkm2.py` (quality reports), `pdb_holdings_file.py`.
+- `all_the_bacteria/schema/` and `gtdb/schema/`: JSON Schemas for ATB and GTDB tabular files (no Python modules).
+
+### `ncbi_ftp/`
+
+NCBI FTP assembly domain logic, phased: `manifest.py` (phase 1: assembly summary diffing and manifest generation), download is the phase 2 pipeline, `promote.py` (phase 3: promote staged files to final Lakehouse paths in S3), plus `assembly.py` (assembly-specific logic), `metadata.py` (frictionless data package descriptors), `constants.py`.
+
+### `pdb/`
+
+PDB pipeline constants, enums, and globals (`constants.py`); the pipeline itself is `pipelines/pdb_manifest.py`.
+
+### `pipelines/`
+
+dlt-based orchestrators; each exposes a `cli = PipelineCli(...)` registered in `pyproject.toml`:
+
+- `core.py`: `PipelineCli`, `run_cli`, `run_pipeline`, `setup_dlt_Pipeline`, `execute_dlt_pipeline`, and `configure_pipeline_logging`.
+- `all_the_bacteria.py`: All The Bacteria bulk data downloader (excluded from fast test runs).
+- `jsonlines/`: JSONL ingest pipelines (`settings.py`, shared machinery in `base.py`) - `extract_pipeline.py` (`jsonl_extract`, plain load into `<table_name>` / `<table_name>_invalid`), `extract_jsonschema_validate_pipeline.py` (`jsonl_jsonschema`, validates against `ENTITY_SCHEMAS`), `extract_pydantic_validate_pipeline.py` (`jsonl_pydantic`, validates against `ENTITY_MODELS`). See `docs/jsonschema_ingest.md`.
+- `ncbi_ftp_download.py` / `ncbi_ftp_promote.py`: phases 2 (`ncbi_ftp_sync`) and 3 (`ncbi_ftp_promote`) of the NCBI FTP flow.
+- `ncbi_rest_api.py`: NCBI REST API import.
+- `pdb_manifest.py`: PDB holdings transfer manifest.
+- `uniprot_kb.py` / `uniref.py`: UniProt (`uniprot`) and UniRef (`uniref`) dlt pipelines.
+- `xml2db/` / `xmltodict/` / `xsv/`: XML-to-relational, XML-to-dictionary, and XSV (CSV/TSV) ingestion, each with `pipeline.py` and `settings.py`; `xml2db/compaction.py` runs DuckDB-based post-load deduplication of "reused" tables on filesystem destinations.
+
+### `utils/`
+
+Cross-cutting helpers: `batcher.py` (`NumericFileSequenceBatcher`, `get_file_batches`), `buffer.py` (per-table row buffers for dlt: `DictBuffer`, `ListBuffer`, `ArrowDictBuffer`, `ArrowListBuffer`), `cdm_logger.py` (structured JSON logging), `checksums.py`, `gz.py` (`open_maybe_gzip`), `helpers.py`, `progress.py` (tqdm helpers), `spark.py`, `xml_utils.py`, `jsonschema/dereferencer.py` (`$ref` resolution), `dlt/dlt_normalization.py` (lossless parent-linked normalization of dlt table definitions, used by the converters' `DltReader`), and `file_transfer/` - `core.py` (tenacity-retried downloads), `sync_client.py` / `async_client.py` / `ftp_client.py`, and `s3/` (`client.py`, `object_utils.py`, HTTP-to-S3 streaming in `streamer.py`, `transfer_config.py`, `cli.py`).
+
+### Other Top-Level Directories
+
+- `scripts/`: `entrypoint.sh` (container entrypoint dispatching to registered pipeline CLIs), `run_tests.sh` (`unit`, `container`, or `all` test suites), `filter_ncbi_updates.py` (CLI filtering NCBI assembly IDs by modification date), `s3_local.py` (thin S3 CLI for local Ceph testing), `dtd2xsd.pl` (DTD-to-XSD converter), `run_tools.sh`, `parse_index.py`.
+- `docs/`: architecture and pipeline guides (`converters_ir.md`, `jsonschema_ingest.md`, `ncbi_ftp_e2e_walkthrough.md`, `pdb_e2e_walkthrough.md`, `rclone_instructions.md`, `uniprot.md`).
+- `cdl_skill/`: skill definition (`SKILL.md`) and generator (`scripts/skill_gen.sh`) for packaging pipeline documentation and schemas.
+
+## Conventions
+
+- Frozen, slotted dataclasses for models; constructors validate and recursively copy mappings.
+- CLI argument parsing by pydantic-settings
+- Fully type-annotated code (ruff `ANN`); datetime values carry an explicit timezone (`DTZ`); use `Decimal` for exact arithmetic, never constructed from floats.
+- External HTTP calls in tests are recorded and replayed with vcrpy; AWS is mocked with moto; Spark and Ceph tests run in containers via `docker-compose.yml` and are skipped by default (`-m "not requires_spark and not requires_ceph"`).
+- Tests are module-based `def test_*() -> None:` functions with pytest fixtures in `conftest.py`, named `test_<function>_<pass|fail>_<conditions>`.
+
+## Gotchas
+
+- dlt 1.30 DuckDB destination fails with a Binder Error "Ambiguous reference to catalog or schema" when the db filename stem equals the dataset name; keep the db filename distinct from both pipeline and dataset names.
+- dlt 1.30: `AnyDestination` is a quoted TypeAlias (a str at runtime), so `AnyDestination | None` in an evaluated annotation raises TypeError on import. Use `TDestinationReferenceArg` from `dlt.common.destination.reference`.
+- pyiceberg SqlCatalog engines leak sqlite connections; use the context-manager protocol (`with load_catalog(...) as catalog`) so `close()` disposes the engine.
+- pipelines are run on the KBase CTS (Task Service), which only saves the contents of the output directory (or dlt destination) if the container exits with exit code 0. For any other error code, the container STDERR and STDOUT are saved, but nothing else. As a result, even if pipelines experience failure (e.g. `load_info.has_failed_packages` is true), the container must exit 0 to ensure that logs, parsed data, and other artefacts are not deleted. Errors that occur before a pipeline is run (e.g. argument parsing errors) are not caught, so they should be picked up by the container STDERR/STDOUT.

pyproject.toml

@@ -29,7 +29,7 @@ dependencies = [
     "pandas>=3.0.3",
     "pydantic>=2.13.4",
     "pydantic-settings>=2.14.1",
-    "pyiceberg[pandas,pyarrow,pyiceberg-core,snappy]>=0.11.1",
+    "pyiceberg[pandas,pyarrow,pyiceberg-core,snappy]>=0.12",
     "smart-open>=8.0.1",
     "tqdm>=4.70.1",
     "xml2db>=0.14.0",
@@ -39,9 +39,9 @@ dependencies = [
 
 [project.scripts]
 all_the_bacteria = "cdm_data_loaders.pipelines.all_the_bacteria:cli"
-jsonl_extract = "cdm_data_loaders.pipelines.jsonlines.extract_pipeline:cli"
-jsonl_jsonschema = "cdm_data_loaders.pipelines.jsonlines.extract_jsonschema_validate_pipeline:cli"
-jsonl_pydantic = "cdm_data_loaders.pipelines.jsonlines.extract_pydantic_validate_pipeline:cli"
+jsonlines_extract = "cdm_data_loaders.pipelines.jsonlines.extract_pipeline:cli"
+jsonlines_jsonschema = "cdm_data_loaders.pipelines.jsonlines.extract_jsonschema_validate_pipeline:cli"
+jsonlines_pydantic = "cdm_data_loaders.pipelines.jsonlines.extract_pydantic_validate_pipeline:cli"
 ncbi_ftp_promote = "cdm_data_loaders.pipelines.ncbi_ftp_promote:cli"
 ncbi_ftp_sync = "cdm_data_loaders.pipelines.ncbi_ftp_download:cli"
 ncbi_rest_api = "cdm_data_loaders.pipelines.ncbi_rest_api:cli"
@@ -51,7 +51,7 @@ uniref = "cdm_data_loaders.pipelines.uniref:cli"
 xmltodict = "cdm_data_loaders.pipelines.xmltodict.pipeline:cli"
 xml2db = "cdm_data_loaders.pipelines.xml2db.pipeline:cli"
 xsv = "cdm_data_loaders.pipelines.xsv.pipeline:cli"
-
+hello_world = "cdm_data_loaders.utils.hello_world:cli"
 
 [dependency-groups]
 dev = [
@@ -75,6 +75,10 @@ models = [
 xml = [
     "xsdata[cli,lxml]>=26.2",
 ]
+mutants = [
+    "mutmut>=3.8.0",
+    "pytest-gremlins>=1.12.2",
+]
 
 [tool.ruff]
 line-length = 120
@@ -213,6 +217,12 @@ max-complexity = 15
 requires = ["uv_build>=0.12.0,<0.20.0"]
 build-backend = "uv_build"
 
+[tool.mutmut]
+source_paths = [ "src/" ]
+pytest_add_cli_args_test_selection= [ "-m", "not requires_spark", "-m", "not slow_test", "-m", "not requires_ceph", "-m", "not external_request", "-m", "not docker", "--ignore", "tests/cdm_data_loaders/pipelines/ncbi_rest_api/test_ncbi_rest_api.py" ]
+
+
+
 [tool.pytest]
 pythonpath = ["src"]
 log_cli = true
@@ -225,6 +235,7 @@ markers = [
     "slow_test: does what it says on the tin",
     "requires_ceph: end-to-end tests requiring a running CEPH instance and network access",
     "external_request: tests that make real network requests to external services (e.g. NCBI FTP)",
+    "docker: tests that build and run the pipeline inside a Docker container",
     "settings_cls(cls): the pydantic-settings class under test for alias parametrization",   
     "cli_fields: used for restricting cli-related tests to a subset of fields"
 ]
@@ -243,13 +254,18 @@ AWS_SECRET_ACCESS_KEY = "test_access_secret"
 # required for KBase Lakehouse env
 USER = "fake_user"
 KBASE_AUTH_TOKEN = "test-token-123"
-CDM_TASK_SERVICE_URL = "http://localhost:8080"
-BERDL_POD_IP = "192.168.1.100"
-SPARK_MASTER_URL = "spark://localhost:7077"
+
 BERDL_HIVE_METASTORE_URI = "thrift://localhost:9083"
-SPARK_CLUSTER_MANAGER_API_URL = "http://localhost:8000"
-GOVERNANCE_API_URL = "http://localhost:8000"
+# BERDL_POD_IP = "192.168.1.100"
+BERDL_POD_IP = "spark-notebook"
+CDM_TASK_SERVICE_URL = "http://localhost:8080"
 DATALAKE_MCP_SERVER_URL = "http://localhost:8080"
+GOVERNANCE_API_URL = "http://localhost:8000"
+POLARIS_CATALOG_URI = "http://polaris:8181/api/catalog"
+SPARK_CLUSTER_MANAGER_API_URL = "http://localhost:8000"
+SPARK_MASTER_URL = "spark://localhost:7077"
+# SPARK_MASTER_URL = "spark://spark-master:7077"
+DLT_PROJECT_DIR = "/tmp/dlt"
 
 [tool.uv.sources]
 cdm-schema = { git = "https://github.com/kbase/cdm-schema.git" }

src/cdm_data_loaders/pipelines/jsonlines/settings.py

@@ -22,9 +22,9 @@
     default_settings_with_shortcuts,
 )
 
-EXTRACT_PIPELINE_NAME: Final[str] = "jsonlines"
-PYDANTIC_PIPELINE_NAME: Final[str] = "jsonlines_pydantic_validator"
-JSONSCHEMA_PIPELINE_NAME: Final[str] = "jsonlines_jsonschema_validator"
+EXTRACT_PIPELINE_NAME: Final[str] = "jsonlines_extract"
+PYDANTIC_PIPELINE_NAME: Final[str] = "jsonlines_pydantic"
+JSONSCHEMA_PIPELINE_NAME: Final[str] = "jsonlines_jsonschema"
 
 ENTITY_MODELS_MODULE: Final[str] = "entity_models_module"
 SCHEMA_FILES_MODULE: Final[str] = "schema_files_module"

src/cdm_data_loaders/pipelines/ncbi_ftp_download.py

@@ -15,14 +15,15 @@
 from ftplib import error_temp
 from logging import Logger, getLogger
 from pathlib import Path, PurePosixPath
-from typing import Any
+from typing import Any, Final
 
 import tqdm
 from pydantic import AliasChoices, Field
+from pydantic_settings import SettingsConfigDict
 from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_exponential
 
 from cdm_data_loaders.core.fields import INPUT_MOUNT, OUTPUT_MOUNT
-from cdm_data_loaders.core.settings import LoggerSettings
+from cdm_data_loaders.core.settings import LoggerSettings, default_settings_with_shortcuts
 from cdm_data_loaders.ncbi_ftp.assembly import (
     FTP_HOST,
     build_accession_path,
@@ -36,13 +37,15 @@
 
 logger: Logger = getLogger(__name__)
 
-
+PIPELINE_NAME: Final[str] = "ncbi_ftp_sync"
 DEFAULT_STAGING_KEY_PREFIX: PurePosixPath = PurePosixPath("staging")
 
 
 class DownloadSettings(LoggerSettings):
     """Configuration for the NCBI FTP assembly download pipeline."""
 
+    model_config: SettingsConfigDict = default_settings_with_shortcuts(cli_prog_name=PIPELINE_NAME)
+
     manifest: Path = Field(
         default=Path(INPUT_MOUNT) / "transfer_manifest.txt",
         description="Path to the transfer manifest file listing FTP paths to download",

src/cdm_data_loaders/pipelines/ncbi_ftp_promote.py

@@ -6,14 +6,16 @@
 
 import logging
 from pathlib import Path, PurePosixPath
+from typing import Final
 
 from pydantic import AliasChoices, Field
-from pydantic_settings import CliImplicitFlag
+from pydantic_settings import CliImplicitFlag, SettingsConfigDict
 
-from cdm_data_loaders.core.settings import LoggerSettings
+from cdm_data_loaders.core.settings import LoggerSettings, default_settings_with_shortcuts
 from cdm_data_loaders.ncbi_ftp.promote import promote_from_s3
 from cdm_data_loaders.pipelines.core import PipelineCli
 
+PIPELINE_NAME: Final[str] = "ncbi_ftp_promote"
 DEFAULT_STAGING_BUCKET: PurePosixPath = PurePosixPath("cts")
 DEFAULT_DESTINATION_BUCKET: PurePosixPath = PurePosixPath("cdm-lake")
 DEFAULT_DESTINATION_PREFIX: PurePosixPath = PurePosixPath("tenant-general-warehouse/kbase/datasets/ncbi")
@@ -26,6 +28,8 @@
 class PromoteSettings(LoggerSettings):
     """Configuration for the NCBI FTP file promotion pipeline."""
 
+    model_config: SettingsConfigDict = default_settings_with_shortcuts(cli_prog_name=PIPELINE_NAME)
+
     staging_bucket: PurePosixPath = Field(
         default=DEFAULT_STAGING_BUCKET,
         description="Bucket where staged files are located after download",

src/cdm_data_loaders/pipelines/uniprot_kb.py

@@ -21,15 +21,15 @@
 )
 from cdm_data_loaders.readers.xml import build_xml_file_resource
 
-PIPELINE_NAME: Final[str] = "uniprot_kb"
+PIPELINE_NAME: Final[str] = "uniprot"
 UNIPROT_LOG_INTERVAL: Final[int] = 1000
 
 
 class UniProtSettings(CtsSettings):
     """Configuration for running the UniProt KB import pipeline."""
 
     model_config: SettingsConfigDict = default_settings_with_shortcuts(
-        cli_prog_name="uniprot", cli_shortcuts=CLI_SHORTCUTS
+        cli_prog_name=PIPELINE_NAME, cli_shortcuts=CLI_SHORTCUTS
     )
 
     log_interval: Annotated[