diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py index 24742fa..e992864 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py @@ -12,6 +12,7 @@ import copy import inspect import json +import math import re import time import uuid @@ -739,6 +740,25 @@ def normalize_submit_body_in_place(body: dict[str, Any]) -> dict[str, Any]: def is_terminal_status(status: str | None) -> bool: return bool(status and status.upper() in _TERMINAL_STATUSES) + @staticmethod + def _validate_prepared_body(body: dict[str, Any]) -> dict[str, Any]: + """Fail fast on malformed prepared bodies, returning the componentRef. + + Locator-style bodies (a componentRef with ``name``/``digest`` and no + inline ``spec``) are valid; only the ``root_task``/``componentRef`` + mappings themselves are required. + """ + + root_task = body.get("root_task") + if not isinstance(root_task, dict): + raise PipelineRunError("Prepared submit body must contain a 'root_task' mapping") + component_ref = root_task.get("componentRef") + if not isinstance(component_ref, dict): + raise PipelineRunError( + "Prepared submit body 'root_task' must contain a 'componentRef' mapping" + ) + return component_ref + @staticmethod def status_counts_from_run(run: Mapping[str, Any]) -> dict[str, int]: stats = run.get("execution_status_stats") @@ -1122,7 +1142,7 @@ def submit_prepared_body( notify_submit_error: bool = True, ) -> dict[str, Any]: self.normalize_submit_body_in_place(body) - pipeline_spec = body["root_task"]["componentRef"]["spec"] + pipeline_spec = self._validate_prepared_body(body).get("spec") submit_context = context or PipelineRunContext( pipeline_path=pipeline_path, start_time=time.time(), @@ -1452,6 +1472,36 @@ def _poll_run_status( execution_state_timings=execution_state_timings, ) + @staticmethod + def _validate_wait_params( + *, + max_wait: float | None, + poll_interval: float, + timeout_clock: str, + allow_zero_poll_interval: bool, + ) -> None: + """Validate wait/poll parameters, raising before any run is submitted. + + The CLI wait/run commands and the programmatic API share this path, so + error messages name both the parameter and its CLI flag. + """ + + # NaN/inf pass the sign checks below (NaN compares False, inf is + # "positive") and would become a never-firing deadline or a raw + # ValueError from time.sleep; reject them up front. An unbounded wait is + # spelled max_wait=None, not an explicit inf. + if max_wait is not None: + if not math.isfinite(max_wait): + raise PipelineRunError("max_wait (--max-wait) must be a finite number") + if max_wait < 0: + raise PipelineRunError("max_wait (--max-wait) must be non-negative") + if not math.isfinite(poll_interval): + raise PipelineRunError("poll_interval (--poll-interval) must be a finite number") + if poll_interval < 0 or (poll_interval == 0 and not allow_zero_poll_interval): + raise PipelineRunError("poll_interval (--poll-interval) must be positive") + if timeout_clock not in {"monotonic", "wall"}: + raise PipelineRunError("timeout_clock must be 'monotonic' or 'wall'") + def wait_for_completion( self, run_id: str, @@ -1467,12 +1517,12 @@ def wait_for_completion( wait_context = context or PipelineRunContext(run_id=run_id, start_time=time.time()) if exit_on_first_failure: wait_context.metadata["exit_on_first_failure"] = True - if max_wait is not None and max_wait < 0: - raise PipelineRunError("--max-wait must be non-negative") - if poll_interval < 0 or (poll_interval == 0 and not allow_zero_poll_interval): - raise PipelineRunError("--poll-interval must be positive") - if timeout_clock not in {"monotonic", "wall"}: - raise PipelineRunError("timeout_clock must be 'monotonic' or 'wall'") + self._validate_wait_params( + max_wait=max_wait, + poll_interval=poll_interval, + timeout_clock=timeout_clock, + allow_zero_poll_interval=allow_zero_poll_interval, + ) enforce_max_wait = max_wait is not None and self.hooks.should_enforce_max_wait(wait_context) poll_started_at = time.monotonic() deadline_now: Callable[[], float] = time.time if timeout_clock == "wall" else time.monotonic @@ -1751,6 +1801,15 @@ def _run_body_factory( if max_attempts < 1: raise PipelineRunError("max_attempts must be at least 1") + if wait: + # Validate wait/poll params up front so an invalid request never + # submits a run it can't wait on. + self._validate_wait_params( + max_wait=max_wait, + poll_interval=poll_interval, + timeout_clock=timeout_clock, + allow_zero_poll_interval=allow_zero_poll_interval, + ) last_error: Exception | None = None previous_context: PipelineRunContext | None = None attempts: list[PipelineRunContext] = [] @@ -1797,7 +1856,9 @@ def _run_body_factory( context.metadata["submission_id"] = submission_id if metadata_factory is not None: context.metadata.update(metadata_factory(attempt, previous_context, last_error)) - pipeline_spec = body.get("root_task", {}).get("componentRef", {}).get("spec") + # Validate before the tolerant spec extraction so a malformed body + # raises PipelineRunError instead of an AttributeError. + pipeline_spec = self._validate_prepared_body(body).get("spec") context.submit_body = body context.pipeline_spec = pipeline_spec if isinstance(pipeline_spec, dict) else None if context.pipeline_spec is not None: @@ -2076,6 +2137,111 @@ def body_factory( ) +def _build_default_manager(*, logger: Logger | None = None) -> PipelineRunManager: + """Construct a manager backed by the native ``TangleApiClient``. + + Imported lazily so the lightweight top-level package and local-only SDK + commands stay native-free. Credentials/base URL come from the standard + ``TangleApiClient`` environment defaults. + """ + + try: + from .client import TangleApiClient + except ModuleNotFoundError as exc: + # Catch both the top-level package and any missing submodule (e.g. a + # partially-installed ``tangle_api.generated``). + if exc.name is not None and (exc.name == "tangle_api" or exc.name.startswith("tangle_api.")): + raise PipelineRunError( + "Native generated Tangle API bindings are required to submit a prepared " + "body without an explicit client. Install tangle-cli[native], provide a " + "local tangle_api.generated package, or pass client=/manager=." + ) from exc + raise + + hooks = PipelineRunHooks(logger=logger) if logger is not None else PipelineRunHooks() + return PipelineRunManager(client=TangleApiClient(), hooks=hooks, logger=logger or hooks.logger) + + +def submit_and_wait_prepared_body( + body: dict[str, Any], + *, + manager: PipelineRunManager | None = None, + client: Any | None = None, + logger: Logger | None = None, + wait: bool = True, + max_wait: float | None = 600.0, + poll_interval: float = 10.0, + use_graph_state: bool = False, + allow_zero_poll_interval: bool = False, + timeout_clock: str = "monotonic", + exit_on_first_failure: bool = False, + metadata: dict[str, Any] | None = None, + submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS, +) -> dict[str, Any]: + """Submit an already-prepared API submit body and optionally wait for completion. + + Thin wrapper over :meth:`PipelineRunManager.run_prepared_body` for callers + that already hold a fully formed submit ``body`` (``{"root_task": {...}}``); + it reuses the existing submit/wait/poll and submit-recovery logic and never + mutates ``body``. Provide ``manager`` to reuse a configured manager, + ``client`` to wrap an existing API client, or neither to build a + :class:`tangle_cli.client.TangleApiClient` from the environment (requires + the native extra). ``logger`` applies only when this helper constructs the + manager (the ``client`` and default paths); a supplied ``manager`` keeps + its own configured logger and ``logger`` is ignored. + + Returns a dict with ``response``, ``run_id``, ``root_execution_id``, and + ``wait`` (when ``wait=True``); the result is JSON-serializable whenever the + API responses are. With ``wait=True``, a submit response carrying no run id + raises :class:`PipelineRunError`; use ``wait=False`` to inspect such + responses. + """ + + if manager is not None and client is not None: + raise PipelineRunError("Pass at most one of manager= or client=") + + if manager is None: + if client is not None: + hooks = PipelineRunHooks(logger=logger) if logger is not None else PipelineRunHooks() + manager = PipelineRunManager(client=client, hooks=hooks, logger=logger or hooks.logger) + else: + manager = _build_default_manager(logger=logger) + + raw = manager.run_prepared_body( + body, + wait=wait, + max_wait=max_wait, + poll_interval=poll_interval, + use_graph_state=use_graph_state, + allow_zero_poll_interval=allow_zero_poll_interval, + timeout_clock=timeout_clock, + exit_on_first_failure=exit_on_first_failure, + metadata=metadata, + submit_recovery_attempts=submit_recovery_attempts, + ) + + context: PipelineRunContext | None = raw.get("context") + result: dict[str, Any] = { + "response": raw.get("response"), + "run_id": context.run_id if context is not None else None, + "root_execution_id": context.root_execution_id if context is not None else None, + } + if "wait" in raw: + result["wait"] = raw["wait"] + elif wait: + # The underlying run skips waiting when the submit response carries no + # run id; surface that instead of silently returning without waiting. + raise PipelineRunError( + "Run was submitted but the submit response did not include a run id, " + "so completion cannot be awaited; use wait=False to inspect the response." + ) + return result + + +# Short public alias; ``submit_and_wait_prepared_body`` is the definition. +submit_and_wait = submit_and_wait_prepared_body + + def parse_key_value_entries(entries: list[str] | None) -> dict[str, str]: parsed: dict[str, str] = {} for entry in entries or []: diff --git a/tests/test_submit_and_wait_prepared_body.py b/tests/test_submit_and_wait_prepared_body.py new file mode 100644 index 0000000..04877e4 --- /dev/null +++ b/tests/test_submit_and_wait_prepared_body.py @@ -0,0 +1,461 @@ +from __future__ import annotations + +import builtins +import copy +import json +from typing import Any + +import pytest +from tangle_cli.logger import CaptureLogger +from tangle_cli.pipeline_run_manager import ( + PipelineRunError, + PipelineRunManager, + submit_and_wait_prepared_body, +) + +RUN_ID = "run-1" +ROOT_EXECUTION_ID = "exec-1" + +# The submit lifecycle injects a submission-id annotation into the (deep-copied) +# body before submit so post-failure recovery can find the created run. Strip it +# when comparing the submitted body to the caller's original. +_SUBMISSION_ANNOTATION_KEY = "tangle-cli/submission-id" + + +def _without_submission_annotation(body: dict[str, Any]) -> dict[str, Any]: + stripped = copy.deepcopy(body) + annotations = stripped.get("annotations") + if isinstance(annotations, dict): + annotations.pop(_SUBMISSION_ANNOTATION_KEY, None) + if not annotations: + stripped.pop("annotations", None) + return stripped + + +def _prepared_body() -> dict[str, Any]: + return { + "root_task": { + "componentRef": { + "spec": {"name": "Prepared", "implementation": {"graph": {"tasks": {}}}} + }, + "arguments": {"query": "value"}, + } + } + + +class _SubmitClient: + """Minimal API client: submit succeeds, status reports terminal SUCCEEDED.""" + + def __init__(self) -> None: + self.created: list[Any] = [] + self.get_calls: int = 0 + + def pipeline_runs_create(self, body: Any = None) -> dict[str, Any]: + self.created.append(copy.deepcopy(body)) + return {"id": RUN_ID, "root_execution_id": ROOT_EXECUTION_ID} + + def pipeline_runs_get(self, id: str, include_execution_stats: bool | None = None) -> dict[str, Any]: + self.get_calls += 1 + return { + "id": id, + "root_execution_id": ROOT_EXECUTION_ID, + "execution_summary": {"has_ended": True}, + "execution_status_stats": {"SUCCEEDED": 1}, + } + + +def _locator_body() -> dict[str, Any]: + """A fully-formed submit body that references a component by name/digest. + + This shape has no inline ``componentRef.spec``; the helper must submit it + verbatim instead of failing while trying to read a spec. + """ + return { + "root_task": { + "componentRef": {"name": "my-pipeline", "digest": "sha256:abc123"}, + "arguments": {"query": "value"}, + } + } + + +def test_submit_only_returns_run_metadata_without_wait() -> None: + client = _SubmitClient() + result = submit_and_wait_prepared_body(_prepared_body(), client=client, wait=False) + + assert result["run_id"] == RUN_ID + assert result["root_execution_id"] == ROOT_EXECUTION_ID + assert result["response"] == {"id": RUN_ID, "root_execution_id": ROOT_EXECUTION_ID} + assert "wait" not in result + assert client.get_calls == 0 + # No PipelineRunContext / attempts leak into the default output. + assert set(result) == {"response", "run_id", "root_execution_id"} + assert json.dumps(result) + + +def test_submit_and_wait_success_includes_serializable_wait_result() -> None: + client = _SubmitClient() + result = submit_and_wait_prepared_body( + _prepared_body(), client=client, wait=True, poll_interval=0.01 + ) + + assert result["run_id"] == RUN_ID + assert result["wait"]["status"] == "SUCCEEDED" + assert result["wait"]["timed_out"] is False + assert client.get_calls >= 1 + assert json.dumps(result) + + +def test_timeout_metadata_preserved_and_serializable() -> None: + class _NeverTerminalClient(_SubmitClient): + def pipeline_runs_get( + self, id: str, include_execution_stats: bool | None = None + ) -> dict[str, Any]: + self.get_calls += 1 + return { + "id": id, + "root_execution_id": ROOT_EXECUTION_ID, + "execution_summary": {"has_ended": False}, + "execution_status_stats": {"RUNNING": 1}, + } + + result = submit_and_wait_prepared_body( + _prepared_body(), + client=_NeverTerminalClient(), + wait=True, + max_wait=0.0, + poll_interval=0.01, + ) + + assert result["run_id"] == RUN_ID + assert result["wait"]["timed_out"] is True + assert "status_counts" in result["wait"] + assert json.dumps(result) + + +def test_exit_on_first_failure_returns_serializable_early_exit() -> None: + class _FailingGraphClient(_SubmitClient): + # Run stays nonterminal at the top level; graph state reports a FAILED + # child alongside a still-running one, so exit_on_first_failure trips. + def pipeline_runs_get( + self, id: str, include_execution_stats: bool | None = None + ) -> dict[str, Any]: + self.get_calls += 1 + return { + "id": id, + "root_execution_id": ROOT_EXECUTION_ID, + "execution_status_stats": {"RUNNING": 1}, + } + + def executions_graph_execution_state(self, id: str) -> dict[str, Any]: + return {"status_totals": {"RUNNING": 1, "FAILED": 1}} + + result = submit_and_wait_prepared_body( + _prepared_body(), + client=_FailingGraphClient(), + wait=True, + use_graph_state=True, + exit_on_first_failure=True, + poll_interval=0.01, + ) + + assert result["run_id"] == RUN_ID + assert result["wait"]["early_exit"] is True + assert result["wait"]["timed_out"] is False + assert result["wait"]["failed_count"] == 1 + # Default output stays JSON-serializable and free of context/attempts leakage. + assert "context" not in result + assert "attempts" not in result + assert json.dumps(result) + + +def test_invalid_poll_interval_raises_before_submit() -> None: + client = _SubmitClient() + with pytest.raises(PipelineRunError): + submit_and_wait_prepared_body( + _prepared_body(), client=client, wait=True, poll_interval=0 + ) + # The run must never be submitted when the wait params are invalid. + assert client.created == [] + + +def test_negative_max_wait_raises_before_submit() -> None: + client = _SubmitClient() + with pytest.raises(PipelineRunError): + submit_and_wait_prepared_body( + _prepared_body(), client=client, wait=True, max_wait=-1.0 + ) + assert client.created == [] + + +@pytest.mark.parametrize("bad", [float("nan"), float("inf"), float("-inf")]) +def test_non_finite_max_wait_raises_before_submit(bad: float) -> None: + # NaN/inf pass sign checks (NaN compares False, inf is "positive") and + # would become a never-firing deadline; they must be rejected up front. + client = _SubmitClient() + with pytest.raises(PipelineRunError, match=r"max_wait \(--max-wait\) must be a finite number"): + submit_and_wait_prepared_body( + _prepared_body(), client=client, wait=True, max_wait=bad + ) + assert client.created == [] + + +@pytest.mark.parametrize("bad", [float("nan"), float("inf"), float("-inf")]) +def test_non_finite_poll_interval_raises_before_submit(bad: float) -> None: + client = _SubmitClient() + with pytest.raises(PipelineRunError, match=r"poll_interval \(--poll-interval\) must be a finite number"): + submit_and_wait_prepared_body( + _prepared_body(), client=client, wait=True, poll_interval=bad + ) + assert client.created == [] + + +def test_invalid_timeout_clock_raises_before_submit() -> None: + client = _SubmitClient() + with pytest.raises(PipelineRunError): + submit_and_wait_prepared_body( + _prepared_body(), + client=client, + wait=True, + poll_interval=0.01, + timeout_clock="bogus", + ) + assert client.created == [] + + +def test_wait_true_without_run_id_raises_instead_of_silent_no_wait() -> None: + class _NoRunIdClient(_SubmitClient): + def pipeline_runs_create(self, body: Any = None) -> dict[str, Any]: + self.created.append(copy.deepcopy(body)) + return {"root_execution_id": ROOT_EXECUTION_ID} + + client = _NoRunIdClient() + with pytest.raises(PipelineRunError, match="did not include a run id"): + submit_and_wait_prepared_body( + _prepared_body(), client=client, wait=True, poll_interval=0.01 + ) + # The run was submitted; only the wait was refused. + assert len(client.created) == 1 + assert client.get_calls == 0 + + # wait=False keeps the id-less response inspectable. + result = submit_and_wait_prepared_body(_prepared_body(), client=client, wait=False) + assert result["run_id"] is None + assert "wait" not in result + + +@pytest.mark.parametrize( + "body", [{"arguments": {}}, {"root_task": None}, {"root_task": "not-a-mapping"}] +) +def test_body_without_root_task_mapping_fails_before_submit(body: dict) -> None: + client = _SubmitClient() + with pytest.raises(PipelineRunError, match="root_task"): + submit_and_wait_prepared_body(body, client=client, wait=False) + assert client.created == [] + + +@pytest.mark.parametrize( + "root_task", + [{"arguments": {}}, {"componentRef": None}, {"componentRef": ["not-a-mapping"]}], +) +def test_body_without_component_ref_mapping_fails_before_submit(root_task: dict) -> None: + client = _SubmitClient() + with pytest.raises(PipelineRunError, match="componentRef"): + submit_and_wait_prepared_body( + {"root_task": root_task}, client=client, wait=False + ) + assert client.created == [] + + +def test_caller_body_not_mutated() -> None: + body = _prepared_body() + original = copy.deepcopy(body) + submit_and_wait_prepared_body(body, client=_SubmitClient(), wait=True, poll_interval=0.01) + assert body == original + + +def test_manager_and_client_are_mutually_exclusive() -> None: + manager = PipelineRunManager(client=_SubmitClient()) + with pytest.raises(PipelineRunError): + submit_and_wait_prepared_body( + _prepared_body(), manager=manager, client=_SubmitClient() + ) + + +def test_locator_body_without_inline_spec_submits_original_body() -> None: + client = _SubmitClient() + body = _locator_body() + original = copy.deepcopy(body) + + result = submit_and_wait_prepared_body(body, client=client, wait=False) + + # The client receives the original locator body (modulo the submission-id + # annotation the submit lifecycle injects for post-failure recovery). + assert len(client.created) == 1 + assert _without_submission_annotation(client.created[0]) == original + assert result["run_id"] == RUN_ID + assert result["root_execution_id"] == ROOT_EXECUTION_ID + assert body == original + + +def test_manager_submit_prepared_body_accepts_locator_body() -> None: + """Regression for the shipped submit path (not just the new helper): + ``PipelineRunManager.submit_prepared_body`` used to raise ``KeyError`` on a + locator-style body with no inline ``componentRef.spec``; it must now submit + the body verbatim with a spec-less run context.""" + + client = _SubmitClient() + manager = PipelineRunManager(client=client) + body = _locator_body() + original = copy.deepcopy(body) + + response = manager.submit_prepared_body(body) + + # submit_prepared_body normalizes/submits the caller's body directly (no + # submission-id annotation injection, which lives in the run lifecycle). + assert client.created == [original] + assert response == {"id": RUN_ID, "root_execution_id": ROOT_EXECUTION_ID} + assert body == original + + +def test_partial_native_package_raises_actionable_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A partially-installed native package (missing ``tangle_api.generated``) + should surface the actionable install hint, not a raw ModuleNotFoundError.""" + + real_import = builtins.__import__ + + def fake_import( + name: str, + globals: Any = None, + locals: Any = None, + fromlist: Any = (), + level: int = 0, + ) -> Any: + # Intercept the lazy ``from .client import TangleApiClient`` and fail as + # if a native submodule were missing rather than the top-level package. + if level == 1 and name == "client" and fromlist and "TangleApiClient" in fromlist: + raise ModuleNotFoundError( + "No module named 'tangle_api.generated'", name="tangle_api.generated" + ) + return real_import(name, globals, locals, fromlist, level) + + monkeypatch.setattr(builtins, "__import__", fake_import) + + # No manager/client -> the default manager path performs the native import. + with pytest.raises(PipelineRunError, match="Native generated Tangle API bindings"): + submit_and_wait_prepared_body(_prepared_body()) + + +def test_explicit_manager_is_reused() -> None: + client = _SubmitClient() + manager = PipelineRunManager(client=client) + result = submit_and_wait_prepared_body( + _prepared_body(), manager=manager, wait=False + ) + assert result["run_id"] == RUN_ID + assert client.created # the provided manager's client handled the submit + + +class _RecoveringClient(_SubmitClient): + """Submit dies without a response although the run was actually created; + the run is then discoverable via the submission-id list lookup.""" + + def __init__(self) -> None: + super().__init__() + self.list_calls: list[dict[str, Any]] = [] + + def pipeline_runs_create(self, body: Any = None) -> dict[str, Any]: + self.created.append(copy.deepcopy(body)) + raise TimeoutError("submit connection dropped") + + def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: + self.list_calls.append(kwargs) + return { + "pipeline_runs": [{"id": RUN_ID, "root_execution_id": ROOT_EXECUTION_ID}] + } + + +def test_submit_failure_adopts_run_recovered_by_submission_id(monkeypatch) -> None: + # Recovery finds the already-created run by the injected submission-id + # annotation and adopts it instead of resubmitting a duplicate. + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda _delay: None) + client = _RecoveringClient() + + result = submit_and_wait_prepared_body(_prepared_body(), client=client, wait=False) + + assert result["run_id"] == RUN_ID + assert result["root_execution_id"] == ROOT_EXECUTION_ID + # Exactly one submit was attempted; the run was adopted, not resubmitted. + assert len(client.created) == 1 + submission_id = client.created[0]["annotations"][_SUBMISSION_ANNOTATION_KEY] + assert len(client.list_calls) == 1 + assert submission_id in client.list_calls[0]["filter_query"] + + +def test_submit_recovery_attempts_zero_disables_lookup(monkeypatch) -> None: + class _FailingClient(_SubmitClient): + def __init__(self) -> None: + super().__init__() + self.list_calls: list[dict[str, Any]] = [] + + def pipeline_runs_create(self, body: Any = None) -> dict[str, Any]: + self.created.append(copy.deepcopy(body)) + raise TimeoutError("submit connection dropped") + + def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: + self.list_calls.append(kwargs) + return {"pipeline_runs": []} + + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda _delay: None) + client = _FailingClient() + + with pytest.raises(TimeoutError): + submit_and_wait_prepared_body( + _prepared_body(), client=client, wait=False, submit_recovery_attempts=0 + ) + + assert client.list_calls == [] + + +def test_logger_used_when_helper_builds_the_manager(monkeypatch) -> None: + # The recovery path logs through the manager's logger, so it makes the + # logger= wiring observable: with client=, the helper must hand the caller's + # logger to the manager it builds. + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda _delay: None) + logger = CaptureLogger() + + result = submit_and_wait_prepared_body( + _prepared_body(), client=_RecoveringClient(), logger=logger, wait=False + ) + + assert result["run_id"] == RUN_ID + assert "Recovered existing pipeline run" in (logger.get_logs() or "") + + +def test_logger_ignored_when_manager_is_supplied(monkeypatch) -> None: + # Documented contract: a supplied manager keeps its own configured logger. + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda _delay: None) + manager_logger = CaptureLogger() + ignored_logger = CaptureLogger() + manager = PipelineRunManager(client=_RecoveringClient(), logger=manager_logger) + + result = submit_and_wait_prepared_body( + _prepared_body(), manager=manager, logger=ignored_logger, wait=False + ) + + assert result["run_id"] == RUN_ID + assert "Recovered existing pipeline run" in (manager_logger.get_logs() or "") + assert ignored_logger.get_logs() is None + + +def test_wait_with_max_wait_none_waits_without_deadline() -> None: + client = _SubmitClient() + + result = submit_and_wait_prepared_body( + _prepared_body(), client=client, wait=True, max_wait=None, poll_interval=0.01 + ) + + assert result["wait"]["status"] == "SUCCEEDED" + assert result["wait"]["timed_out"] is False + assert client.get_calls >= 1