From d13ce2eeea4a05e23532002586928088118c789b Mon Sep 17 00:00:00 2001 From: Arseniy Dugin Date: Mon, 13 Jul 2026 14:00:53 +0000 Subject: [PATCH] feat(pipeline-run): stream execution logs Add --stream to tangle sdk pipeline-runs logs to follow container logs live instead of fetching a one-shot snapshot. Opening the stream has its own bounded retry budget for transport errors and retryable 5xx responses, spent before the first line is yielded. An established stream is never reopened, so output cannot be duplicated; mid-stream drops, open failures, BrokenPipe and Ctrl-C all exit without tracebacks. --- packages/tangle-cli/src/tangle_cli/client.py | 164 ++++++- .../src/tangle_cli/pipeline_run_manager.py | 64 ++- .../src/tangle_cli/pipeline_runs_cli.py | 23 + tests/test_client.py | 426 ++++++++++++++++++ tests/test_pipeline_runs_cli.py | 202 +++++++++ 5 files changed, 865 insertions(+), 14 deletions(-) diff --git a/packages/tangle-cli/src/tangle_cli/client.py b/packages/tangle-cli/src/tangle_cli/client.py index 060c960..e1dac0f 100644 --- a/packages/tangle-cli/src/tangle_cli/client.py +++ b/packages/tangle-cli/src/tangle_cli/client.py @@ -9,7 +9,7 @@ from __future__ import annotations import time -from collections.abc import Iterable, Mapping +from collections.abc import Iterable, Iterator, Mapping from dataclasses import asdict, is_dataclass from email.utils import parsedate_to_datetime from typing import Any @@ -53,6 +53,14 @@ class TangleApiClient(GeneratedTangleApiOperations): _MAX_RATE_LIMIT_RETRIES = 3 _RATE_LIMIT_BACKOFF_SECONDS = 1.0 _MAX_RETRY_AFTER_SECONDS = 60.0 + # Opening a log stream retries transient failures (transport-open errors + # and retryable 5xx) with doubling backoff, spent entirely before any line + # is yielded. An already-open stream that drops is the caller's to handle; + # never re-opening an established stream means lines cannot be duplicated. + _RETRYABLE_STREAM_STATUSES = frozenset({500, 502, 503, 504}) + _MAX_STREAM_OPEN_ATTEMPTS = 7 + _STREAM_OPEN_BACKOFF_SECONDS = 1.0 + _MAX_STREAM_OPEN_BACKOFF_SECONDS = 30.0 def __init__( self, @@ -132,6 +140,12 @@ def _make_request( request_kwargs=kwargs, ) if response.status_code == 401: + # The 401 response is discarded by the auth-refresh retry. For a + # streamed request it is an open streamed connection, so close it + # before refreshing auth and issuing the second request to avoid + # leaking it. ``response.headers`` stays available after close. + if kwargs.get("stream"): + response.close() self._refresh_auth() response = self._request_with_rate_limit_retries( request_method, @@ -152,7 +166,7 @@ def _request_with_rate_limit_retries( params: Mapping[str, Any] | None, json_data: Any, extra_headers: Mapping[str, str] | None, - timeout: float, + timeout: float | tuple[float, float | None], request_kwargs: Mapping[str, Any], ) -> requests.Response: response: requests.Response | None = None @@ -168,6 +182,13 @@ def _request_with_rate_limit_retries( ) if response.status_code != 429 or attempt == self._MAX_RATE_LIMIT_RETRIES: return response + # The 429 response is discarded by the retry. For a streamed request + # it is an open streamed connection, so close it before the sleep to + # avoid holding it open while we wait. ``response.headers`` stays + # available after close, so ``_sleep_for_rate_limit`` can still read + # Retry-After. + if request_kwargs.get("stream"): + response.close() self._sleep_for_rate_limit(response, attempt) return response @@ -181,6 +202,14 @@ def _sleep_for_rate_limit(self, response: requests.Response, attempt: int) -> No self.logger.info(f"429 rate limited; retrying in {delay:.1f}s") time.sleep(delay) + def _sleep_for_stream_open_retry(self, backoff: float, next_attempt: int, reason: str) -> None: + delay = min(backoff, self._MAX_STREAM_OPEN_BACKOFF_SECONDS) + self.logger.warn( + f"transient {reason} opening log stream; retrying in {delay:.1f}s " + f"(attempt {next_attempt}/{self._MAX_STREAM_OPEN_ATTEMPTS})" + ) + time.sleep(delay) + @staticmethod def _retry_after_delay(value: str | None) -> float | None: if not value: @@ -205,7 +234,7 @@ def _request_with_same_origin_redirects( params: Mapping[str, Any] | None, json_data: Any, extra_headers: Mapping[str, str] | None, - timeout: float, + timeout: float | tuple[float, float | None], request_kwargs: Mapping[str, Any], ) -> requests.Response: """Send one request, following only same-origin redirects. @@ -235,6 +264,16 @@ def _request_with_same_origin_redirects( **request_kwargs, ) if self.verbose: + # For streamed responses, reading ``response.text`` would buffer + # the entire body here, defeating callers that stream via + # ``iter_content``/``iter_lines``; a followed container-log + # stream may never terminate. Log a placeholder and leave the + # body unread. + response_body = ( + "" + if request_kwargs.get("stream") + else response.text + ) log_http_exchange( self.logger, method=current_method, @@ -243,7 +282,7 @@ def _request_with_same_origin_redirects( request_body=current_json, response_status=response.status_code, response_headers=dict(response.headers), - response_body=response.text, + response_body=response_body, ) if response.status_code not in self._REDIRECT_STATUSES: return response @@ -254,6 +293,13 @@ def _request_with_same_origin_redirects( next_url = urljoin(response.url, location) if not self._same_origin(response.url, next_url): + # Close before raising so callers that catch this and fall back + # to another route (or a streamed open) do not leak the open + # streamed redirect response and its pooled connection. + try: + response.close() + except Exception: + pass raise requests.HTTPError( f"Refusing to follow cross-origin redirect from {response.url} to {next_url}", response=response, @@ -358,16 +404,108 @@ def get_execution_details(self, execution_id: str) -> GetExecutionInfoResponse: return details def stream_execution_container_log(self, execution_id: str) -> requests.Response: - response = self._make_request( - "GET", - self._format_path( - "/api/executions/{id}/stream_container_log", - {"id": execution_id}, - ), - stream=True, + """Open the streaming container-log response for ``execution_id``. + + The endpoint delivers raw log lines over a long-lived chunked HTTP + response; the client streams those lines as-is and does no + event-protocol parsing. + + Establishing the stream (open + status check) follows a transient-error + retry budget: transport-open errors (connection/timeout) and retryable + 5xx responses are retried with exponential backoff before any line is + read. Same-origin redirect protection errors (cross-origin ``HTTPError`` + / ``TooManyRedirects``) are not transport blips and propagate + immediately. Once the stream is open the caller owns the response and + must close it; :meth:`iter_execution_container_log_lines` does that. + + The request uses ``(connect, read)`` timeouts of ``(self.timeout, + None)``: the connect timeout still bounds opening the stream, but there + is no per-read timeout, because a healthy follow stream stays silent for + as long as the container emits no output. + """ + + path = self._format_path( + "/api/executions/{id}/stream_container_log", + {"id": execution_id}, ) - response.raise_for_status() - return response + backoff = self._STREAM_OPEN_BACKOFF_SECONDS + last_exc: requests.RequestException | None = None + last_error_response: requests.Response | None = None + for attempt in range(1, self._MAX_STREAM_OPEN_ATTEMPTS + 1): + try: + response = self._make_request( + "GET", path, stream=True, timeout=(self.timeout, None) + ) + except (requests.HTTPError, requests.TooManyRedirects) as exc: + # Same-origin redirect guard errors carry the rejected streamed + # response and are intentionally not retried. No iterator ever + # receives that response, so close it before re-raising. + if exc.response is not None: + exc.response.close() + raise + except (requests.ConnectionError, requests.Timeout) as exc: + last_exc = exc + last_error_response = None + if attempt == self._MAX_STREAM_OPEN_ATTEMPTS: + break + self._sleep_for_stream_open_retry(backoff, attempt + 1, type(exc).__name__) + backoff *= 2.0 + continue + if response.status_code in self._RETRYABLE_STREAM_STATUSES: + response.close() + last_error_response = response + last_exc = None + if attempt == self._MAX_STREAM_OPEN_ATTEMPTS: + break + self._sleep_for_stream_open_retry( + backoff, attempt + 1, f"HTTP {response.status_code}" + ) + backoff *= 2.0 + continue + try: + response.raise_for_status() + except requests.HTTPError: + # Non-retryable status (e.g. 400/403/404): close the open + # streamed response before propagating so it is not leaked. + response.close() + raise + return response + if last_exc is not None: + raise last_exc + if last_error_response is not None: + last_error_response.raise_for_status() + # Defensive: every exhausted attempt records either a transport error + # (re-raised above) or a retryable-status response (raise_for_status + # always raises for those), so this cannot be reached. + raise RuntimeError( # pragma: no cover + "log stream open retries exhausted without a failure to re-raise" + ) + + def iter_execution_container_log_lines(self, execution_id: str) -> Iterable[str]: + """Return an iterator of decoded container-log lines for ``execution_id``. + + The stream is opened eagerly, so open failures (HTTP status or transport + errors) raise from this call rather than on first iteration; anything + raised while iterating is a drop of an already-open stream. The + underlying streaming response is always closed when iteration finishes + or the consumer stops early. + """ + + response = self.stream_execution_container_log(execution_id) + + def lines() -> Iterator[str]: + try: + # Decode whole ``bytes`` lines as UTF-8 explicitly rather than + # via ``decode_unicode=True``: requests' charset guessing falls + # back to latin-1 when the response declares no charset and + # would mojibake non-ASCII output. Whole-line decoding also + # reassembles multibyte sequences split across stream chunks. + for raw in response.iter_lines(): + yield raw.decode("utf-8", "replace") + finally: + response.close() + + return lines() def get_component_spec(self, digest: str) -> ComponentSpec: """Return a parsed domain component spec from the generated component endpoint.""" 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..66e17a1 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py @@ -15,12 +15,13 @@ import re import time import uuid -from collections.abc import Callable +from collections.abc import Callable, Iterable from contextlib import AbstractContextManager, nullcontext from dataclasses import dataclass, field from pathlib import Path from typing import Any, Mapping +import requests import yaml from .handler import TangleCliHandler @@ -54,6 +55,27 @@ class AmbiguousPipelineRunRecoveryError(PipelineRunError): """Raised when submit recovery finds multiple runs for one submission id.""" +def _describe_http_error(exc: requests.HTTPError) -> str: + """Summarize an HTTP status failure with the attempted request target. + + ``str(exc)`` alone can omit the URL (e.g. errors constructed by client + helpers), so read the status, reason, and method/URL off the attached + response when present. + """ + + response = exc.response + if response is None: + return str(exc) + request = response.request + target = ( + f"{request.method} {request.url}" + if request is not None and request.url + else (response.url or "Tangle API") + ) + reason = f" {response.reason}" if response.reason else "" + return f"HTTP {response.status_code}{reason} for {target}" + + @dataclass class PipelineSubmitPayload: """Prepared submit payload state before calling ``pipeline_runs_create``. @@ -599,6 +621,14 @@ def fetch_logs(self, client: Any, execution_id: str) -> Any: """Hook for alternate TD log providers; OSS uses the Tangle API only.""" return client.executions_container_log(execution_id) + def stream_logs(self, client: Any, execution_id: str) -> Iterable[str]: + """Hook for alternate TD log providers; OSS streams via the Tangle API. + + Failures to open the stream must raise from this call; exceptions raised + while iterating are reported as mid-stream interruptions. + """ + return client.iter_execution_container_log_lines(execution_id) + @dataclass class PipelineRunManager(TangleCliHandler): @@ -1261,6 +1291,38 @@ def graph_state_output(self, run_ids: list[str], *, timeout: float = 30.0) -> di def logs(self, execution_id: str) -> dict[str, Any]: return self.to_plain(self.hooks.fetch_logs(self.client, execution_id)) + def stream_logs(self, execution_id: str) -> Iterable[str]: + try: + lines = self.hooks.stream_logs(self.client, execution_id) + except requests.HTTPError as exc: + # A definitive non-2xx answer to the stream-open request (404 for a + # missing execution or endpoint, 403, ...): keep the status and the + # attempted target visible, since that is what the caller acts on. + raise PipelineRunError( + f"Failed to open log stream for execution {execution_id}: " + f"{_describe_http_error(exc)}" + ) from exc + except requests.RequestException as exc: + # Non-HTTP transport failures (connection refused, timeout) raise + # from the open call itself; surface them as a clean open failure. + raise PipelineRunError( + f"Failed to open log stream for execution {execution_id}: {exc}" + ) from exc + return self._relabel_stream_drops(lines, execution_id) + + @staticmethod + def _relabel_stream_drops(lines: Iterable[str], execution_id: str) -> Iterable[str]: + try: + yield from lines + except requests.RequestException as exc: + # The stream opened (the hook call succeeded), so a transport + # failure here is a drop of the live follow, possibly before the + # first line arrived. Surface it as an interruption rather than a + # fetch failure, which would wrongly imply the initial open failed. + raise PipelineRunError( + f"Log stream for execution {execution_id} was interrupted: {exc}" + ) from exc + def search_runs( self, *, diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py index 2da0eb5..bb9052b 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py @@ -3,7 +3,9 @@ from __future__ import annotations import json +import os import pathlib +import sys from typing import Annotated, Any from cyclopts import App, Parameter @@ -359,6 +361,14 @@ def pipeline_runs_wait( def pipeline_runs_logs( execution_id: str | None = None, *, + stream: Annotated[ + bool | None, + Parameter( + help="Follow the live log stream instead of fetching a one-shot snapshot. " + "The follow has no read timeout and stays open silently while the " + "container emits no output." + ), + ] = None, base_url: BaseUrlOption = None, token: TokenOption = None, auth_header: AuthHeaderOption = None, @@ -369,11 +379,24 @@ def pipeline_runs_logs( """Print Tangle API container logs for an execution id.""" specs = { "execution_id": (execution_id,), + "stream": (stream, None), "log_type": (log_type, "console"), **api_arg_specs(base_url=base_url, token=token, auth_header=auth_header, header=header), } def action(manager: PipelineRunManager, args: ArgsContainer) -> object: + if args.stream: + try: + for line in manager.stream_logs(args.execution_id): + print(line, flush=True) + except BrokenPipeError: + # The downstream reader closed the pipe (e.g. `... | head`). + # Point stdout at devnull so the interpreter's exit-time flush + # of the closed pipe cannot raise a second BrokenPipeError. + devnull_fd = os.open(os.devnull, os.O_WRONLY) + os.dup2(devnull_fd, sys.stdout.fileno()) + os.close(devnull_fd) + return None result = manager.logs(args.execution_id) if isinstance(result, dict) and isinstance(result.get("log_text"), str): print(result["log_text"], end="" if result["log_text"].endswith("\n") else "\n") diff --git a/tests/test_client.py b/tests/test_client.py index 5456ac0..881ac0f 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,13 +1,47 @@ from __future__ import annotations +import io +import json from types import SimpleNamespace +from typing import Any from unittest.mock import MagicMock +import pytest +import requests + import tangle_cli.client as client_module from tangle_cli.client import TangleApiClient +from tangle_cli.logger import CaptureLogger from tangle_cli.models import ComponentInfo +def _response(payload: Any = None, status_code: int = 200) -> requests.Response: + r = requests.Response() + r.status_code = status_code + if payload is None: + r._content = b"" + else: + r._content = json.dumps(payload).encode("utf-8") + r.headers["Content-Type"] = "application/json" + r.request = requests.Request("GET", "https://api.test").prepare() + return r + + +class _FakeSession: + def __init__(self, responses: list[requests.Response | Exception] | None = None) -> None: + self.calls: list[dict[str, Any]] = [] + self.responses = responses or [] + + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + self.calls.append({"method": method, "url": url, **kwargs}) + if self.responses: + next_response = self.responses.pop(0) + if isinstance(next_response, Exception): + raise next_response + return next_response + return _response({}) + + def test_find_existing_components_matches_exact_names_case_insensitively() -> None: client = TangleApiClient("https://api.test") client.list_published_component_infos = MagicMock( @@ -53,3 +87,395 @@ def fail_from_dict(*args, **kwargs): assert client.get_run_pipeline_spec("run-1") is task_spec client.executions_details.assert_called_once_with("root-exec-1") + + +def _tracked_stream_response(raw: Any, status_code: int = 200) -> requests.Response: + """A streaming-style response reading from ``raw`` that records ``close()`` in ``_closed``.""" + + r = requests.Response() + r.status_code = status_code + r.raw = raw + r.headers["Content-Type"] = "text/event-stream" + r.request = requests.Request("GET", "https://api.test").prepare() + r._closed = False + original_close = r.close + + def tracked_close() -> None: + r._closed = True + original_close() + + r.close = tracked_close # type: ignore[method-assign] + return r + + +def _stream_response(lines: list[bytes] | None = None, status_code: int = 200) -> requests.Response: + body = b"\n".join(lines) if lines else b"" + return _tracked_stream_response(io.BytesIO(body), status_code) + + +def test_stream_execution_container_log_yields_lines_and_closes() -> None: + stream = _stream_response([b"line-1", b"line-2", b"line-3"]) + session = _FakeSession([stream]) + client = TangleApiClient("https://api.test", session=session) + + lines = list(client.iter_execution_container_log_lines("exec-1")) + + assert lines == ["line-1", "line-2", "line-3"] + assert stream._closed is True + assert session.calls[0]["url"] == "https://api.test/api/executions/exec-1/stream_container_log" + assert session.calls[0]["stream"] is True + # The follow stream keeps the connect timeout but has no per-read timeout, + # so a healthy stream that is idle (container quiet) is never killed. + assert session.calls[0]["timeout"] == (client.timeout, None) + + +def test_stream_execution_container_log_closes_on_early_break() -> None: + stream = _stream_response([b"a", b"b", b"c"]) + session = _FakeSession([stream]) + client = TangleApiClient("https://api.test", session=session) + + gen = client.iter_execution_container_log_lines("exec-1") + assert next(iter(gen)) == "a" + gen.close() # type: ignore[union-attr] + + assert stream._closed is True + + +def test_stream_open_retries_transient_status_then_succeeds(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + bad = _stream_response(status_code=503) + ok = _stream_response([b"recovered"]) + session = _FakeSession([bad, ok]) + logger = CaptureLogger() + client = TangleApiClient("https://api.test", session=session, logger=logger) + + lines = list(client.iter_execution_container_log_lines("exec-1")) + + assert lines == ["recovered"] + assert bad._closed is True + assert sleeps == [1.0] + assert len(session.calls) == 2 + # Every stream-open retry sleep is announced through the client logger. + assert "transient HTTP 503 opening log stream; retrying in 1.0s (attempt 2/7)" in ( + logger.get_logs() or "" + ) + + +def test_stream_open_retries_transport_error_then_succeeds(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + ok = _stream_response([b"after-blip"]) + calls = {"n": 0} + + class FlakySession(_FakeSession): + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + calls["n"] += 1 + if calls["n"] == 1: + raise requests.ConnectionError("transient transport blip") + return ok + + client = TangleApiClient("https://api.test", session=FlakySession()) + + lines = list(client.iter_execution_container_log_lines("exec-1")) + + assert lines == ["after-blip"] + assert calls["n"] == 2 + assert sleeps == [1.0] + + +def test_stream_open_backoff_doubles_and_is_capped(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + attempts = TangleApiClient._MAX_STREAM_OPEN_ATTEMPTS + session = _FakeSession([_stream_response(status_code=503) for _ in range(attempts)]) + logger = CaptureLogger() + client = TangleApiClient("https://api.test", session=session, logger=logger) + + with pytest.raises(requests.HTTPError): + client.stream_execution_container_log("exec-1") + + assert sleeps == [1.0, 2.0, 4.0, 8.0, 16.0, 30.0] + # The last retry announces the final attempt; the exhausted 7th attempt + # raises without announcing an 8th. + logs = logger.get_logs() or "" + assert "(attempt 7/7)" in logs + assert "8/7" not in logs + + +def test_stream_open_raises_non_retryable_status_immediately(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + bad = _stream_response(status_code=404) + session = _FakeSession([bad]) + client = TangleApiClient("https://api.test", session=session) + + with pytest.raises(requests.HTTPError): + client.stream_execution_container_log("exec-1") + + # The streamed response must be closed before the non-retryable error + # propagates so the open connection is not leaked. + assert bad._closed is True + assert sleeps == [] + assert len(session.calls) == 1 + + +def test_stream_open_exhausts_retries_and_raises_last_status(monkeypatch) -> None: + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _seconds: None) + attempts = TangleApiClient._MAX_STREAM_OPEN_ATTEMPTS + session = _FakeSession([_stream_response(status_code=502) for _ in range(attempts)]) + client = TangleApiClient("https://api.test", session=session) + + with pytest.raises(requests.HTTPError) as exc_info: + client.stream_execution_container_log("exec-1") + + assert exc_info.value.response.status_code == 502 + assert len(session.calls) == attempts + + +def test_stream_open_exhausts_retries_and_raises_last_transport_error(monkeypatch) -> None: + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _seconds: None) + + class AlwaysFailingSession(_FakeSession): + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + raise requests.ConnectionError("permanent transport failure") + + client = TangleApiClient("https://api.test", session=AlwaysFailingSession()) + + with pytest.raises(requests.ConnectionError, match="permanent transport failure"): + client.stream_execution_container_log("exec-1") + + +def test_stream_open_cross_origin_redirect_is_not_retried(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + redirect = _stream_response(status_code=307) + redirect.url = "https://api.test/api/executions/exec-1/stream_container_log" + redirect.headers["Location"] = "https://attacker.example/leak" + session = _FakeSession([redirect]) + client = TangleApiClient("https://api.test", session=session) + + with pytest.raises(requests.HTTPError, match="cross-origin redirect") as exc_info: + client.stream_execution_container_log("exec-1") + + # Same-origin redirect protection must propagate immediately, not be retried. + assert sleeps == [] + assert len(session.calls) == 1 + # The rejected streamed response is attached to the guard error and no + # iterator ever receives it, so it must be closed before the error + # propagates to avoid leaking the open connection. + assert exc_info.value.response is redirect + assert redirect._closed is True + + +def test_stream_open_too_many_redirects_is_not_retried(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + + def same_origin_redirect() -> requests.Response: + r = _stream_response(status_code=307) + r.url = "https://api.test/api/executions/exec-1/stream_container_log" + r.headers["Location"] = "/api/executions/exec-1/stream_container_log" + return r + + redirect_calls = TangleApiClient._MAX_REDIRECTS + 1 + responses = [same_origin_redirect() for _ in range(redirect_calls)] + session = _FakeSession(list(responses)) + client = TangleApiClient("https://api.test", session=session) + + with pytest.raises(requests.TooManyRedirects) as exc_info: + client.stream_execution_container_log("exec-1") + + # One stream-open attempt that exhausts redirects; no retry of the open. + assert sleeps == [] + assert len(session.calls) == redirect_calls + # Every streamed redirect response must be closed; the final one is + # attached to the guard error and must not leak. + assert all(r._closed is True for r in responses) + assert exc_info.value.response is responses[-1] + + +def test_stream_open_verbose_does_not_read_streamed_body(monkeypatch) -> None: + monkeypatch.setenv("TANGLE_VERBOSE", "1") + stream = _stream_response([b"line-1", b"line-2"]) + text_reads: list[int] = [] + original_text = type(stream).text + + def tracked_text(self: requests.Response) -> str: + text_reads.append(1) + return original_text.fget(self) # type: ignore[attr-defined] + + monkeypatch.setattr(type(stream), "text", property(tracked_text)) + logger = CaptureLogger() + session = _FakeSession([stream]) + client = TangleApiClient("https://api.test", session=session, logger=logger) + + response = client.stream_execution_container_log("exec-1") + + # Verbose logging must not drain the streamed body before the caller can + # iterate it; the log stream stays readable. + assert text_reads == [] + assert response._closed is False + assert list(response.iter_lines()) == [b"line-1", b"line-2"] + logs = logger.get_logs() or "" + assert "" in logs + + +def test_rate_limit_retry_closes_streamed_response_before_sleep(monkeypatch) -> None: + closed_at_sleep: list[bool] = [] + rate_limited = _stream_response(status_code=429) + rate_limited.headers["Retry-After"] = "0" + ok = _stream_response([b"recovered"]) + + def tracking_sleep(_seconds: float) -> None: + # Record whether the 429 stream is already closed when the rate-limit + # sleep runs; it must not be held open during the sleep. + closed_at_sleep.append(rate_limited._closed) + + monkeypatch.setattr("tangle_cli.client.time.sleep", tracking_sleep) + closed_at_retry: list[bool] = [] + + class TrackingSession(_FakeSession): + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + # Record whether the prior 429 stream was already closed by the + # time the successful retry is issued. + if self.calls: + closed_at_retry.append(rate_limited._closed) + return super().request(method, url, **kwargs) + + session = TrackingSession([rate_limited, ok]) + client = TangleApiClient("https://api.test", session=session) + + response = client.stream_execution_container_log("exec-1") + + assert response is ok + assert len(session.calls) == 2 + # The intermediate 429 streamed response must be closed before sleeping and + # before the retry is issued. + assert closed_at_sleep == [True] + assert closed_at_retry == [True] + assert rate_limited._closed is True + assert list(response.iter_lines()) == [b"recovered"] + + +def test_auth_refresh_closes_streamed_response_before_retry() -> None: + unauthorized = _stream_response(status_code=401) + ok = _stream_response([b"authorized"]) + closed_at_refresh: list[bool] = [] + closed_at_retry: list[bool] = [] + + class RefreshingClient(TangleApiClient): + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.refreshes = 0 + + def _refresh_auth(self) -> None: + self.refreshes += 1 + # On the auth-refresh triggered by the 401, the streamed 401 + # response must already be closed (not held open during refresh). + if self.refreshes == 2: + closed_at_refresh.append(unauthorized._closed) + self.headers["Authorization"] = f"Bearer refreshed-{self.refreshes}" + + class TrackingSession(_FakeSession): + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + # Record whether the prior 401 stream was already closed by the + # time the successful retry is issued. + if self.calls: + closed_at_retry.append(unauthorized._closed) + return super().request(method, url, **kwargs) + + session = TrackingSession([unauthorized, ok]) + client = RefreshingClient("https://api.test", session=session) + + response = client._make_request("GET", "/api/users/me", stream=True) + + assert response is ok + assert client.refreshes == 2 + assert len(session.calls) == 2 + # The intermediate 401 streamed response must be closed before the auth + # refresh and before the retry is issued. + assert closed_at_refresh == [True] + assert closed_at_retry == [True] + assert unauthorized._closed is True + # The successful retry stream remains open and readable for the caller. + assert response._closed is False + assert list(response.iter_lines()) == [b"authorized"] + + +def test_stream_open_synthetic_http_error_from_make_request_is_not_retried(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + client = TangleApiClient("https://api.test", session=_FakeSession()) + calls = {"n": 0} + + def fake_make_request(*args: Any, **kwargs: Any) -> requests.Response: + calls["n"] += 1 + raise requests.HTTPError("redirect guard tripped") + + monkeypatch.setattr(client, "_make_request", fake_make_request) + + with pytest.raises(requests.HTTPError, match="redirect guard tripped"): + client.stream_execution_container_log("exec-1") + + assert sleeps == [] + assert calls["n"] == 1 + + +class _ScriptedRaw: + """A raw stream whose ``read`` replays scripted byte chunks/exceptions. + + Each ``read`` returns the next queued ``bytes`` chunk verbatim (ignoring the + requested size, so a multi-byte char can be split across reads) or raises a + queued exception, modelling a mid-stream transport failure. + """ + + def __init__(self, chunks: list[bytes | Exception]) -> None: + self._chunks = list(chunks) + + def read(self, _size: int = -1) -> bytes: + if not self._chunks: + return b"" + item = self._chunks.pop(0) + if isinstance(item, Exception): + raise item + return item + + def close(self) -> None: + self._chunks.clear() + + +def _scripted_stream_response(chunks: list[bytes | Exception]) -> requests.Response: + return _tracked_stream_response(_ScriptedRaw(chunks)) + + +def test_stream_decodes_multibyte_char_split_across_chunks() -> None: + # "café" and "日本語" each contain multi-byte UTF-8 sequences; feeding the + # stream one byte at a time splits those sequences across chunk reads. + # Decoding whole lines as UTF-8 must reassemble them rather than yield + # replacement characters or mojibake. + payload = "café\n日本語\n".encode("utf-8") + stream = _scripted_stream_response([payload[i : i + 1] for i in range(len(payload))]) + client = TangleApiClient("https://api.test", session=_FakeSession([stream])) + + lines = list(client.iter_execution_container_log_lines("exec-1")) + + assert lines == ["café", "日本語"] + assert stream._closed is True + + +def test_stream_read_error_mid_iteration_propagates_and_closes() -> None: + # Once the stream is open the retry budget is spent; a transport failure + # during iteration must propagate (not be retried or swallowed) and the + # streamed response must still be closed by the iterator's finally block. + stream = _scripted_stream_response( + [b"line-1\n", requests.exceptions.ChunkedEncodingError("connection broken mid-stream")] + ) + client = TangleApiClient("https://api.test", session=_FakeSession([stream])) + + gen = iter(client.iter_execution_container_log_lines("exec-1")) + assert next(gen) == "line-1" + with pytest.raises(requests.exceptions.ChunkedEncodingError, match="connection broken mid-stream"): + next(gen) + + assert stream._closed is True diff --git a/tests/test_pipeline_runs_cli.py b/tests/test_pipeline_runs_cli.py index b74b0ad..6d642df 100644 --- a/tests/test_pipeline_runs_cli.py +++ b/tests/test_pipeline_runs_cli.py @@ -1,13 +1,17 @@ from __future__ import annotations import copy +import io import json +import os +import sys from contextlib import nullcontext from pathlib import Path from types import SimpleNamespace from typing import Any import pytest +import requests import yaml from tangle_cli import cli, pipeline_run_manager, pipeline_runs_cli @@ -114,6 +118,10 @@ def executions_graph_execution_state(self, id: str) -> dict[str, Any]: def executions_container_log(self, id: str) -> dict[str, Any]: return {"log_text": f"logs for {id}\n"} + def iter_execution_container_log_lines(self, id: str): + yield f"stream {id} line 1" + yield f"stream {id} line 2" + def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: self.list_calls.append(kwargs) return {"pipeline_runs": [{"id": "run-1"}], "next_page_token": None} @@ -2790,3 +2798,197 @@ def cleanup_prepared_pipeline(self, preparation, *, error=None): # type: ignore assert cleaned == [(temp_effective_path, "Pipeline validation failed:\n - boom")] assert not temp_effective_path.exists() + + +def test_pipeline_runs_logs_stream_prints_lines(monkeypatch, capsys) -> None: + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "logs", "exec-7", "--stream"]) + + assert capsys.readouterr().out == "stream exec-7 line 1\nstream exec-7 line 2\n" + + +def test_pipeline_runs_logs_snapshot_is_default_when_stream_absent(monkeypatch, capsys) -> None: + class TrackingClient(FakeClient): + def __init__(self) -> None: + super().__init__() + self.streamed_ids: list[str] = [] + + def iter_execution_container_log_lines(self, id: str): + self.streamed_ids.append(id) + yield from () + + fake_client = TrackingClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "logs", "exec-1"]) + + assert capsys.readouterr().out == "logs for exec-1\n" + assert fake_client.streamed_ids == [] + + +def test_stream_logs_open_transport_error_is_clean() -> None: + class RefusingClient(FakeClient): + def iter_execution_container_log_lines(self, id: str): + raise requests.ConnectionError("connection refused") + + manager = pipeline_run_manager.PipelineRunManager(client=RefusingClient()) + + with pytest.raises( + PipelineRunError, match="Failed to open log stream for execution exec-1" + ): + manager.stream_logs("exec-1") + + +def test_pipeline_runs_logs_stream_open_failure_exits_cleanly(monkeypatch) -> None: + class RefusingClient(FakeClient): + def iter_execution_container_log_lines(self, id: str): + raise requests.ConnectionError("connection refused") + + fake_client = RefusingClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "logs", "exec-1", "--stream"]) + + assert "Failed to open log stream for execution exec-1" in str(exc_info.value) + assert "connection refused" in str(exc_info.value) + + +def test_pipeline_runs_logs_stream_broken_pipe_exits_cleanly(monkeypatch) -> None: + # Simulates `... logs --stream | head`: the downstream reader closes the + # pipe, so the first stdout write raises BrokenPipeError. The command must + # stop following and exit cleanly instead of tracebacking. + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + devnull_fd = os.open(os.devnull, os.O_WRONLY) + write_attempts = {"count": 0} + + class ClosedPipeStdout(io.TextIOBase): + def write(self, _text: str) -> int: + write_attempts["count"] += 1 + raise BrokenPipeError + + def fileno(self) -> int: + return devnull_fd + + monkeypatch.setattr(sys, "stdout", ClosedPipeStdout()) + app = cli.build_app() + + try: + run_app(app, ["sdk", "pipeline-runs", "logs", "exec-7", "--stream"]) + finally: + os.close(devnull_fd) + + assert write_attempts["count"] == 1 + + +def test_pipeline_runs_logs_stream_ctrl_c_exits_cleanly(monkeypatch, capsys) -> None: + # An interactive Ctrl-C is the normal way to stop a live follow; it must + # exit with the conventional interrupt code (128 + SIGINT), not a + # KeyboardInterrupt traceback. cyclopts provides this by default + # (suppress_keyboard_interrupt); this locks that contract in for --stream. + class InterruptedStreamClient(FakeClient): + def iter_execution_container_log_lines(self, id: str): + yield "line-1" + raise KeyboardInterrupt + + fake_client = InterruptedStreamClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "logs", "exec-1", "--stream"]) + + assert exc_info.value.code == 130 + assert capsys.readouterr().out == "line-1\n" + + +def test_pipeline_runs_logs_help_documents_stream(capsys) -> None: + app = cli.build_app() + run_app(app, ["sdk", "pipeline-runs", "logs", "--help"]) + assert "--stream" in capsys.readouterr().out + + +def test_pipeline_runs_stream_logs_uses_client_iterator() -> None: + fake_client = FakeClient() + manager = PipelineRunManager(client=fake_client) + + assert list(manager.stream_logs("exec-9")) == ["stream exec-9 line 1", "stream exec-9 line 2"] + + +def test_pipeline_runs_stream_logs_midstream_error_is_interruption() -> None: + # A transport failure after some lines have flowed must surface as a + # mid-stream interruption, not the "failed to open" wording reserved for a + # stream that never opened. + class MidStreamDropClient(FakeClient): + def iter_execution_container_log_lines(self, id: str): + yield "line-1" + raise requests.exceptions.ChunkedEncodingError("connection broken mid-stream") + + manager = PipelineRunManager(client=MidStreamDropClient()) + + gen = iter(manager.stream_logs("exec-9")) + assert next(gen) == "line-1" + with pytest.raises(PipelineRunError, match="interrupted") as exc_info: + next(gen) + assert "Failed to open" not in str(exc_info.value) + + +def test_pipeline_runs_stream_logs_zero_line_drop_is_interruption() -> None: + # A connection that opens successfully but drops before the first line is + # still a mid-stream interruption; only open failures (raised from the + # hook call itself) get the "failed to open" wording. + class ZeroLineDropClient(FakeClient): + def iter_execution_container_log_lines(self, id: str): + raise requests.exceptions.ChunkedEncodingError("Response ended prematurely") + yield # pragma: no cover + + manager = PipelineRunManager(client=ZeroLineDropClient()) + + with pytest.raises(PipelineRunError, match="interrupted") as exc_info: + list(manager.stream_logs("exec-9")) + assert "Failed to open" not in str(exc_info.value) + + +def test_pipeline_runs_logs_stream_honors_log_type_none(monkeypatch, capsys) -> None: + # --log-type controls the CLI's diagnostic logger and is orthogonal to + # --stream; the streamed log content still reaches stdout under any mode. + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "logs", "exec-7", "--stream", "--log-type", "none"]) + + assert capsys.readouterr().out == "stream exec-7 line 1\nstream exec-7 line 2\n" + + +def _logs_http_error(status: int, path: str) -> requests.HTTPError: + response = requests.Response() + response.status_code = status + response.request = requests.Request("GET", f"https://api.test{path}").prepare() + return requests.HTTPError(f"{status} Client Error", response=response) + + +def test_pipeline_runs_logs_stream_missing_endpoint_is_clean(monkeypatch) -> None: + class NoStreamClient(FakeClient): + def iter_execution_container_log_lines(self, id: str): + # Like the real client, open failures raise from the call itself + # rather than on first iteration. + raise _logs_http_error(404, f"/api/executions/{id}/stream_container_log") + + fake_client = NoStreamClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + # A clean SystemExit (not a raw requests.HTTPError) means no traceback. + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "logs", "exec-1", "--stream"]) + + message = str(exc_info.value) + assert "404" in message + assert "/api/executions/exec-1/stream_container_log" in message