From 82692c7bc5273c6c570321b62457478fde16811f Mon Sep 17 00:00:00 2001 From: Datata1 Date: Sun, 12 Jul 2026 14:03:15 +0200 Subject: [PATCH] refactor(logs,landscape): unified log targets, injection-safe profile writes Implements plans/refactor/05-logs-landscape-hardening.md: - One stream/collect implementation instead of six: frozen target dataclasses (StageTarget, ServerTarget, ReplicaTarget) select the StreamOperation; WorkspaceLogManager.open/stream/collect take a LogTarget. The six legacy methods (stream_server, stream_replica, collect_server, collect_replica, open_*_stream) and the positional stream(stage, step) form remain as deprecated shims that warn and delegate (tested via pytest.warns) - save_profile no longer builds a shell heredoc from user content: YAML is transported base64-encoded (printf '%s' '' | base64 -d), so content containing PROFILE_EOF, , backticks, or quotes is inert; get/delete_profile filenames are single-quoted. Adversarial round-trip tests cover terminator/substitution/quote/CRLF/unicode payloads - LogStream takes the APIHttpClient and uses its new public stream() passthrough and stream_timeout property (configured connect timeout, unbounded read) instead of reaching into _get_client() with a hardcoded Timeout(5.0); SSE terminator event names are a module constant - LogEntry.kind is a LogKind enum (I/E) with str fallback for unknown values (left_to_right union) No breaking changes: all legacy method names keep working with a DeprecationWarning. --- src/codesphere/http_client.py | 21 +- .../workspace/landscape/resources.py | 13 +- .../resources/workspace/logs/__init__.py | 16 +- .../resources/workspace/logs/resources.py | 264 +++++++++++------- .../resources/workspace/logs/schemas.py | 10 +- .../workspace/landscape/test_landscape.py | 70 +++++ tests/resources/workspace/logs/test_logs.py | 181 +++++++++++- 7 files changed, 446 insertions(+), 129 deletions(-) diff --git a/src/codesphere/http_client.py b/src/codesphere/http_client.py index b8f85ff..7f2e102 100644 --- a/src/codesphere/http_client.py +++ b/src/codesphere/http_client.py @@ -1,6 +1,7 @@ import logging +from contextlib import AbstractAsyncContextManager from types import TracebackType -from typing import Any +from typing import Any, cast import httpx from pydantic import SecretStr @@ -38,6 +39,24 @@ def _get_client(self) -> httpx.AsyncClient: ) return self._client + @property + def timeout(self) -> httpx.Timeout: + """The configured request timeouts.""" + return self._timeout_config + + @property + def stream_timeout(self) -> httpx.Timeout: + """Timeouts for long-lived streams: configured connect, unbounded read.""" + # httpx.Timeout attributes are untyped upstream; connect is float | None + connect = cast("float | None", self._timeout_config.connect) + return httpx.Timeout(connect, read=None) + + def stream( + self, method: str, endpoint: str, **kwargs: Any + ) -> AbstractAsyncContextManager[httpx.Response]: + """Open a streaming request (e.g. SSE) on the underlying client.""" + return self._get_client().stream(method, endpoint, **kwargs) + async def open(self) -> None: if not self._client: self._client = httpx.AsyncClient(**self._client_config) diff --git a/src/codesphere/resources/workspace/landscape/resources.py b/src/codesphere/resources/workspace/landscape/resources.py index 68601e8..bbb9ef3 100644 --- a/src/codesphere/resources/workspace/landscape/resources.py +++ b/src/codesphere/resources/workspace/landscape/resources.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import base64 import logging import re from typing import TYPE_CHECKING @@ -75,16 +76,18 @@ async def save_profile(self, name: str, config: ProfileConfig | str) -> None: yaml_content = config.to_yaml() if isinstance(config, ProfileConfig) else config body = yaml_content if yaml_content.endswith("\n") else yaml_content + "\n" - await self._run_command( - f"cat > {filename} << 'PROFILE_EOF'\n{body}PROFILE_EOF\n" - ) + # Transport the content base64-encoded: the base64 alphabet is inert + # inside single quotes, so arbitrary YAML cannot break out of the + # shell command (the filename is regex-validated above). + encoded = base64.b64encode(body.encode("utf-8")).decode("ascii") + await self._run_command(f"printf '%s' '{encoded}' | base64 -d > '{filename}'") async def get_profile(self, name: str) -> str: - result = await self._run_command(f"cat {_profile_filename(name)}") + result = await self._run_command(f"cat '{_profile_filename(name)}'") return result.output async def delete_profile(self, name: str) -> None: - await self._run_command(f"rm -f {_profile_filename(name)}") + await self._run_command(f"rm -f '{_profile_filename(name)}'") async def deploy(self, profile: str | None = None) -> None: if profile is not None: diff --git a/src/codesphere/resources/workspace/logs/__init__.py b/src/codesphere/resources/workspace/logs/__init__.py index 9950129..176ecfd 100644 --- a/src/codesphere/resources/workspace/logs/__init__.py +++ b/src/codesphere/resources/workspace/logs/__init__.py @@ -1,10 +1,22 @@ -from .resources import LogStream, WorkspaceLogManager -from .schemas import LogEntry, LogProblem, LogStage +from .resources import ( + LogStream, + LogTarget, + ReplicaTarget, + ServerTarget, + StageTarget, + WorkspaceLogManager, +) +from .schemas import LogEntry, LogKind, LogProblem, LogStage __all__ = [ "LogEntry", + "LogKind", "LogProblem", "LogStage", "LogStream", + "LogTarget", + "ReplicaTarget", + "ServerTarget", + "StageTarget", "WorkspaceLogManager", ] diff --git a/src/codesphere/resources/workspace/logs/resources.py b/src/codesphere/resources/workspace/logs/resources.py index 129daaa..40e9808 100644 --- a/src/codesphere/resources/workspace/logs/resources.py +++ b/src/codesphere/resources/workspace/logs/resources.py @@ -3,13 +3,17 @@ import asyncio import json import logging +import warnings from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Any import httpx from ....core.base import WorkspaceScopedResource from ....core.operations import StreamOperation from ....exceptions import APIError, ValidationError, raise_for_status +from ....http_client import APIHttpClient from .operations import ( _STREAM_REPLICA_LOGS_OP, _STREAM_SERVER_LOGS_OP, @@ -19,31 +23,83 @@ log = logging.getLogger(__name__) +# SSE event names that terminate a stream. +_STREAM_END_EVENTS = frozenset({"end", "close", "done", "complete"}) + + +@dataclass(frozen=True, slots=True) +class StageTarget: + """Logs of a pipeline stage (prepare/test/run).""" + + stage: LogStage | str + step: int = 0 + + +@dataclass(frozen=True, slots=True) +class ServerTarget: + """Run logs of a specific server in a Multi Server Deployment.""" + + step: int + server: str + + +@dataclass(frozen=True, slots=True) +class ReplicaTarget: + """Run logs of a specific replica.""" + + step: int + replica: str + + +type LogTarget = StageTarget | ServerTarget | ReplicaTarget + + +def _resolve_target( + target: LogTarget, +) -> tuple[StreamOperation[LogEntry], dict[str, Any]]: + match target: + case StageTarget(stage=stage, step=step): + stage_value = stage.value if isinstance(stage, LogStage) else stage + return _STREAM_STAGE_LOGS_OP, {"stage": stage_value, "step": step} + case ServerTarget(step=step, server=server): + return _STREAM_SERVER_LOGS_OP, {"step": step, "server": server} + case ReplicaTarget(step=step, replica=replica): + return _STREAM_REPLICA_LOGS_OP, {"step": step, "replica": replica} + raise TypeError(f"Unsupported log target: {target!r}") + + +def _deprecated(old: str, new: str) -> None: + warnings.warn( + f"'{old}' is deprecated; use '{new}' instead.", + DeprecationWarning, + stacklevel=3, + ) + class LogStream: """Async context manager for streaming logs via SSE.""" def __init__( self, - client: httpx.AsyncClient, + http_client: APIHttpClient, endpoint: str, entry_model: type[LogEntry], timeout: float | None = None, ): - self._client = client + self._http_client = http_client self._endpoint = endpoint self._entry_model = entry_model self._timeout = timeout self._response: httpx.Response | None = None - self._stream_context = None + self._stream_context: Any = None async def __aenter__(self) -> LogStream: headers = {"Accept": "text/event-stream"} - self._stream_context = self._client.stream( + self._stream_context = self._http_client.stream( "GET", self._endpoint, headers=headers, - timeout=httpx.Timeout(5.0, read=None), + timeout=self._http_client.stream_timeout, ) self._response = await self._stream_context.__aenter__() @@ -86,7 +142,7 @@ async def _parse_sse_stream(self) -> AsyncIterator[LogEntry]: if event_type and data_buffer: data_str = "\n".join(data_buffer) - if event_type in ("end", "close", "done", "complete"): + if event_type in _STREAM_END_EVENTS: return if event_type == "problem": @@ -95,7 +151,7 @@ async def _parse_sse_stream(self) -> AsyncIterator[LogEntry]: for entry in self._parse_data(data_str): yield entry - elif event_type in ("end", "close", "done", "complete"): + elif event_type in _STREAM_END_EVENTS: return event_type = None @@ -145,104 +201,50 @@ def _handle_problem(self, data_str: str) -> None: class WorkspaceLogManager(WorkspaceScopedResource): - """Manager for streaming workspace logs via SSE.""" + """Manager for streaming workspace logs via SSE. - def _build_endpoint(self, operation: StreamOperation, **kwargs) -> str: - return operation.endpoint_template.format(id=self.workspace_id, **kwargs) + The unified API takes a :data:`LogTarget`: - def _open_stream( - self, - operation: StreamOperation, - timeout: float | None = None, - **kwargs, - ) -> LogStream: - endpoint = self._build_endpoint(operation, **kwargs) - return LogStream( - client=self._http_client._get_client(), - endpoint=endpoint, - entry_model=operation.entry_model, - timeout=timeout, - ) - - def open_stream( - self, - stage: LogStage | str, - step: int, - timeout: float | None = None, - ) -> LogStream: - """Open a log stream as an async context manager.""" - if isinstance(stage, LogStage): - stage = stage.value - return self._open_stream( - _STREAM_STAGE_LOGS_OP, timeout=timeout, stage=stage, step=step - ) + ```python + async for entry in workspace.logs.stream(StageTarget(LogStage.RUN, step=1)): + ... + entries = await workspace.logs.collect(ServerTarget(step=1, server="web")) + ``` + """ - def open_server_stream( - self, - step: int, - server: str, - timeout: float | None = None, - ) -> LogStream: - """Open a server log stream as an async context manager.""" - return self._open_stream( - _STREAM_SERVER_LOGS_OP, timeout=timeout, step=step, server=server - ) + def _build_endpoint(self, operation: StreamOperation, **kwargs: Any) -> str: + return operation.endpoint_template.format(id=self.workspace_id, **kwargs) - def open_replica_stream( - self, - step: int, - replica: str, - timeout: float | None = None, - ) -> LogStream: - """Open a replica log stream as an async context manager.""" - return self._open_stream( - _STREAM_REPLICA_LOGS_OP, timeout=timeout, step=step, replica=replica - ) + def open(self, target: LogTarget, timeout: float | None = None) -> LogStream: + """Open a log stream for ``target`` as an async context manager.""" + op, params = _resolve_target(target) + endpoint = self._build_endpoint(op, **params) + return LogStream(self._http_client, endpoint, op.entry_model, timeout=timeout) async def stream( self, - stage: LogStage | str, - step: int, + target: LogTarget | LogStage | str, + step: int | None = None, timeout: float | None = 30.0, ) -> AsyncIterator[LogEntry]: - """Stream logs for a given stage and step.""" - async with self.open_stream(stage, step, timeout) as stream: - async for entry in stream: - yield entry - - async def stream_server( - self, - step: int, - server: str, - timeout: float | None = None, - ) -> AsyncIterator[LogEntry]: - """Stream run logs for a specific server.""" - async with self.open_server_stream(step, server, timeout) as stream: - async for entry in stream: - yield entry - - async def stream_replica( - self, - step: int, - replica: str, - timeout: float | None = None, - ) -> AsyncIterator[LogEntry]: - """Stream run logs for a specific replica.""" - async with self.open_replica_stream(step, replica, timeout) as stream: + """Stream log entries for ``target`` as they arrive.""" + resolved = self._coerce_target(target, step, "stream") + async with self.open(resolved, timeout) as stream: async for entry in stream: yield entry async def collect( self, - stage: LogStage | str, - step: int, + target: LogTarget | LogStage | str, + step: int | None = None, max_entries: int | None = None, timeout: float | None = 30.0, ) -> list[LogEntry]: - """Collect all logs for a stage and step into a list.""" + """Collect log entries for ``target`` into a list.""" + resolved = self._coerce_target(target, step, "collect") entries: list[LogEntry] = [] try: - async with self.open_stream(stage, step, timeout) as stream: + async with self.open(resolved, timeout) as stream: async for entry in stream: entries.append(entry) if max_entries and len(entries) >= max_entries: @@ -251,6 +253,60 @@ async def collect( pass return entries + @staticmethod + def _coerce_target( + target: LogTarget | LogStage | str, step: int | None, method: str + ) -> LogTarget: + if isinstance(target, StageTarget | ServerTarget | ReplicaTarget): + return target + _deprecated(f"{method}(stage, step)", f"{method}(StageTarget(stage, step))") + return StageTarget(stage=target, step=step if step is not None else 0) + + # ------------------------------------------------------------------ + # Deprecated aliases (kept for backwards compatibility) + # ------------------------------------------------------------------ + + def open_stream( + self, stage: LogStage | str, step: int, timeout: float | None = None + ) -> LogStream: + """Deprecated: use :meth:`open` with a :class:`StageTarget`.""" + _deprecated("open_stream", "open(StageTarget(...))") + return self.open(StageTarget(stage=stage, step=step), timeout=timeout) + + def open_server_stream( + self, step: int, server: str, timeout: float | None = None + ) -> LogStream: + """Deprecated: use :meth:`open` with a :class:`ServerTarget`.""" + _deprecated("open_server_stream", "open(ServerTarget(...))") + return self.open(ServerTarget(step=step, server=server), timeout=timeout) + + def open_replica_stream( + self, step: int, replica: str, timeout: float | None = None + ) -> LogStream: + """Deprecated: use :meth:`open` with a :class:`ReplicaTarget`.""" + _deprecated("open_replica_stream", "open(ReplicaTarget(...))") + return self.open(ReplicaTarget(step=step, replica=replica), timeout=timeout) + + async def stream_server( + self, step: int, server: str, timeout: float | None = None + ) -> AsyncIterator[LogEntry]: + """Deprecated: use :meth:`stream` with a :class:`ServerTarget`.""" + _deprecated("stream_server", "stream(ServerTarget(...))") + async for entry in self.stream( + ServerTarget(step=step, server=server), timeout=timeout + ): + yield entry + + async def stream_replica( + self, step: int, replica: str, timeout: float | None = None + ) -> AsyncIterator[LogEntry]: + """Deprecated: use :meth:`stream` with a :class:`ReplicaTarget`.""" + _deprecated("stream_replica", "stream(ReplicaTarget(...))") + async for entry in self.stream( + ReplicaTarget(step=step, replica=replica), timeout=timeout + ): + yield entry + async def collect_server( self, step: int, @@ -258,17 +314,13 @@ async def collect_server( max_entries: int | None = None, timeout: float | None = 30.0, ) -> list[LogEntry]: - """Collect all logs for a server into a list.""" - entries: list[LogEntry] = [] - try: - async with self.open_server_stream(step, server, timeout) as stream: - async for entry in stream: - entries.append(entry) - if max_entries and len(entries) >= max_entries: - break - except TimeoutError: - pass - return entries + """Deprecated: use :meth:`collect` with a :class:`ServerTarget`.""" + _deprecated("collect_server", "collect(ServerTarget(...))") + return await self.collect( + ServerTarget(step=step, server=server), + max_entries=max_entries, + timeout=timeout, + ) async def collect_replica( self, @@ -277,14 +329,10 @@ async def collect_replica( max_entries: int | None = None, timeout: float | None = 30.0, ) -> list[LogEntry]: - """Collect all logs for a replica into a list.""" - entries: list[LogEntry] = [] - try: - async with self.open_replica_stream(step, replica, timeout) as stream: - async for entry in stream: - entries.append(entry) - if max_entries and len(entries) >= max_entries: - break - except TimeoutError: - pass - return entries + """Deprecated: use :meth:`collect` with a :class:`ReplicaTarget`.""" + _deprecated("collect_replica", "collect(ReplicaTarget(...))") + return await self.collect( + ReplicaTarget(step=step, replica=replica), + max_entries=max_entries, + timeout=timeout, + ) diff --git a/src/codesphere/resources/workspace/logs/schemas.py b/src/codesphere/resources/workspace/logs/schemas.py index 25762c9..970f14c 100644 --- a/src/codesphere/resources/workspace/logs/schemas.py +++ b/src/codesphere/resources/workspace/logs/schemas.py @@ -2,7 +2,7 @@ from enum import Enum -from pydantic import ConfigDict +from pydantic import ConfigDict, Field from ....core.base import CamelModel @@ -13,11 +13,17 @@ class LogStage(str, Enum): RUN = "run" +class LogKind(str, Enum): + INFO = "I" + ERROR = "E" + + class LogEntry(CamelModel): model_config = ConfigDict(extra="allow") timestamp: str | None = None - kind: str | None = None # "I" for info, "E" for error + # Known kinds are typed; unknown values fall back to the raw string. + kind: LogKind | str | None = Field(default=None, union_mode="left_to_right") data: str | None = None # The actual log content def get_text(self) -> str: diff --git a/tests/resources/workspace/landscape/test_landscape.py b/tests/resources/workspace/landscape/test_landscape.py index 314a247..947720d 100644 --- a/tests/resources/workspace/landscape/test_landscape.py +++ b/tests/resources/workspace/landscape/test_landscape.py @@ -660,3 +660,73 @@ def test_profile_config_camel_case_serialization(self): assert "schemaVersion" in data assert "schema_version" not in data + + +class TestProfileCommandHardening: + """save/get/delete_profile must be safe against hostile file content.""" + + @pytest.fixture + def manager_with_capture(self, mock_http_client_for_resource): + from unittest.mock import AsyncMock + + manager = WorkspaceLandscapeManager( + mock_http_client_for_resource(None), workspace_id=123 + ) + manager._run_command = AsyncMock() + return manager + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "content", + [ + "version: 1\nPROFILE_EOF\nrest: after-terminator\n", + "cmd: $(reboot)\n", + "quote: \"it's a 'test'\"\n", + "backtick: `id`\n", + "crlf: yes\r\nnext: line\r\n", + "unicode: käse 🚀\n", + ], + ids=[ + "heredoc-terminator", + "cmd-subst", + "quotes", + "backticks", + "crlf", + "unicode", + ], + ) + async def test_save_profile_roundtrips_adversarial_content( + self, manager_with_capture, content + ): + import base64 as b64 + import re + + await manager_with_capture.save_profile("hardened", content) + + command = manager_with_capture._run_command.await_args.args[0] + match = re.fullmatch( + r"printf '%s' '([A-Za-z0-9+/=]*)' \| base64 -d > 'ci\.hardened\.yml'", + command, + ) + assert match, f"unexpected command shape: {command}" + decoded = b64.b64decode(match.group(1)).decode("utf-8") + expected = content if content.endswith("\n") else content + "\n" + assert decoded == expected + + @pytest.mark.asyncio + async def test_get_profile_quotes_filename(self, manager_with_capture): + from unittest.mock import AsyncMock, MagicMock + + response = MagicMock() + response.output = "version: 1" + manager_with_capture._run_command = AsyncMock(return_value=response) + + await manager_with_capture.get_profile("prod") + command = manager_with_capture._run_command.await_args.args[0] + assert command == "cat 'ci.prod.yml'" + + @pytest.mark.asyncio + async def test_delete_profile_quotes_filename(self, manager_with_capture): + await manager_with_capture.delete_profile("prod") + command = manager_with_capture._run_command.await_args.args[0] + assert command == "rm -f 'ci.prod.yml'" diff --git a/tests/resources/workspace/logs/test_logs.py b/tests/resources/workspace/logs/test_logs.py index 4fcad9f..bfc50bd 100644 --- a/tests/resources/workspace/logs/test_logs.py +++ b/tests/resources/workspace/logs/test_logs.py @@ -10,6 +10,9 @@ LogProblem, LogStage, LogStream, + ReplicaTarget, + ServerTarget, + StageTarget, WorkspaceLogManager, ) from codesphere.resources.workspace.logs.operations import ( @@ -101,7 +104,7 @@ class TestLogStream: def test_init(self): mock_client = MagicMock() stream = LogStream(mock_client, "/test/endpoint", LogEntry, timeout=30.0) - assert stream._client == mock_client + assert stream._http_client == mock_client assert stream._endpoint == "/test/endpoint" assert stream._entry_model == LogEntry assert stream._timeout == 30.0 @@ -187,26 +190,182 @@ def test_build_endpoint_server(self, log_manager): ) assert endpoint == "/workspaces/123/logs/run/0/server/web" - def test_open_stream_returns_log_stream(self, log_manager): - stream = log_manager.open_stream(stage=LogStage.PREPARE, step=0) + def test_open_stage_target(self, log_manager): + stream = log_manager.open(StageTarget(stage=LogStage.PREPARE, step=0)) assert isinstance(stream, LogStream) assert stream._endpoint == "/workspaces/123/logs/prepare/0" assert stream._entry_model == LogEntry - def test_open_stream_with_string_stage(self, log_manager): - stream = log_manager.open_stream(stage="test", step=1) + def test_open_stage_target_with_string_stage(self, log_manager): + stream = log_manager.open(StageTarget(stage="test", step=1)) assert stream._endpoint == "/workspaces/123/logs/test/1" - def test_open_stream_with_timeout(self, log_manager): - stream = log_manager.open_stream(stage="run", step=0, timeout=60.0) + def test_open_with_timeout(self, log_manager): + stream = log_manager.open(StageTarget(stage="run", step=0), timeout=60.0) assert stream._timeout == 60.0 - def test_open_server_stream_returns_log_stream(self, log_manager): - stream = log_manager.open_server_stream(step=0, server="web") + def test_open_server_target(self, log_manager): + stream = log_manager.open(ServerTarget(step=0, server="web")) assert isinstance(stream, LogStream) assert stream._endpoint == "/workspaces/123/logs/run/0/server/web" - def test_open_replica_stream_returns_log_stream(self, log_manager): - stream = log_manager.open_replica_stream(step=0, replica="replica-1") + def test_open_replica_target(self, log_manager): + stream = log_manager.open(ReplicaTarget(step=0, replica="replica-1")) assert isinstance(stream, LogStream) assert stream._endpoint == "/workspaces/123/logs/run/0/replica/replica-1" + + +class FakeStream: + """Minimal LogStream stand-in: async CM yielding canned entries.""" + + def __init__(self, entries): + self._entries = entries + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return None + + def __aiter__(self): + return self._iterate() + + async def _iterate(self): + for entry in self._entries: + yield entry + + +class TestUnifiedStreamCollect: + @pytest.fixture + def log_manager(self): + return WorkspaceLogManager(MagicMock(), workspace_id=123) + + @pytest.mark.asyncio + async def test_collect_with_target(self, log_manager, monkeypatch): + entries = [LogEntry(data="a"), LogEntry(data="b"), LogEntry(data="c")] + opened = [] + + def fake_open(target, timeout=None): + opened.append((target, timeout)) + return FakeStream(entries) + + monkeypatch.setattr(log_manager, "open", fake_open) + result = await log_manager.collect(ServerTarget(step=1, server="web")) + + assert result == entries + assert opened == [(ServerTarget(step=1, server="web"), 30.0)] + + @pytest.mark.asyncio + async def test_collect_respects_max_entries(self, log_manager, monkeypatch): + entries = [LogEntry(data=str(i)) for i in range(10)] + monkeypatch.setattr( + log_manager, "open", lambda t, timeout=None: FakeStream(entries) + ) + + result = await log_manager.collect( + StageTarget(stage="run", step=0), max_entries=3 + ) + assert len(result) == 3 + + @pytest.mark.asyncio + async def test_stream_with_target(self, log_manager, monkeypatch): + entries = [LogEntry(data="x")] + monkeypatch.setattr( + log_manager, "open", lambda t, timeout=None: FakeStream(entries) + ) + + got = [e async for e in log_manager.stream(StageTarget(stage="run", step=1))] + assert got == entries + + +class TestDeprecatedLogAliases: + @pytest.fixture + def log_manager(self): + return WorkspaceLogManager(MagicMock(), workspace_id=123) + + def test_open_stream_warns_and_delegates(self, log_manager): + with pytest.warns(DeprecationWarning, match="open_stream"): + stream = log_manager.open_stream(stage=LogStage.PREPARE, step=0) + assert stream._endpoint == "/workspaces/123/logs/prepare/0" + + def test_open_server_stream_warns(self, log_manager): + with pytest.warns(DeprecationWarning, match="open_server_stream"): + stream = log_manager.open_server_stream(step=0, server="web") + assert stream._endpoint == "/workspaces/123/logs/run/0/server/web" + + def test_open_replica_stream_warns(self, log_manager): + with pytest.warns(DeprecationWarning, match="open_replica_stream"): + stream = log_manager.open_replica_stream(step=0, replica="r-1") + assert stream._endpoint == "/workspaces/123/logs/run/0/replica/r-1" + + @pytest.mark.asyncio + async def test_legacy_collect_stage_form_warns(self, log_manager, monkeypatch): + entries = [LogEntry(data="legacy")] + opened = [] + + def fake_open(target, timeout=None): + opened.append(target) + return FakeStream(entries) + + monkeypatch.setattr(log_manager, "open", fake_open) + with pytest.warns(DeprecationWarning, match="collect"): + result = await log_manager.collect("run", 1) + + assert result == entries + assert opened == [StageTarget(stage="run", step=1)] + + @pytest.mark.asyncio + async def test_collect_server_warns_and_delegates(self, log_manager, monkeypatch): + entries = [LogEntry(data="srv")] + monkeypatch.setattr( + log_manager, "open", lambda t, timeout=None: FakeStream(entries) + ) + + with pytest.warns(DeprecationWarning, match="collect_server"): + result = await log_manager.collect_server(step=1, server="web") + assert result == entries + + @pytest.mark.asyncio + async def test_collect_replica_warns_and_delegates(self, log_manager, monkeypatch): + entries = [LogEntry(data="rep")] + monkeypatch.setattr( + log_manager, "open", lambda t, timeout=None: FakeStream(entries) + ) + + with pytest.warns(DeprecationWarning, match="collect_replica"): + result = await log_manager.collect_replica(step=1, replica="r-1") + assert result == entries + + @pytest.mark.asyncio + async def test_stream_server_warns_and_delegates(self, log_manager, monkeypatch): + entries = [LogEntry(data="srv")] + monkeypatch.setattr( + log_manager, "open", lambda t, timeout=None: FakeStream(entries) + ) + + with pytest.warns(DeprecationWarning, match="stream_server"): + got = [e async for e in log_manager.stream_server(step=1, server="web")] + assert got == entries + + @pytest.mark.asyncio + async def test_stream_replica_warns_and_delegates(self, log_manager, monkeypatch): + entries = [LogEntry(data="rep")] + monkeypatch.setattr( + log_manager, "open", lambda t, timeout=None: FakeStream(entries) + ) + + with pytest.warns(DeprecationWarning, match="stream_replica"): + got = [e async for e in log_manager.stream_replica(step=1, replica="r-1")] + assert got == entries + + +class TestLogKind: + def test_known_kinds_are_enum(self): + from codesphere.resources.workspace.logs import LogKind + + entry = LogEntry.model_validate({"kind": "I", "data": "hi"}) + assert entry.kind is LogKind.INFO + + def test_unknown_kind_falls_back_to_str(self): + entry = LogEntry.model_validate({"kind": "X", "data": "hi"}) + assert entry.kind == "X"