Extract wheel-check bash orchestration into Python - #372
Open
smatula wants to merge 2 commits into
Open
Conversation
Replace ~420 lines of inline bash in the install-and-import step with a Python orchestrator (run_wheel_check.py) that implements the same two-phase wheel testing flow: individual venv per wheel, then build-group fallback with undeclared dependency resolution. Refactor wheel_helpers.py functions to return values instead of printing, preserving CLI wrappers for backward compatibility. Task YAML reduced from 511 to 114 lines. Verified with 147 unit tests and 23 integration scenarios matching production output. Assisted-by: Claude Opus 4.6 <noreply@anthropic.com>
Reviewer's GuideReplaces a large inline bash wheel-testing orchestration in the install-and-import task with a structured Python orchestrator and refactors wheel_helpers helpers into return-value APIs with thin CLI shims, adding a comprehensive test suite around the new orchestrator and updated helpers. Sequence diagram for the new wheel-check Python orchestratorsequenceDiagram
participant InstallTask
participant run_wheel_check_main as run_wheel_check.main
participant run_phase1
participant run_phase2
participant build_wheel_index_main as build_wheel_index.main
participant detect_built_wheels_main as detect_built_wheels.main
participant verify_import as verify_import.py
InstallTask->>run_wheel_check_main: run_wheel_check.py --python $PYTHON
run_wheel_check_main->>build_wheel_index_main: main(files_dir, WHEEL_INDEX_PATH)
run_wheel_check_main->>detect_built_wheels_main: main(files_dir, WHEEL_INDEX_PATH, BUILT_WHEELS_PATH)
run_wheel_check_main->>run_wheel_check_main: read_built_status(BUILT_WHEELS_PATH)
alt built_status == all_cached
run_wheel_check_main-->>InstallTask: RESULT: PASSED
else built_status != all_cached
run_wheel_check_main->>run_phase1: run_phase1(wheels, built_set,...)
alt run_phase1 had_failures == false
run_phase1-->>run_wheel_check_main: had_failures = False
run_wheel_check_main->>run_wheel_check_main: print_summary(RESULTS_DIR)
run_wheel_check_main-->>InstallTask: RESULT: PASSED
else had_failures == true
run_phase1-->>run_wheel_check_main: had_failures = True
run_wheel_check_main->>run_wheel_check_main: print_summary(RESULTS_DIR)
run_wheel_check_main->>run_phase2: run_phase2(summary_files, built_set,...)
run_phase2->>verify_import: verify_in_venv(..., wheel, result_path)
run_phase2-->>run_wheel_check_main: updated results
run_wheel_check_main->>run_wheel_check_main: print_summary(COMBINED_RESULTS_DIR)
alt fail_count > 0 or total == 0
run_wheel_check_main-->>InstallTask: RESULT: FAILED
else all groups passed
run_wheel_check_main-->>InstallTask: RESULT: PASSED (via build-group fallback)
end
end
end
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 4 security issues, 5 other issues, and left some high level feedback:
Security issues:
- Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
- Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
General comments:
- The new orchestrator relies heavily on hard-coded global paths and venv locations (e.g., /tmp paths, WHEEL_INDEX_PATH, VENV_PATH), which makes reuse and future refactoring harder; consider passing these as parameters or encapsulating them in a small config object so they can be overridden by callers and tests without monkeypatching globals.
- create_venv() currently uses subprocess.run(..., check=True) without handling CalledProcessError, which will raise and bypass your result-reporting logic; it may be safer to catch failures and emit a structured failure (or return a boolean) so the main flow can handle venv-creation errors similarly to pip_install failures.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new orchestrator relies heavily on hard-coded global paths and venv locations (e.g., /tmp paths, WHEEL_INDEX_PATH, VENV_PATH), which makes reuse and future refactoring harder; consider passing these as parameters or encapsulating them in a small config object so they can be overridden by callers and tests without monkeypatching globals.
- create_venv() currently uses subprocess.run(..., check=True) without handling CalledProcessError, which will raise and bypass your result-reporting logic; it may be safer to catch failures and emit a structured failure (or return a boolean) so the main flow can handle venv-creation errors similarly to pip_install failures.
## Individual Comments
### Comment 1
<location path="tasks/scripts/wheel-check/run_wheel_check.py" line_range="44-46" />
<code_context>
+RC_SCRIPT_ERROR = 2
+
+
+def create_venv(python, path):
+ shutil.rmtree(path, ignore_errors=True)
+ subprocess.run([python, '-m', 'venv', path], check=True,
+ stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+ return path
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Handle venv creation failures explicitly instead of letting subprocess.CalledProcessError bubble up
A failure to create the venv (missing interpreter, invalid path, venv issues) will currently raise CalledProcessError and terminate with a raw traceback. Since this script otherwise reports failures via structured messages and return codes, consider catching this exception and mapping it to a clear user-facing message and controlled exit code (or a custom exception) for main() to handle.
Suggested implementation:
```python
import os
import shutil
import subprocess
import sys
```
```python
RC_SCRIPT_ERROR = 2
def create_venv(python, path):
shutil.rmtree(path, ignore_errors=True)
try:
subprocess.run(
[python, '-m', 'venv', path],
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except FileNotFoundError as exc:
print(
f"ERROR: Failed to create virtual environment at {path!r}: "
f"Python interpreter {python!r} not found ({exc}).",
file=sys.stderr,
)
sys.exit(RC_SCRIPT_ERROR)
except subprocess.CalledProcessError as exc:
print(
f"ERROR: Failed to create virtual environment at {path!r} "
f"using interpreter {python!r} (exit code {exc.returncode}).",
file=sys.stderr,
)
sys.exit(RC_SCRIPT_ERROR)
return path
```
If the import section in this file does not exactly match the `SEARCH` block above, you should instead add `import sys` alongside the existing standard-library imports at the top of `tasks/scripts/wheel-check/run_wheel_check.py`. The rest of the change (the updated `create_venv` function) can remain as-is.
</issue_to_address>
### Comment 2
<location path="tasks/scripts/wheel-check/run_wheel_check.py" line_range="64-72" />
<code_context>
+ return True
+
+
+def pip_list_json(venv):
+ result = subprocess.run(
+ [os.path.join(venv, 'bin', 'pip'), 'list', '--format=json'],
+ capture_output=True, text=True)
+ if result.returncode != 0:
+ return []
+ try:
+ return json.loads(result.stdout)
+ except json.JSONDecodeError:
+ return []
+
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Consider surfacing a warning when pip list JSON parsing fails
Right now a JSONDecodeError is treated as "no packages installed" with no signal to the user. Since malformed `pip --format=json` output usually indicates a real problem (e.g. broken pip or encoding issues), this silent fallback can hide failures and make downstream behavior (like empty group discovery) confusing. Please log a warning to stderr when JSON parsing fails, while still returning an empty list as a fallback if you prefer that behavior.
Suggested implementation:
```python
def pip_list_json(venv):
result = subprocess.run(
[os.path.join(venv, 'bin', 'pip'), 'list', '--format=json'],
capture_output=True, text=True)
if result.returncode != 0:
return []
try:
return json.loads(result.stdout)
except json.JSONDecodeError as e:
print(
f"warning: failed to parse 'pip list --format=json' output for virtualenv {venv}: {e}",
file=sys.stderr,
)
return []
```
If `json` is not already imported at the top of `run_wheel_check.py`, add `import json`. Likewise, ensure `sys` is imported since this function now writes to `sys.stderr`.
</issue_to_address>
### Comment 3
<location path="tasks/scripts/wheel-check/test_wheel_helpers.py" line_range="264" />
<code_context>
+ def test_cwd_fallback_skips_tried(self):
</code_context>
<issue_to_address>
**suggestion (testing):** Add a regression test to cover `find_missing_wheel` when `tried` is passed as a comma-separated string (backwards-compat path).
Since `find_missing_wheel` still supports `tried` as a comma-separated string (used by the CLI), the string-handling branch isn’t currently covered by tests. Please add a test that passes a non-empty comma-separated `tried` value and verifies that previously tried wheels are correctly skipped, to protect this backward-compatible path from regressions.
</issue_to_address>
### Comment 4
<location path="tasks/scripts/wheel-check/test_wheel_helpers.py" line_range="523-503" />
<code_context>
os.unlink(f.name)
- def test_empty_csv_no_false_positive(self):
+ def test_empty_set_no_false_positive(self):
summary_path, index_path = self._setup(
[{'name': 'click', 'version': '8.1.0'}],
{'click-8.1.0': 'click-8.1.0-py3-none-any.whl'},
'',
)
- with patch('sys.stdout', new_callable=io.StringIO) as mock_out:
- group_has_built(summary_path, '',
- wheel_index_path=index_path)
- self.assertEqual(mock_out.getvalue().strip(), 'no')
+ result = group_has_built(summary_path, set(),
+ wheel_index_path=index_path)
+ self.assertFalse(result)
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test that exercises `group_has_built` when `built_wheels` is provided as a CSV string to cover the legacy interface.
`group_has_built` now handles both sets and CSV strings via an `isinstance(built_wheels, str)` branch, but the tests only cover the set path. Since the CLI still passes a CSV string, please add a test that calls `group_has_built(summary_path, 'foo.whl,bar.whl', ...)` and asserts the correct True/False result so the legacy string interface remains covered.
</issue_to_address>
### Comment 5
<location path="tasks/scripts/wheel-check/test_run_wheel_check.py" line_range="167-72" />
<code_context>
+ rc = run_wheel_check.main(['--files-dir', files_dir])
+ self.assertEqual(rc, 0)
+
+ @patch('run_wheel_check.verify_in_venv', return_value=1)
+ @patch('run_wheel_check.pip_install', return_value=True)
+ @patch('run_wheel_check.create_venv', return_value='/tmp/test-venv')
+ def test_fail_no_summary(self, mock_venv, mock_pip, mock_verify):
+ files_dir = os.path.join(self.tmpdir, 'files')
+ os.makedirs(files_dir)
+ whl = make_wheel(files_dir, 'badpkg', '1.0')
+ make_sdist(files_dir, 'badpkg', '1.0')
+
+ def write_fail(venv, sd, wheel, rf):
+ run_wheel_check.write_result(rf, {
+ 'wheel': wheel, 'status': 'FAIL', 'reason': 'import failures',
+ 'imports_tested': [{'name': 'badpkg', 'success': False, 'message': 'err'}]})
+ return 1
+ mock_verify.side_effect = write_fail
+
+ rc = run_wheel_check.main(['--files-dir', files_dir])
+ self.assertEqual(rc, 1)
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add a phase-1 test for verify failures where no result file is written, to assert `_fail_no_output` behavior.
Right now the mocked `verify_in_venv` always writes a failure result, so the `_fail_no_output` fallback is never exercised. Please add a test where `verify_in_venv` returns a non-zero code without writing the result file, and assert that the JSON at `result_path` has `status='FAIL'` and the `'verify_import failed without output'` reason to cover this error path.
Suggested implementation:
```python
def write_pass_result(venv, sd, wheel, rf):
run_wheel_check.write_result(rf, {
'wheel': wheel, 'status': 'PASS', 'reason': '', 'imports_tested': []})
return 0
mock_verify.side_effect = write_pass_result
result = run_wheel_check.run_phase1(
[whl], None, self.results_dir, 'python3.12', self.tmpdir, self.script_dir)
self.assertFalse(result)
@patch('run_wheel_check.verify_in_venv', return_value=1)
@patch('run_wheel_check.pip_install', return_value=True)
@patch('run_wheel_check.create_venv', return_value='/tmp/test-venv')
def test_phase1_fail_no_output(self, mock_venv, mock_pip, mock_verify):
files_dir = os.path.join(self.tmpdir, 'files')
os.makedirs(files_dir)
whl = make_wheel(files_dir, 'badpkg', '1.0')
os.makedirs(self.results_dir, exist_ok=True)
# Simulate verify_in_venv returning non-zero without writing any result file
def verify_no_output(venv, sd, wheel, rf):
return 1
mock_verify.side_effect = verify_no_output
result = run_wheel_check.run_phase1(
[whl], None, self.results_dir, 'python3.12', self.tmpdir, self.script_dir)
# Phase1 should report failure and write a fallback result file
self.assertTrue(result)
result_path = os.path.join(self.results_dir, f'{Path(whl).name}.json')
with open(result_path, 'r', encoding='utf-8') as f:
data = json.load(f)
self.assertEqual(data.get('status'), 'FAIL')
self.assertEqual(data.get('reason'), 'verify_import failed without output')
sys.path.insert(0, str(Path(__file__).resolve().parent))
import run_wheel_check
```
1. Ensure `json` is imported at the top of `test_run_wheel_check.py`:
- Add `import json` if it is not already present.
2. If `Path` is not already imported, make sure `from pathlib import Path` is present (it appears to be, since `Path` is already used near `sys.path.insert`).
3. If your `run_phase1` uses a different naming convention for result files, adjust the `result_path` construction accordingly to match how `run_phase1` computes `result_path`.
</issue_to_address>
### Comment 6
<location path="tasks/scripts/wheel-check/run_wheel_check.py" line_range="46-47" />
<code_context>
subprocess.run([python, '-m', 'venv', path], check=True,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 7
<location path="tasks/scripts/wheel-check/run_wheel_check.py" line_range="54" />
<code_context>
result = subprocess.run(cmd, capture_output=True, text=True)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 8
<location path="tasks/scripts/wheel-check/run_wheel_check.py" line_range="65-67" />
<code_context>
result = subprocess.run(
[os.path.join(venv, 'bin', 'pip'), 'list', '--format=json'],
capture_output=True, text=True)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>
### Comment 9
<location path="tasks/scripts/wheel-check/run_wheel_check.py" line_range="77-80" />
<code_context>
result = subprocess.run(
[os.path.join(venv, 'bin', 'python'),
os.path.join(script_dir, 'verify_import.py'), wheel],
capture_output=True, text=True)
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.
*Source: opengrep*
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
… validation, backward-compat tests - create_venv() now catches FileNotFoundError and CalledProcessError with clear error messages instead of raw tracebacks - pip_list_json() warns to stderr on JSON parse failure instead of silently returning empty list - Validate --python arg against pattern (pythonN.N) before any subprocess call - Add # nosemgrep to suppress false-positive subprocess security findings - Add tests for _fail_no_output fallback, CSV string backward compat paths in find_missing_wheel and group_has_built, and input validation Assisted-by: Claude Opus 4.6 <noreply@anthropic.com>
smatula
force-pushed
the
extract-wheel-check-orchestrator
branch
from
August 5, 2026 12:43
4330b43 to
1c0a51c
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Replace ~420 lines of inline bash in the install-and-import step with a Python orchestrator (run_wheel_check.py) that implements the same two-phase wheel testing flow: individual venv per wheel, then build-group fallback with undeclared dependency resolution.
Refactor wheel_helpers.py functions to return values instead of printing, preserving CLI wrappers for backward compatibility.
Task YAML reduced from 511 to 114 lines. Verified with 147 unit tests and 23 integration scenarios matching production output.
Assisted-by: Claude Opus 4.6 noreply@anthropic.com
Summary by Sourcery
Replace inline bash wheel testing logic with a Python-based orchestrator while preserving the two-phase test flow and reporting, and refactor wheel helper utilities into return-value functions with CLI shims for backward compatibility.
Enhancements:
Build:
Tests: