Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 86 additions & 8 deletions packages/tangle-cli/src/tangle_cli/pipeline_dehydrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,48 @@

PATH_SEPARATOR = "|" # Use | as separator since task names can contain dots.

# Per-component filename limit shared by ext4, APFS and NTFS. It counts
# BYTES, not characters, so a multi-byte stem has to be measured encoded.
_MAX_FILENAME_BYTES = 255


def _truncate_utf8(text: str, budget: int) -> str:
"""Return the longest prefix of *text* that fits *budget* UTF-8 bytes.

Truncation lands on a character boundary: cutting the encoded form can
leave a partial multi-byte sequence, and dropping it keeps the result
decodable and deterministic.
"""
if budget <= 0:
return ""
encoded = text.encode("utf-8")
if len(encoded) <= budget:
return text
return encoded[:budget].decode("utf-8", "ignore")


class ComponentFilenameError(RuntimeError):
"""Base class for problems naming an extracted component file."""


class ComponentFilenameCollisionError(ComponentFilenameError):
"""Raised when two different component specs claim one extraction filename.

Unreachable while filenames carry a full content address; it exists so that
any future change which shortens or weakens the address fails loudly here
instead of silently overwriting a previously extracted component.
"""


class ComponentFilenameTooLongError(ComponentFilenameError):
"""Raised when the configured extension leaves no room for a filename.

The separator, content address and extension are reserved before the
readable stem, so only an extension longer than the limit itself can make
a valid name impossible. Reported here rather than as an ``OSError`` from
whichever writer happened to receive the oversized path.
"""


@dataclass(frozen=True)
class Jinja2ExportResult:
Expand Down Expand Up @@ -107,6 +149,7 @@ def __init__(

self.interactive = interactive
self._saved_components: dict[str, Path | str] = {}
self._component_filenames: dict[str, str] = {}
self._current_reference_file: Path | str | None = output_file
self._io = PipelineHydrator(
enable_resolution=False,
Expand Down Expand Up @@ -317,7 +360,7 @@ def _process_task(
new_task["componentRef"] = {"digest": resolved_digest}
self.log.info(" → Auto: Replaced with componentRef.digest (found in library)")
else:
file_url = self._save_component_to_file(name, resolved_digest, spec)
file_url = self._save_component_to_file(name, spec)
new_task["componentRef"] = {"url": file_url}
self.log.info(" → Auto: Extracted to file (no URL, not in library or no client)")
elif choice == DehydrateChoice.DIGEST:
Expand All @@ -330,7 +373,7 @@ def _process_task(
new_task["componentRef"] = {"url": canonical_url}
self.log.info(" → Replaced with componentRef.url")
elif choice == DehydrateChoice.FILE:
file_url = self._save_component_to_file(name, resolved_digest, spec)
file_url = self._save_component_to_file(name, spec)
new_task["componentRef"] = {"url": file_url}
self.log.info(f" → Extracted to {file_url}")
else:
Expand All @@ -344,20 +387,55 @@ def _safe_filename(self, name: str, fallback: str = "component") -> str:
safe_name = "".join(c for c in safe_name if c.isalnum() or c == "_")
return safe_name or fallback

def _save_component_to_file(self, name: str, digest: str, spec: dict[str, Any]) -> str:
"""Save a component spec once and return a reference URL for this file."""
def _component_filename(self, name: str, address: str) -> str:
"""Return ``<stem>-<address><extension>``, byte-bounded.

if digest not in self._saved_components:
filename = f"{self._safe_filename(name)}{self.component_extension}"
The readable stem takes whatever UTF-8 bytes the address and extension
leave; it is dropped, separator included, when nothing fits.
"""
stemless = f"{address}{self.component_extension}"
required = len(stemless.encode("utf-8"))
if required > _MAX_FILENAME_BYTES:
raise ComponentFilenameTooLongError(
f"component_extension is too long to build a component filename: the "
f"content address and extension need {required} bytes, over the "
f"{_MAX_FILENAME_BYTES}-byte limit"
)
# The separator is charged here rather than reserved above, so an
# extension that fits by exactly one byte is not rejected.
stem = _truncate_utf8(self._safe_filename(name), _MAX_FILENAME_BYTES - required - 1)
if not stem:
return stemless
# ``_safe_filename`` folds ``-`` to ``_``, so this separator cannot
# occur inside the stem.
return f"{stem}-{stemless}"

def _save_component_to_file(self, name: str, spec: dict[str, Any]) -> str:
"""Write a component spec once and return a reference URL for it."""

# Address the spec, never ``componentRef.digest``: that digest is a
# locator, not proof of content. The hydrator digests raw source TEXT,
# so equal specs can carry different valid digests, and an edited spec
# can still carry the stale digest of what it used to be.
address = utils.compute_spec_digest(spec)
if address not in self._saved_components:
filename = self._component_filename(name, address)
claimed_by = self._component_filenames.get(filename)
if claimed_by is not None and claimed_by != address:
raise ComponentFilenameCollisionError(
f"extracted component filename {filename!r} is already taken by a "
f"different component spec; refusing to overwrite it"
)
self._component_filenames[filename] = address
destination = self._join_destination(self.components_dir, filename)
self._write_text(destination, utils.dump_yaml(spec), kind="component")
if self._is_local_destination(destination):
destination_text = str(destination)
if destination_text.startswith("file://"):
destination_text = destination_text[7:]
destination = Path(destination_text).resolve()
self._saved_components[digest] = destination
return self._make_ref_url(self._saved_components[digest])
self._saved_components[address] = destination
return self._make_ref_url(self._saved_components[address])

def _make_ref_url(self, target: Path | str) -> str:
"""Create a componentRef URL for a saved target."""
Expand Down
128 changes: 116 additions & 12 deletions tests/test_pipeline_dehydrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

from tangle_cli import utils
from tangle_cli.pipeline_dehydrator import (
ComponentFilenameTooLongError,
DehydrateChoice,
Jinja2ExportResult,
PipelineDehydrator,
Expand All @@ -16,14 +17,19 @@
)


def _leaf_spec(name: str, *, canonical_url: str | None = None) -> dict[str, Any]:
def _leaf_spec(
name: str, *, canonical_url: str | None = None, marker: str | None = None
) -> dict[str, Any]:
annotations = {}
if canonical_url:
annotations["canonical_location"] = canonical_url
container: dict[str, Any] = {"image": "example/image:latest"}
if marker:
container["command"] = [marker]
return {
"name": name,
"metadata": {"annotations": annotations},
"implementation": {"container": {"image": "example/image:latest"}},
"implementation": {"container": container},
}


Expand All @@ -35,6 +41,32 @@ def _pipeline(tasks: dict[str, Any]) -> dict[str, Any]:
return {"name": "Pipeline", "implementation": {"graph": {"tasks": tasks}}}


def _extracted_name(name: str, *, canonical_url: str | None = None, marker: str | None = None) -> str:
"""Filename ``_save_component_to_file`` gives the leaf component *name*.

Derived rather than hard-coded so these expectations assert an exact
filename without being rewritten whenever a fixture spec changes.
"""
spec = _leaf_spec(name, canonical_url=canonical_url, marker=marker)
stem = name.lower().replace(" ", "_")
return f"{stem}-{utils.compute_spec_digest(spec)}.yaml"


def _file_ref(name: str, marker: str, digest: str | None = None) -> dict[str, Any]:
ref: dict[str, Any] = {"name": name, "spec": _leaf_spec(name, marker=marker)}
if digest is not None:
ref["digest"] = digest
return {"componentRef": ref}


def _extract(tmp_path: Path, tasks: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
dehydrator = PipelineDehydrator(
{"": DehydrateChoice.FILE}, output_file=tmp_path / "out.yaml", **kwargs
)
result = dehydrator.dehydrate(_pipeline(tasks))
return result["implementation"]["graph"]["tasks"]


class FakeClient:
def __init__(self, found: set[str]) -> None:
self.found = found
Expand Down Expand Up @@ -182,8 +214,9 @@ def _api_client(self):
).dehydrate(data)

tasks = result["implementation"]["graph"]["tasks"]
assert tasks["local"]["componentRef"] == {"url": "file://./components/local_only.yaml"}
saved_component = yaml.safe_load((tmp_path / "components" / "local_only.yaml").read_text(encoding="utf-8"))
extracted = _extracted_name("Local Only")
assert tasks["local"]["componentRef"] == {"url": f"file://./components/{extracted}"}
saved_component = yaml.safe_load((tmp_path / "components" / extracted).read_text(encoding="utf-8"))
assert saved_component["name"] == "Local Only"


Expand All @@ -206,9 +239,10 @@ def test_pipeline_dehydrator_auto_uses_url_digest_then_file(tmp_path: Path) -> N
tasks = result["implementation"]["graph"]["tasks"]
assert tasks["canonical"]["componentRef"] == {"url": "https://example.test/canonical.yaml"}
assert tasks["published"]["componentRef"] == {"digest": "digest-found"}
assert tasks["local"]["componentRef"] == {"url": "file://./components/local_only.yaml"}
extracted = _extracted_name("Local Only")
assert tasks["local"]["componentRef"] == {"url": f"file://./components/{extracted}"}
assert client.calls == ["digest-found", "digest-missing"]
saved_component = yaml.safe_load((tmp_path / "components" / "local_only.yaml").read_text(encoding="utf-8"))
saved_component = yaml.safe_load((tmp_path / "components" / extracted).read_text(encoding="utf-8"))
assert saved_component["name"] == "Local Only"


Expand All @@ -226,10 +260,11 @@ def test_pipeline_dehydrator_auto_extracts_subgraphs_and_rewrites_relative_urls(

subgraph_file = tmp_path / "subgraphs" / "nested_subgraph_0.yaml"
subgraph = yaml.safe_load(subgraph_file.read_text(encoding="utf-8"))
extracted = _extracted_name("Inner Leaf")
assert subgraph["implementation"]["graph"]["tasks"]["leaf"]["componentRef"] == {
"url": "file://./../components/inner_leaf.yaml"
"url": f"file://./../components/{extracted}"
}
assert yaml.safe_load((tmp_path / "components" / "inner_leaf.yaml").read_text(encoding="utf-8"))["name"] == "Inner Leaf"
assert yaml.safe_load((tmp_path / "components" / extracted).read_text(encoding="utf-8"))["name"] == "Inner Leaf"


def test_pipeline_dehydrator_exports_jinja2_template_and_config(tmp_path: Path) -> None:
Expand Down Expand Up @@ -280,13 +315,82 @@ def writer(_hydrator, uri, content, _context):
uri_writers={"mem": writer},
).dehydrate_file("mem://bucket/input.yaml", "mem://bucket/out/pipeline.yaml")

assert result["implementation"]["graph"]["tasks"]["leaf"]["componentRef"] == {
"url": "mem://bucket/out/components/remote_leaf.yaml"
}
assert yaml.safe_load(writes["mem://bucket/out/components/remote_leaf.yaml"])["name"] == "Remote Leaf"
extracted = f"mem://bucket/out/components/{_extracted_name('Remote Leaf')}"
assert result["implementation"]["graph"]["tasks"]["leaf"]["componentRef"] == {"url": extracted}
assert yaml.safe_load(writes[extracted])["name"] == "Remote Leaf"
assert yaml.safe_load(writes["mem://bucket/out/pipeline.yaml"]) == result


def test_pipeline_dehydrator_helper_exports_are_importable() -> None:
assert _extract_input_defaults({"inputs": {"User Name": {"default": "Ada"}}}) == {"user_name": "Ada"}
assert _build_subgraph_processing_queue(_pipeline({}))[0] == (0, "Pipeline")


def test_two_components_sharing_a_name_are_not_written_over_each_other(tmp_path: Path) -> None:
"""Extraction is addressed by spec content, not by display name.

Naming files after the component alone let the second component overwrite
the first and redirected both tasks onto the survivor. Both refs carry the
same digest here on purpose: a locator digest is not proof of content.
"""
stale = "d" * 64
tasks = _extract(
tmp_path,
{"a": _file_ref("Leaf", "first", stale), "b": _file_ref("Leaf", "second", stale)},
)

assert tasks["a"]["componentRef"]["url"] != tasks["b"]["componentRef"]["url"]
payloads = sorted(
yaml.safe_load(path.read_text(encoding="utf-8"))["implementation"]["container"]["command"][0]
for path in (tmp_path / "components").glob("*")
)
assert payloads == ["first", "second"]


@pytest.mark.parametrize("other_digest", ["b" * 64, "unknown", None])
def test_one_spec_is_written_once_whatever_digest_the_refs_carry(
tmp_path: Path, other_digest: str | None
) -> None:
"""Equal specs share one file, named from the canonical spec digest."""
tasks = _extract(
tmp_path,
{
"a": _file_ref("Leaf", "payload", "a" * 64),
"b": _file_ref("Leaf", "payload", other_digest),
},
)

expected = f"file://./components/{_extracted_name('Leaf', marker='payload')}"
assert tasks["a"]["componentRef"]["url"] == expected
assert tasks["b"]["componentRef"]["url"] == expected
assert len(list((tmp_path / "components").glob("*"))) == 1


@pytest.mark.parametrize(
"name,extension,rejected",
[
("\u754c" * 400, ".yaml", False), # 3 bytes per character, so truncate on bytes
("Leaf", "." + "e" * 190, False), # address + extension exactly fills the limit
("Leaf", "." + "e" * 191, True), # one byte over, no valid name exists
],
)
def test_extracted_filenames_respect_the_255_byte_limit(
tmp_path: Path, name: str, extension: str, rejected: bool
) -> None:
dehydrator = PipelineDehydrator(
{"": DehydrateChoice.FILE},
output_file=tmp_path / "out.yaml",
component_extension=extension,
)
data = _pipeline({"a": _file_ref(name, "payload")})

if rejected:
with pytest.raises(ComponentFilenameTooLongError, match="component_extension"):
dehydrator.dehydrate(data)
assert not list((tmp_path / "components").glob("*")), "nothing may be written"
return

dehydrator.dehydrate(data)
written = list((tmp_path / "components").glob("*"))
assert len(written) == 1 and written[0].is_file()
assert len(written[0].name.encode("utf-8")) <= 255
Loading