Skip to content
Closed
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
17 changes: 17 additions & 0 deletions crates/hm-dsl-engine/harmont-py/harmont/_decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

import re
import warnings
from functools import wraps
from typing import TYPE_CHECKING, Any

Expand Down Expand Up @@ -35,6 +36,7 @@ def pipeline(
allow_manual: bool = True,
env: dict[str, str] | None = None,
timeout: str | int | None = None,
default_image: str | None = None,
) -> Callable[[Callable[..., Any]], Callable[[], Any]]:
"""Register a function as a CI pipeline (decorator form).

Expand All @@ -58,6 +60,11 @@ def pipeline(
timeout: Whole-build wall-clock budget ("30m", "1h", or int
seconds). The build is killed and fails as timed out once it
elapses.
default_image: Deprecated. Root steps now default to
``ubuntu:24.04`` automatically; set a per-step ``image=`` (or
pass ``image=`` to ``hm.apt_base``) instead. When given, it is
still applied to root steps for back-compat and a
``DeprecationWarning`` is emitted.

Returns:
A decorator that registers the wrapped function and returns it
Expand All @@ -67,6 +74,15 @@ def pipeline(
ValueError: If ``slug`` does not match the allowed pattern.
"""

if default_image is not None:
warnings.warn(
"`default_image` is deprecated and will be removed in a future "
"release. Root steps now default to `ubuntu:24.04`; set a "
"per-step `image=` (or pass `image=` to `hm.apt_base`) instead.",
DeprecationWarning,
stacklevel=2,
)

def decorator(fn: Callable[..., Any]) -> Callable[[], Any]:
validate_target_signature(fn)
resolved = slug if slug is not None else fn.__name__ # ty: ignore[unresolved-attribute]
Expand All @@ -85,6 +101,7 @@ def wrapper() -> Any:
env=env,
fn=wrapper,
timeout=timeout,
default_image=default_image,
)
)
return wrapper
Expand Down
2 changes: 1 addition & 1 deletion crates/hm-dsl-engine/harmont-py/harmont/_envelope.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ def _render_one(
except TypeError as e:
msg = f"pipeline {reg.slug!r}: invalid return value\n → {e}"
raise TypeError(msg) from e
ir = _assemble(leaves, env=reg.env, timeout=reg.timeout)
ir = _assemble(leaves, env=reg.env, timeout=reg.timeout, default_image=reg.default_image)
resolve_pipeline_keys(
ir.get("graph", {}),
pipeline_org=pipeline_org,
Expand Down
12 changes: 10 additions & 2 deletions crates/hm-dsl-engine/harmont-py/harmont/_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ def pipeline(
*,
env: dict[str, str] | None = None,
timeout: str | int | None = None,
default_image: str | None = None,
) -> dict[str, Any]:
"""Top-level factory. Returns a JSON-shaped dict (version "0").

Expand All @@ -49,6 +50,11 @@ def pipeline(
``timeout`` is a whole-build wall-clock budget (``"30m"``, ``"1h"``,
or an int number of seconds). When it elapses the build is killed and
fails as *timed out*, regardless of how far the step graph got.

``default_image`` is deprecated: when given it overrides the root
stamp (in place of ``DEFAULT_IMAGE``) for back-compat. Prefer a
per-step ``image=``. The decorator (``@hm.pipeline``) emits the
deprecation warning; the factory applies it silently.
"""
if not leaves:
msg = (
Expand All @@ -59,14 +65,15 @@ def pipeline(
out: dict[str, Any] = {"version": "0"}
if timeout is not None:
out["timeout_seconds"] = parse_duration(timeout)
out["graph"] = _lower_to_graph(list(leaves), env=env)
out["graph"] = _lower_to_graph(list(leaves), env=env, default_image=default_image)
return out


def _lower_to_graph(
leaves: list[Step],
*,
env: dict[str, str] | None = None,
default_image: str | None = None,
) -> dict[str, Any]:
"""Walk back via `parent`, topo-sort, emit petgraph-serde graph dict.

Expand Down Expand Up @@ -158,9 +165,10 @@ def _lower_to_graph(
# explicit one. Root steps boot from an image tag (not a parent
# snapshot); child steps inherit the parent's committed snapshot and
# must stay image-less.
root_image = default_image if default_image is not None else DEFAULT_IMAGE
for i, node in enumerate(nodes):
if i not in has_builds_in_parent and "image" not in node["step"]:
node["step"]["image"] = DEFAULT_IMAGE
node["step"]["image"] = root_image

return {
"nodes": nodes,
Expand Down
1 change: 1 addition & 0 deletions crates/hm-dsl-engine/harmont-py/harmont/_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ class PipelineRegistration:
env: dict[str, str] | None
fn: Callable[[], object]
timeout: str | int | None = None
default_image: str | None = None


REGISTRATIONS: list[PipelineRegistration] = []
Expand Down
38 changes: 38 additions & 0 deletions crates/hm-dsl-engine/harmont-py/tests/test_decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,3 +104,41 @@ def a() -> hm.Step:
@hm.pipeline("ci")
def b() -> hm.Step:
return hm.scratch().sh("echo")


def test_default_image_deprecation_warns():
"""`default_image` is accepted for back-compat but warns (deprecated)."""
with pytest.warns(DeprecationWarning, match="default_image"):

@hm.pipeline("ci", default_image="custom:1")
def ci() -> hm.Step:
return hm.scratch().sh("echo")

assert REGISTRATIONS[0].default_image == "custom:1"


def test_no_default_image_does_not_warn(recwarn):
"""The common (migrated) path emits no deprecation noise."""

@hm.pipeline("ci")
def ci() -> hm.Step:
return hm.scratch().sh("echo")

assert not [w for w in recwarn if issubclass(w.category, DeprecationWarning)]
assert REGISTRATIONS[0].default_image is None


def test_default_image_applies_to_root_step():
"""A deprecated default_image still stamps root steps in the rendered IR."""
import json

with pytest.warns(DeprecationWarning):

@hm.pipeline("ci", default_image="custom:1")
def ci() -> hm.Step:
return hm.scratch().sh("echo hi", label="root")

envelope = json.loads(hm.dump_registry_json())
nodes = envelope["pipelines"][0]["definition"]["graph"]["nodes"]
root = next(n for n in nodes if n["step"].get("label") == "root")
assert root["step"]["image"] == "custom:1"
Loading