Log and stream cargo command outputs during tests - #32
Conversation
![terragon-labs[bot]](https://github.githubassets.com/images/gravatars/gravatar-user-420.png){kind=avatar}
Introduced streaming of stdout and stderr from cargo subprocesses to the console while capturing their content, using background threads. This replaces previous blocking capture approach with real-time output, improving feedback during command execution. Added related tests and updated fakery to simulate streaming behavior in unit tests. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
Reviewer's GuideThis PR refactors command execution in run_publish_check to use subprocess.Popen with streaming pipes, background drain threads, and logging for real-time stdout/stderr output and robust timeout handling, updates tests and mocks to simulate and verify streaming behavior and invocation logs, and polishes the CMD-MOX stub documentation. Sequence diagram for streaming and logging cargo command output during test executionsequenceDiagram
participant TestRunner
participant CargoInvocation
participant Process
participant Logger
participant StreamThreads
TestRunner->>CargoInvocation: popen(stdout=PIPE, stderr=PIPE)
CargoInvocation->>Process: Start subprocess
TestRunner->>Logger: info("Running cargo command for ...")
TestRunner->>StreamThreads: Start drain threads for stdout/stderr
StreamThreads->>Process: Read stdout/stderr
StreamThreads->>TestRunner: Write output to console
Process-->>TestRunner: wait(timeout)
alt Timeout
TestRunner->>Process: kill()
TestRunner->>Logger: exception("Cargo command timed out")
end
StreamThreads->>TestRunner: Join threads
TestRunner->>Process: Close streams
TestRunner->>TestRunner: Collect full stdout/stderr
TestRunner->>TestRunner: Return CommandResult
Class diagram for new and updated streaming/mocking classesclassDiagram
class CommandResult {
+command: tuple[str]
+return_code: int
+stdout: str
+stderr: str
}
class FakeProcess {
+args: list[str]
+stdout: io.BufferedReader
+stderr: io.BufferedReader
+wait(timeout: int|None): int
+kill(): None
}
class FakeCargoInvocation {
+_local: FakeLocal
+_args: list[str]
+run(retcode, timeout): tuple[int, str, str]
+popen(...): FakeProcess
}
class FakeLocal {
+run_callable: RunCallable
+cwd_calls: list[Path]
+env_calls: list[dict[str, str]]
+invocations: list[list[str]]
+__getitem__(command: str): FakeCargo
}
class _drain_stream {
+stream: IO
+sink: TextIO
+buffer: list[str]
}
class _stream_process_output {
+process: subprocess.Popen
+command: Command
+timeout_secs: int
+returns: CommandResult
}
FakeCargoInvocation --> FakeProcess
FakeCargoInvocation --> FakeLocal
FakeLocal --> FakeCargoInvocation
FakeProcess --> _drain_stream
_stream_process_output --> _drain_stream
_stream_process_output --> CommandResult
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
Caution Review failedThe pull request is closed. Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. WalkthroughReplaces synchronous Cargo invocations with popen-based streaming in Changes
Sequence Diagram(s)sequenceDiagram
participant Caller
participant Executor as _execute_cargo_command_with_timeout
participant Popen as popen
participant Streamer as _stream_process_output
participant Threads as StreamThreads
participant Pipes as stdout/stderr
Caller->>Executor: run(command)
Executor->>Popen: spawn process (popen(..., stdout=PIPE, stderr=PIPE))
Popen-->>Executor: process handle
rect rgb(230,245,240)
Note over Executor,Streamer: start streaming phase
Executor->>Streamer: stream process output (process, timeout)
Streamer->>Threads: start reader threads for stdout & stderr
Threads->>Pipes: read chunks concurrently
Pipes-->>Threads: output chunks
Threads->>Caller: mirror chunks to console (write/flush)
Threads-->>Streamer: append to buffers
end
alt Timeout
Streamer->>Popen: kill process
Streamer->>Threads: join with timeout
Streamer-->>Executor: raise ProcessTimedOut
else Normal exit
Popen-->>Streamer: exit code
Streamer-->>Executor: (exit_code, stdout_buf, stderr_buf)
end
Executor-->>Caller: CommandResult
Estimated Code Review Effort🎯 4 (Complex) | ⏱️ ~45 minutes
Possibly related PRs
Poem
Pre-merge checks and finishing touches✅ Passed checks (2 passed)
📜 Recent review detailsConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Disabled knowledge base sources:
📒 Files selected for processing (1)
Comment |
![codescene-delta-analysis[bot]](https://avatars.githubusercontent.com/in/53921?s=60&v=4){kind=small}
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey there - I've reviewed your changes and found some issues that need to be addressed.
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location> `crate_tools/unittests/publish_check/test_command_handling.py:127` </location>
<code_context>
+def test_run_cargo_command_streams_output(
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding assertions to verify that streaming output is actually printed to the console in real time.
The current test only confirms output presence, not real-time streaming. Consider using a mock or custom sink for sys.stdout/sys.stderr to assert incremental writes.
</issue_to_address>
### Comment 2
<location> `crate_tools/unittests/publish_check/test_command_handling.py:331` </location>
<code_context>
+ assert fake_local.invocations == [["cargo", "oops"]]
def test_run_cargo_command_times_out(
+ monkeypatch: pytest.MonkeyPatch,
patch_local_runner: typ.Callable[[RunCallable], FakeLocal],
</code_context>
<issue_to_address>
**suggestion (testing):** Good coverage of the timeout scenario, but consider testing cleanup of resources after timeout.
Please add assertions or mocks to confirm that process streams and threads are properly cleaned up after a timeout.
Suggested implementation:
```python
def test_run_cargo_command_times_out(
monkeypatch: pytest.MonkeyPatch,
patch_local_runner: typ.Callable[[RunCallable], FakeLocal],
mocker,
```
```python
patch_local_runner(lambda _args: (0, "", ""))
# Mock threading.Thread to track thread creation and cleanup
mock_thread = mocker.patch("threading.Thread", autospec=True)
mock_process = mocker.patch("subprocess.Popen", autospec=True)
context = run_publish_check_module.build_cargo_command_context(
"demo",
fake_workspace,
```
```python
# Simulate timeout scenario
# (Assume the function under test starts a thread/process and times out)
# Call the function that triggers the timeout
# For example:
# result = run_publish_check_module.run_cargo_command_with_timeout(context, timeout=0.01)
# Assert that threads/processes are cleaned up after timeout
assert mock_thread.call_count > 0 # Thread was started
for call in mock_thread.call_args_list:
thread_instance = call[0][0] if call[0] else None
if thread_instance:
assert thread_instance.is_alive() is False # Thread should be stopped after timeout
assert mock_process.call_count > 0 # Process was started
for call in mock_process.call_args_list:
process_instance = call[0][0] if call[0] else None
if process_instance:
assert process_instance.terminate.called or process_instance.kill.called # Process should be terminated
```
- You may need to adjust the mock patch targets (`threading.Thread`, `subprocess.Popen`) to match the actual implementation in your codebase.
- If your timeout logic uses other resource types (e.g., asyncio tasks), mock and assert cleanup for those as well.
- Ensure the function under test actually triggers a timeout and attempts cleanup for the test to be meaningful.
</issue_to_address>
### Comment 3
<location> `crate_tools/unittests/publish_check/conftest.py:183` </location>
<code_context>
return FakeCargoInvocation(self._local, extras)
+class FakeProcess:
+ """Simulate a subprocess for cargo command tests."""
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider replacing the custom FakeProcess and buffer logic with a lightweight result object like SimpleNamespace or CompletedProcess for popen.
```suggestion
Instead of re‐implementing a full buffered subprocess, return a simple result object (e.g. a SimpleNamespace or subprocess.CompletedProcess) that carries stdout/stderr/returncode. You can then drop FakeProcess, _to_buffer, wait(), kill(), and all io imports:
```python
from types import SimpleNamespace
# or: from subprocess import CompletedProcess
class FakeLocal:
# ... existing __init__, __getitem__ etc. ...
def popen(self, *args: str, timeout: int | None = None) -> SimpleNamespace:
"""Stub popen by returning a lightweight object with stdout, stderr, returncode."""
cmd = list(args)
self.invocations.append(cmd)
returncode, stdout, stderr = self.run_callable(cmd, timeout)
# Using SimpleNamespace:
return SimpleNamespace(
returncode=returncode,
stdout=stdout,
stderr=stderr
)
# Or, if you prefer subprocess.CompletedProcess:
# return CompletedProcess(args=cmd, returncode=returncode,
# stdout=stdout, stderr=stderr)
```
Then in tests you can assert on `.stdout`, `.stderr`, and `.returncode` directly—no need for buffered readers, manual BytesIO, or extra methods. This preserves all existing behavior but greatly simplifies the fake.
</issue_to_address>
### Comment 4
<location> `crate_tools/run_publish_check.py:184` </location>
<code_context>
+ sink.flush()
+
+
+def _stream_process_output(
+ process: subprocess.Popen[bytes],
+ command: Command,
</code_context>
<issue_to_address>
**issue (review_instructions):** Add behavioural and unit tests for the new _stream_process_output function and related streaming logic.
The new streaming process output logic, including _stream_process_output and its helpers, must be covered by both behavioural and unit tests to ensure correctness and reliability. Please add appropriate tests in the test suite.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `**/*`
**Instructions:**
For any new feature or change to an existing feature, both behavioural *and* unit tests are required.
</details>
</issue_to_address>
### Comment 5
<location> `crate_tools/run_publish_check.py:164` </location>
<code_context>
stderr: str
+def _drain_stream(
+ stream: typ.IO[typ.Any],
+ sink: typ.TextIO,
</code_context>
<issue_to_address>
**issue (review_instructions):** Add behavioural and unit tests for the new _drain_stream function.
The _drain_stream function is a new addition and must be covered by both behavioural and unit tests. Please ensure that its functionality is properly tested.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `**/*`
**Instructions:**
For any new feature or change to an existing feature, both behavioural *and* unit tests are required.
</details>
</issue_to_address>
### Comment 6
<location> `crate_tools/run_publish_check.py:207` </location>
<code_context>
+ )
+
+
+def _start_stream_threads(
+ process: subprocess.Popen[bytes],
+) -> tuple[list[threading.Thread], list[str], list[str]]:
</code_context>
<issue_to_address>
**issue (review_instructions):** Add behavioural and unit tests for the new _start_stream_threads function.
The _start_stream_threads function is a new feature and must be covered by both behavioural and unit tests. Please add tests to verify its correct operation.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `**/*`
**Instructions:**
For any new feature or change to an existing feature, both behavioural *and* unit tests are required.
</details>
</issue_to_address>
### Comment 7
<location> `crate_tools/run_publish_check.py:235` </location>
<code_context>
+ return threads, stdout_chunks, stderr_chunks
+
+
+def _wait_for_stream_threads(threads: list[threading.Thread]) -> None:
+ """Wait for any running stream mirrors to exit."""
+ for thread in threads:
</code_context>
<issue_to_address>
**issue (review_instructions):** Add behavioural and unit tests for the new _wait_for_stream_threads function.
The _wait_for_stream_threads function is a new addition and must be covered by both behavioural and unit tests. Please ensure its behaviour is tested.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `**/*`
**Instructions:**
For any new feature or change to an existing feature, both behavioural *and* unit tests are required.
</details>
</issue_to_address>
### Comment 8
<location> `crate_tools/run_publish_check.py:241` </location>
<code_context>
+ thread.join()
+
+
+def _close_process_streams(process: subprocess.Popen[bytes]) -> None:
+ """Close stdout/stderr pipes after streaming completes."""
+ for stream in (process.stdout, process.stderr):
</code_context>
<issue_to_address>
**issue (review_instructions):** Add behavioural and unit tests for the new _close_process_streams function.
The _close_process_streams function is a new feature and must be covered by both behavioural and unit tests. Please add tests to verify its correct operation.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `**/*`
**Instructions:**
For any new feature or change to an existing feature, both behavioural *and* unit tests are required.
</details>
</issue_to_address>
### Comment 9
<location> `crate_tools/run_publish_check.py:249` </location>
<code_context>
+ stream.close()
+
+
+def _handle_process_timeout(
+ process: subprocess.Popen[bytes],
+ threads: list[threading.Thread],
</code_context>
<issue_to_address>
**issue (review_instructions):** Add behavioural and unit tests for the new _handle_process_timeout function.
The _handle_process_timeout function is a new addition and must be covered by both behavioural and unit tests. Please ensure its behaviour is tested.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `**/*`
**Instructions:**
For any new feature or change to an existing feature, both behavioural *and* unit tests are required.
</details>
</issue_to_address>
### Comment 10
<location> `crate_tools/run_publish_check.py:331` </location>
<code_context>
- return_code, stdout, stderr = cargo_invocation.run(
- retcode=None,
- timeout=context.timeout_secs,
+ process = cargo_invocation.popen(
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
</code_context>
<issue_to_address>
**issue (review_instructions):** Add behavioural and unit tests for the new process invocation logic using popen.
The switch to using popen for process invocation is a significant change and must be covered by both behavioural and unit tests. Please ensure this new code path is properly tested.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `**/*`
**Instructions:**
For any new feature or change to an existing feature, both behavioural *and* unit tests are required.
</details>
</issue_to_address>
### Comment 11
<location> `crate_tools/unittests/publish_check/conftest.py:164` </location>
<code_context>
+ self._local.invocations.append(self._args)
+ return self._local.run_callable(self._args)
+
+ def popen(self, *_args: object, **_kwargs: object) -> FakeProcess:
+ """Return a ``FakeProcess`` that mimics streaming behaviour."""
+ self._local.invocations.append(self._args)
</code_context>
<issue_to_address>
**issue (review_instructions):** Add behavioural and unit tests for the new popen method in FakeCargoInvocation and the FakeProcess class.
The new popen method and FakeProcess class are new features and must be covered by both behavioural and unit tests. Please add tests to verify their correct operation and integration.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `**/*`
**Instructions:**
For any new feature or change to an existing feature, both behavioural *and* unit tests are required.
</details>
</issue_to_address>
### Comment 12
<location> `crate_tools/unittests/publish_check/conftest.py:183` </location>
<code_context>
return FakeCargoInvocation(self._local, extras)
+class FakeProcess:
+ """Simulate a subprocess for cargo command tests."""
+
</code_context>
<issue_to_address>
**issue (review_instructions):** Add behavioural and unit tests for the new FakeProcess class.
The FakeProcess class is a new addition and must be covered by both behavioural and unit tests. Please ensure its behaviour is properly tested.
<details>
<summary>Review instructions:</summary>
**Path patterns:** `**/*`
**Instructions:**
For any new feature or change to an existing feature, both behavioural *and* unit tests are required.
</details>
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Disabled knowledge base sources:
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (4)
crate_tools/run_publish_check.py(4 hunks)crate_tools/unittests/publish_check/conftest.py(5 hunks)crate_tools/unittests/publish_check/test_command_handling.py(9 hunks)docs/lading-design.md(1 hunks)
🧰 Additional context used
📓 Path-based instructions (4)
**/*.py
📄 CodeRabbit inference engine (.rules/python-00.md)
**/*.py: Python files must use snake_case filenames (e.g., http_client.py, task_queue.py)
Classes must use PascalCase
Variables and functions must use snake_case
Module-level constants must use UPPER_SNAKE_CASE
Prefix non-exported helpers or internal APIs with a single leading underscore
Use typing everywhere and maintain full static type coverage
Use TypedDict or @DataClass for structured data; prefer @DataClass(slots=True) for internal-only usage
Avoid Any; prefer precise types (TypeVar, Protocol, Literal, Union); use typing.cast only when necessary with justification; use object for unknown opaque values
Be explicit with return types (e.g., -> None, -> str) for all public functions and methods
Favor immutability: prefer tuples to lists and MappingProxyType for read-only mappings; document third-party frozendict if used
Use# pyright: ignoresparingly and include an explanation comment when used
Avoid side effects at import time (no global state mutation or actions on import)
Use docstrings (NumPy format) for public functions, classes, and modules
Explain tricky code with inline comments
**/*.py: Prefer context managers to encapsulate setup/teardown for resources (files, locks, connections) instead of manual try/finally blocks
Use contextlib.contextmanager (@contextmanager) to implement simple, linear setup/teardown context managers
Implement a class with enter and exit for context managers that require internal state or more complex lifecycle handling
For file I/O, prefer with open(...) as ... over open()/try/finally/close patterns
Choose @contextmanager when control flow is linear and no persistent state is needed
Choose a class-based context manager when there is internal state, lifecycle methods, re-entry, or advanced context needs
**/*.py: Exception classes must end with the suffix 'Error' (N818)
Prefer specific built-ins (e.g., TypeError, ValueError) or domain exceptions over raising Exception directly (TRY003/TRY004)
Preserve causal chains...
Files:
crate_tools/run_publish_check.pycrate_tools/unittests/publish_check/test_command_handling.pycrate_tools/unittests/publish_check/conftest.py
{README.md,docs/**}
📄 CodeRabbit inference engine (.rules/python-00.md)
Colocate documentation: keep README.md or a docs/ directory near reusable packages and include usage examples
Files:
docs/lading-design.md
docs/**/*.md
📄 CodeRabbit inference engine (AGENTS.md)
docs/**/*.md: Use docs/ markdown as the knowledge base and source of truth for requirements, dependencies, and architecture
Proactively update relevant docs/ markdown when decisions, requirements, dependencies, or architecture change
Files:
docs/lading-design.md
**/*.md
📄 CodeRabbit inference engine (AGENTS.md)
Markdown quality gates: .md files must pass markdownlint (make markdownlint) and Mermaid validation via nixie (make nixie) before commit
Files:
docs/lading-design.md
🧬 Code graph analysis (2)
crate_tools/run_publish_check.py (2)
crate_tools/unittests/publish_check/conftest.py (5)
wait(194-196)kill(198-200)cwd(228-231)env(233-236)popen(164-167)lading/workspace/metadata.py (1)
argv(132-134)
crate_tools/unittests/publish_check/test_command_handling.py (2)
crate_tools/unittests/publish_check/conftest.py (4)
patch_local_runner(120-130)FakeLocal(209-236)fake_workspace(90-94)run_publish_check_module(78-80)crate_tools/run_publish_check.py (3)
build_cargo_command_context(279-306)run_cargo_command(409-448)CommandResult(155-161)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Sourcery review
…cess simulation - Added 249 lines of comprehensive unit tests for streaming helpers in run_publish_check. - Enhanced _drain_stream to handle incremental UTF-8 decoding correctly. - Refactored FakeProcess to SimpleNamespace with incremental streaming behavior. - Updated tests to verify chunk-by-chunk streaming of stdout/stderr. - Ensured run_cargo_command invokes popen with correct pipe arguments for streaming. - Improved process stream closing and timeout handling in tests. This improves test coverage and robustness of streaming output handling in publish checks. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
leynos
changed the title
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
|
@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix: crate_tools/run_publish_check.py Comment on lines +165 to +193 def _drain_stream(
stream: typ.IO[typ.Any],
sink: typ.TextIO,
buffer: list[str],
) -> None:
"""Forward ``stream`` contents into ``sink`` while caching them."""
read_chunk = getattr(stream, "read1", stream.read)
decoder: codecs.IncrementalDecoder | None = None
def emit(text: str) -> None:
if not text:
return
buffer.append(text)
sink.write(text)
sink.flush()
while True:
chunk = read_chunk(4096)
if not chunk:
if decoder is not None:
emit(decoder.decode(b"", final=True))
break
if isinstance(chunk, bytes):
if decoder is None:
decoder = codecs.getincrementaldecoder("utf-8")("replace")
text = decoder.decode(chunk)
else:
text = str(chunk)
emit(text)❌ New issue: Bumpy Road Ahead |
This comment was marked as resolved.
This comment was marked as resolved.
…check Refactored the _drain_stream function to encapsulate decoding logic in helper functions `decode_chunk` and `finalize_decoder`. This improves readability and handles incremental UTF-8 decoding more clearly by managing the decoder state explicitly. Co-authored-by: terragon-labs[bot] <terragon-labs[bot]@users.noreply.github.com>
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
crate_tools/unittests/publish_check/test_stream_helpers.py (2)
10-27: Consider reusing_ChunkedStreamfrom conftest.py.The
_ChunkStreamclass here is very similar to_ChunkedStreamin conftest.py (lines 225-243), with the main difference being that the conftest version inherits fromio.RawIOBase. While test-specific doubles can be beneficial for isolation, consolidating these similar implementations would reduce duplication and maintenance overhead.If you prefer to keep them separate for test isolation, consider adding a brief comment explaining why this simpler version is preferred for these specific tests.
29-42: Consider extracting_Recorderto conftest.py.The
_Recorderclass is duplicated intest_command_handling.py(lines 118-129). Since it's a useful test utility for verifying streaming behavior across multiple test modules, consider extracting it toconftest.pyas a reusable fixture or helper class.Apply this change to reduce duplication:
In
conftest.py, add:class StreamRecorder: """Recording sink used to assert streaming behaviour.""" def __init__(self) -> None: self.writes: list[str] = [] self.flushes = 0 def write(self, text: str) -> int: self.writes.append(text) return len(text) def flush(self) -> None: self.flushes += 1Then import and use it in both test files.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Disabled knowledge base sources:
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (4)
crate_tools/run_publish_check.py(4 hunks)crate_tools/unittests/publish_check/conftest.py(5 hunks)crate_tools/unittests/publish_check/test_command_handling.py(9 hunks)crate_tools/unittests/publish_check/test_stream_helpers.py(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- crate_tools/run_publish_check.py
🧰 Additional context used
📓 Path-based instructions (1)
**/*.py
📄 CodeRabbit inference engine (.rules/python-00.md)
**/*.py: Python files must use snake_case filenames (e.g., http_client.py, task_queue.py)
Classes must use PascalCase
Variables and functions must use snake_case
Module-level constants must use UPPER_SNAKE_CASE
Prefix non-exported helpers or internal APIs with a single leading underscore
Use typing everywhere and maintain full static type coverage
Use TypedDict or @DataClass for structured data; prefer @DataClass(slots=True) for internal-only usage
Avoid Any; prefer precise types (TypeVar, Protocol, Literal, Union); use typing.cast only when necessary with justification; use object for unknown opaque values
Be explicit with return types (e.g., -> None, -> str) for all public functions and methods
Favor immutability: prefer tuples to lists and MappingProxyType for read-only mappings; document third-party frozendict if used
Use# pyright: ignoresparingly and include an explanation comment when used
Avoid side effects at import time (no global state mutation or actions on import)
Use docstrings (NumPy format) for public functions, classes, and modules
Explain tricky code with inline comments
**/*.py: Prefer context managers to encapsulate setup/teardown for resources (files, locks, connections) instead of manual try/finally blocks
Use contextlib.contextmanager (@contextmanager) to implement simple, linear setup/teardown context managers
Implement a class with enter and exit for context managers that require internal state or more complex lifecycle handling
For file I/O, prefer with open(...) as ... over open()/try/finally/close patterns
Choose @contextmanager when control flow is linear and no persistent state is needed
Choose a class-based context manager when there is internal state, lifecycle methods, re-entry, or advanced context needs
**/*.py: Exception classes must end with the suffix 'Error' (N818)
Prefer specific built-ins (e.g., TypeError, ValueError) or domain exceptions over raising Exception directly (TRY003/TRY004)
Preserve causal chains...
Files:
crate_tools/unittests/publish_check/test_stream_helpers.pycrate_tools/unittests/publish_check/conftest.pycrate_tools/unittests/publish_check/test_command_handling.py
🧬 Code graph analysis (3)
crate_tools/unittests/publish_check/test_stream_helpers.py (2)
crate_tools/unittests/publish_check/conftest.py (6)
read(231-232)read1(234-237)close(242-243)run_publish_check_module(81-83)wait(192-194)kill(196-198)crate_tools/run_publish_check.py (6)
_drain_stream(165-193)_stream_process_output(196-216)_start_stream_threads(219-244)_wait_for_stream_threads(247-250)_close_process_streams(253-258)_handle_process_timeout(261-275)
crate_tools/unittests/publish_check/conftest.py (1)
crate_tools/unittests/publish_check/test_stream_helpers.py (10)
wait(72-74)wait(105-106)wait(223-224)kill(108-109)kill(220-221)read(17-18)read1(20-23)close(25-26)close(191-192)close(209-210)
crate_tools/unittests/publish_check/test_command_handling.py (2)
crate_tools/unittests/publish_check/conftest.py (4)
fake_workspace(93-97)patch_local_runner(123-133)run_publish_check_module(81-83)FakeLocal(246-274)crate_tools/run_publish_check.py (3)
build_cargo_command_context(291-318)run_cargo_command(421-460)CommandResult(156-162)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Sourcery review
🔇 Additional comments (13)
crate_tools/unittests/publish_check/test_stream_helpers.py (1)
44-249: Excellent test coverage for streaming helpers.The test suite comprehensively covers the streaming infrastructure:
- UTF-8 handling across chunk boundaries
- Real-time streaming behavior with flush verification
- Timeout handling with proper cleanup
- Thread lifecycle management
- Stream cleanup
The tests are well-structured, use appropriate mocking, and verify both functional behavior and resource cleanup.
crate_tools/unittests/publish_check/test_command_handling.py (6)
101-152: Good streaming verification with proper flush assertions.The test correctly verifies that output streams chunk-by-chunk to the console and that flushes occur appropriately. The use of
Recorderto capture writes and flushes provides good observability into the streaming behavior.
154-178: LGTM! Good infrastructure verification.This test appropriately verifies that the subprocess is configured with pipes for streaming and line buffering (
bufsize=1), which are essential for the real-time streaming behavior.
181-202: LGTM! Good observability check.This test ensures that cargo command invocations are logged at the INFO level, which improves debugging and operational visibility.
205-252: LGTM! Proper adaptation to streaming model.The test correctly patches
_stream_process_outputto verify that the timeout value from the environment variable is properly propagated through the execution chain.
382-416: LGTM! Timeout handling is properly tested.The test correctly verifies that
ProcessTimedOutexceptions are propagated and converted toSystemExit. Note that the cleanup of threads and streams during timeout is comprehensively tested intest_stream_helpers.py::test_handle_process_timeout_cleans_threads_and_streams, which addresses the previous review comment about verifying resource cleanup after timeout.
255-620: LGTM! Consistent updates throughout.The remaining test updates consistently adapt to the new streaming-based execution model:
- Updated
RunCallablelambdas to return(return_code, stdout, stderr)tuples- Updated invocation assertions to reflect list-based command storage
- All changes maintain test intent while aligning with the new infrastructure
crate_tools/unittests/publish_check/conftest.py (6)
24-25: LGTM! Clean type definitions for streaming support.The
StreamDataandRunCallabletypes appropriately model the streaming interface, supporting flexible output formats (strings, bytes, or sequences of chunks) needed for comprehensive test coverage.
152-172: LGTM! Well-structured fake with proper popen support.The
FakeCargoInvocationupdates appropriately support the new streaming model:
- The
popenmethod correctly records kwargs for test assertions- Delegation to
_build_processprovides a clean separation of concerns- The simplified
runmethod signature aligns with the streaming-based approach
187-207: LGTM! Clean process mock implementation.The
_build_processfunction creates an appropriate test double forsubprocess.Popenwith all necessary attributes and methods. The use ofSimpleNamespaceprovides a lightweight, flexible mock object.The
del timeoutstatement in thewaitmethod (line 193) is acceptable for a test double, as it explicitly documents that the parameter is not used while still maintaining signature compatibility.
210-222: LGTM! Robust stream construction logic.The
_build_streamand_ensure_bytesfunctions handle various data formats correctly:
- Properly distinguishes between sequences of chunks and single string/bytes values
- Creates appropriate stream types for each case
- Clean encoding logic in
_ensure_bytes
225-243: LGTM! Proper stream implementation.The
_ChunkedStreamclass correctly implements theio.RawIOBaseinterface, providing a reusable test double for simulating chunked streaming behavior. The inheritance fromio.RawIOBaseand propersuper().close()call make this more robust than a simple custom implementation.
246-274: LGTM! Clean updates to support streaming tests.The
FakeLocalupdates appropriately support the new testing requirements:
- Simplified
invocationsstorage removes timeout tracking (now handled elsewhere)- New
popen_kwargsattribute enables verification of subprocess configuration- Changes maintain the test fixture's purpose while adapting to the streaming model
1 participant
Summary
Changes
Core
_drain_stream,_start_stream_threads,_stream_process_output,_wait_for_stream_threads,_close_process_streams,_handle_process_timeoutcargo_invocation.popento spawn process withstdout/stderrpipessubprocess.TimeoutExpiredintoProcessTimedOutLOGGER.infoCommandResultwith captured stdout and stderr contentTesting and Mocks
FakeProcesswithwait,kill, andstdout/stderrbuffersFakeLocal.invocationsto record commands only (no timeout tracking)RunCallableandFakeCargoInvocation.popenDocs
Why
How to test
crate_tools/unittests/publish_check)CommandResultcontains full captured output🌿 Generated by Terry
ℹ️ Tag @terragon-labs to ask questions and address PR feedback
📎 Task: https://www.terragonlabs.com/task/3f81e124-82f5-4760-a450-219998bdc780
Summary by Sourcery
Stream and log cargo command outputs during tests by refactoring process execution to use popen with threaded streaming, and update tests and mocks accordingly
New Features:
Enhancements:
Documentation:
Tests:
Summary by CodeRabbit
New Features
Bug Fixes
Tests
Documentation