Conversation
|
Warning Rate limit exceeded
Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 55 minutes and 45 seconds. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. 📝 WalkthroughWalkthroughIntroduces OCI image flashing support to the QEMU driver via a new Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client Code
participant QemuFlasherClient as QemuFlasherClient
participant QemuFlasher as QemuFlasher
participant FlsSubprocess as FLS Subprocess
participant Environment as Environment/Registry
Client->>QemuFlasherClient: flash(oci://image, partition="root")
alt OCI URL detected
QemuFlasherClient->>QemuFlasher: streamingcall("flash_oci", oci_url, partition)
QemuFlasher->>QemuFlasher: Validate OCI URL & credentials
QemuFlasher->>Environment: Resolve FLS binary path
QemuFlasher->>Environment: Inject FLS_REGISTRY_USERNAME/PASSWORD
QemuFlasher->>FlsSubprocess: exec("fls from-url oci://image /path/to/root")
loop Stream Output with Timeout
FlsSubprocess-->>QemuFlasher: stdout/stderr chunks
QemuFlasher-->>QemuFlasherClient: (stdout, stderr, None)
QemuFlasherClient-->>Client: stream chunks
end
FlsSubprocess-->>QemuFlasher: returncode
QemuFlasher-->>QemuFlasherClient: ("", "", returncode)
QemuFlasherClient-->>Client: final result
else Non-OCI Path
QemuFlasherClient->>QemuFlasher: fall back to parent flash()
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Container ImagesThe following container images have been built for this PR:
Images expire after 7 days. |
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/client.py (1)
13-23: Expose OCI credentials on the public client API instead of only readingos.environ.Right now private-registry auth is only configurable through
OCI_USERNAME/OCI_PASSWORD, so SDK callers cannot pass credentials per request and concurrent flashes against different registries have to mutate process-global state. Since the driver already acceptsoci_username/oci_password,QemuClient.flash_oci()should forward them explicitly and keep env lookup, if you still want it, at the CLI boundary.♻️ Possible shape
- def flash_oci(self, oci_url: str, partition: str | None = None): + def flash_oci( + self, + oci_url: str, + partition: str | None = None, + oci_username: str | None = None, + oci_password: str | None = None, + ): """Flash an OCI image to the specified partition using fls. @@ - return self.flasher.flash(oci_url, target=partition) + return self.flasher.call("flash_oci", oci_url, partition, oci_username, oci_password)Based on learnings, "The TFTP driver ... client.py should remain simple as it delegates these responsibilities to the driver."
Also applies to: 47-56
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/client.py` around lines 13 - 23, The flash method currently only reads OCI credentials from environment variables and never forwards per-call credentials; update QemuClient.flash to accept and forward oci_username and oci_password parameters (or read env as a fallback) when calling self.call("flash_oci", ...) so callers can pass credentials per request; specifically, change the signature/usage in flash to pass oci_username and oci_password into the flash_oci RPC (instead of only using os.environ) and preserve the existing env lookup only as a fallback (or move env-only behavior to the CLI layer), making sure the call to flash_oci uses the explicit oci_username/oci_password values.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In
`@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py`:
- Around line 287-304: The test test_flash_oci_no_credentials currently only
checks mock_run.env when the test process already lacks FLS_REGISTRY_*; seed the
ambient env first and assert flasher.flash_oci strips them: before calling
flasher.flash_oci, use patch.dict(os.environ, {"FLS_REGISTRY_USERNAME": "x",
"FLS_REGISTRY_PASSWORD": "y"}, clear=False) (or equivalent) to inject
credentials into the process environment, then call flasher.flash_oci and verify
mock_run.call_args[1]["env"] does not contain "FLS_REGISTRY_USERNAME" or
"FLS_REGISTRY_PASSWORD"; update the test function test_flash_oci_no_credentials
to perform this seeding and the same assertions around flasher.flash_oci and
mock_run.
In `@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver.py`:
- Around line 102-105: The environment for the flash subprocess (fls_env) should
not inherit any existing FLS_REGISTRY_USERNAME/FLS_REGISTRY_PASSWORD from
os.environ; update the code in the driver (around fls_env creation and anywhere
similar at the other block) to explicitly pop or delete "FLS_REGISTRY_USERNAME"
and "FLS_REGISTRY_PASSWORD" from the copied env before using it, and only set
those keys when oci_username and oci_password are provided (i.e., in the fls_env
setup for flash_oci and the similar block at lines ~110-116, ensure you remove
existing keys when credentials are None and only inject keys when non-None).
---
Nitpick comments:
In `@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/client.py`:
- Around line 13-23: The flash method currently only reads OCI credentials from
environment variables and never forwards per-call credentials; update
QemuClient.flash to accept and forward oci_username and oci_password parameters
(or read env as a fallback) when calling self.call("flash_oci", ...) so callers
can pass credentials per request; specifically, change the signature/usage in
flash to pass oci_username and oci_password into the flash_oci RPC (instead of
only using os.environ) and preserve the existing env lookup only as a fallback
(or move env-only behavior to the CLI layer), making sure the call to flash_oci
uses the explicit oci_username/oci_password values.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 5d1b0e8b-7fb1-453f-bb8e-ee6a50fea8bc
📒 Files selected for processing (3)
python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/client.pypython/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver.pypython/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py
| def test_flash_oci_no_credentials(): | ||
| """Without credentials, env vars should not contain registry auth.""" | ||
| driver = Qemu() | ||
| flasher = driver.children["flasher"] | ||
|
|
||
| with patch("jumpstarter_driver_qemu.driver.get_fls_binary", return_value="fls"): | ||
| with patch("jumpstarter_driver_qemu.driver.subprocess.run") as mock_run: | ||
| mock_result = MagicMock() | ||
| mock_result.stdout = "" | ||
| mock_result.stderr = "" | ||
| mock_result.returncode = 0 | ||
| mock_run.return_value = mock_result | ||
|
|
||
| flasher.flash_oci("oci://quay.io/public/image:tag") | ||
|
|
||
| env = mock_run.call_args[1]["env"] | ||
| assert "FLS_REGISTRY_USERNAME" not in env | ||
| assert "FLS_REGISTRY_PASSWORD" not in env |
There was a problem hiding this comment.
Make the "no credentials" test assert against an inherited env.
In its current form this only proves the subprocess env is clean when the test runner already has no FLS_REGISTRY_* variables set. Seed those variables with patch.dict(os.environ, ...) before the call and assert they are removed, otherwise this test won't catch the ambient-env leak.
🧪 Suggested hardening
- with patch("jumpstarter_driver_qemu.driver.get_fls_binary", return_value="fls"):
- with patch("jumpstarter_driver_qemu.driver.subprocess.run") as mock_run:
+ with patch.dict(os.environ, {"FLS_REGISTRY_USERNAME": "stale", "FLS_REGISTRY_PASSWORD": "stale"}):
+ with patch("jumpstarter_driver_qemu.driver.get_fls_binary", return_value="fls"):
+ with patch("jumpstarter_driver_qemu.driver.subprocess.run") as mock_run:
mock_result = MagicMock()
mock_result.stdout = ""
mock_result.stderr = ""
mock_result.returncode = 0
mock_run.return_value = mock_result
flasher.flash_oci("oci://quay.io/public/image:tag")
- env = mock_run.call_args[1]["env"]
- assert "FLS_REGISTRY_USERNAME" not in env
- assert "FLS_REGISTRY_PASSWORD" not in env
+ env = mock_run.call_args[1]["env"]
+ assert "FLS_REGISTRY_USERNAME" not in env
+ assert "FLS_REGISTRY_PASSWORD" not in envAs per coding guidelines, "Each driver package must include comprehensive tests in driver_test.py."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py`
around lines 287 - 304, The test test_flash_oci_no_credentials currently only
checks mock_run.env when the test process already lacks FLS_REGISTRY_*; seed the
ambient env first and assert flasher.flash_oci strips them: before calling
flasher.flash_oci, use patch.dict(os.environ, {"FLS_REGISTRY_USERNAME": "x",
"FLS_REGISTRY_PASSWORD": "y"}, clear=False) (or equivalent) to inject
credentials into the process environment, then call flasher.flash_oci and verify
mock_run.call_args[1]["env"] does not contain "FLS_REGISTRY_USERNAME" or
"FLS_REGISTRY_PASSWORD"; update the test function test_flash_oci_no_credentials
to perform this seeding and the same assertions around flasher.flash_oci and
mock_run.
| def flash(self, path, *, target=None, operator=None, compression=None): | ||
| if isinstance(path, str) and path.startswith("oci://"): | ||
| oci_username = os.environ.get("OCI_USERNAME") | ||
| oci_password = os.environ.get("OCI_PASSWORD") | ||
|
|
||
| if bool(oci_username) != bool(oci_password): | ||
| raise ValueError("OCI authentication requires both OCI_USERNAME and OCI_PASSWORD environment variables") | ||
|
|
||
| return self.call("flash_oci", path, target, oci_username, oci_password) | ||
|
|
||
| return super().flash(path, target=target, operator=operator, compression=compression) |
There was a problem hiding this comment.
if we do the bifurcation on the driver side it will be easier with a plan that @kirkbrauer was proposing to allow multi-language clients :)
| fls_binary = get_fls_binary( | ||
| fls_version=self.parent.fls_version, | ||
| fls_binary_url=self.parent.fls_custom_binary_url, | ||
| allow_custom_binaries=self.parent.fls_allow_custom_binaries, | ||
| ) |
There was a problem hiding this comment.
I would go ahead and start installing the fls binary on the jumpstarter container, I was exactly looking for using fls... :)
There was a problem hiding this comment.
| result = subprocess.run( | ||
| fls_cmd, | ||
| capture_output=True, | ||
| text=True, | ||
| check=True, | ||
| env=fls_env, | ||
| timeout=self.parent.flash_timeout, |
There was a problem hiding this comment.
I would do that later, but we need to find a standard way to stream back the output of the shell process (we can steal this from the shell driver) :)
but not now, don't want to complicate your PR
5e5c1de to
b0dbb2f
Compare
Signed-off-by: Benny Zlotnik <bzlotnik@redhat.com> Assisted-by: claude-opus-4.6
…ests Change --cov to --cov=. in pytest addopts so coverage.xml includes an absolute source root, allowing diff-cover to correctly match coverage data to git diff paths in the monorepo. Without this, diff-cover fell back to low-coverage data from dependent packages that only import the module. Add tests for the inner wait_for timeout handler and process cleanup on early generator exit in _stream_subprocess. Signed-off-by: Benny Zlotnik <bzlotnik@redhat.com> Assisted-by: claude-opus-4.6
Signed-off-by: Benny Zlotnik <bzlotnik@redhat.com>
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py (1)
310-326:⚠️ Potential issue | 🟡 MinorSeed stale
FLS_REGISTRY_*vars in the no-credentials test.Line 325 asserts
env is None, which meansflsinherits any ambientFLS_REGISTRY_USERNAME/FLS_REGISTRY_PASSWORD. Seed those vars and assert they are not propagated, otherwise stale registry credentials can affect public flashes unnoticed.🧪 Suggested hardening
- # Ensure OCI env vars are not set so driver doesn't pick them up - env_clean = {k: v for k, v in os.environ.items() if k not in ("OCI_USERNAME", "OCI_PASSWORD")} + # Ensure OCI env vars are not set and stale fls credentials are stripped. + env_clean = { + k: v + for k, v in os.environ.items() + if k not in ("OCI_USERNAME", "OCI_PASSWORD", "FLS_REGISTRY_USERNAME", "FLS_REGISTRY_PASSWORD") + } + env_clean["FLS_REGISTRY_USERNAME"] = "stale-user" + env_clean["FLS_REGISTRY_PASSWORD"] = "stale-pass" with patch.dict(os.environ, env_clean, clear=True): with patch("jumpstarter_driver_qemu.driver.get_fls_binary", return_value="fls"): with patch( "asyncio.create_subprocess_exec", new_callable=AsyncMock, return_value=mock_process ) as mock_exec: await _collect_flash_oci(flasher, "oci://quay.io/public/image:tag") env = mock_exec.call_args.kwargs["env"] - assert env is None + assert "FLS_REGISTRY_USERNAME" not in env + assert "FLS_REGISTRY_PASSWORD" not in envThis still needs the driver to pass a sanitized env instead of
None. As per coding guidelines, "Each driver package must include comprehensive tests indriver_test.py."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py` around lines 310 - 326, The test test_flash_oci_no_credentials currently asserts env is None which lets ambient FLS_REGISTRY_USERNAME/FLS_REGISTRY_PASSWORD leak; modify the test to seed those FLS_REGISTRY_* variables into the parent environment (e.g., include FLS_REGISTRY_USERNAME and FLS_REGISTRY_PASSWORD in the patched os.environ), call _collect_flash_oci(flasher,...), capture mock_exec = the patched asyncio.create_subprocess_exec and inspect mock_exec.call_args.kwargs["env"], and assert that this env either is a sanitized mapping that does not contain FLS_REGISTRY_USERNAME or FLS_REGISTRY_PASSWORD (preferred) or, if the driver returns None, explicitly assert that those keys are not present in the subprocess environment propagation; reference test_flash_oci_no_credentials, _collect_flash_oci, flasher, and asyncio.create_subprocess_exec when making the changes.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In
`@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py`:
- Around line 597-610: The test test_flash_oci_convenience_method currently only
verifies the URL argument and can pass if partition is ignored; update the test
to assert the partition is forwarded to the flasher invocation by checking
mock_exec.call_args (e.g., assert mock_exec.call_args.args[3] == "bios" or
another appropriate index/value) so Qemu.flash_oci("oci://...",
partition="bios") actually results in the partition being passed through to the
subprocess call.
- Around line 410-423: The test test_flash_oci_partial_credentials_rejected is
flaky because the implementation fills missing explicit
oci_username/oci_password from environment variables; update the test to ensure
no ambient OCI_USERNAME/OCI_PASSWORD are present before calling
QemuFlasher.flash_oci (e.g., monkeypatch or temporarily unset os.environ keys)
so the partial-credential calls reliably raise ValueError, referencing the Qemu
class, its child flasher (QemuFlasher) and the flash_oci method when locating
where to change the test.
---
Duplicate comments:
In
`@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py`:
- Around line 310-326: The test test_flash_oci_no_credentials currently asserts
env is None which lets ambient FLS_REGISTRY_USERNAME/FLS_REGISTRY_PASSWORD leak;
modify the test to seed those FLS_REGISTRY_* variables into the parent
environment (e.g., include FLS_REGISTRY_USERNAME and FLS_REGISTRY_PASSWORD in
the patched os.environ), call _collect_flash_oci(flasher,...), capture mock_exec
= the patched asyncio.create_subprocess_exec and inspect
mock_exec.call_args.kwargs["env"], and assert that this env either is a
sanitized mapping that does not contain FLS_REGISTRY_USERNAME or
FLS_REGISTRY_PASSWORD (preferred) or, if the driver returns None, explicitly
assert that those keys are not present in the subprocess environment
propagation; reference test_flash_oci_no_credentials, _collect_flash_oci,
flasher, and asyncio.create_subprocess_exec when making the changes.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 82d72432-46a9-4193-9b92-a8d12931aef8
📒 Files selected for processing (5)
.github/workflows/python-tests.yamlpython/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/client.pypython/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver.pypython/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.pypython/pyproject.toml
🚧 Files skipped from review as they are similar to previous changes (2)
- python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/client.py
- python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver.py
| @pytest.mark.anyio | ||
| async def test_flash_oci_partial_credentials_rejected(): | ||
| """Providing only username or only password should be rejected.""" | ||
| driver = Qemu() | ||
| flasher = driver.children["flasher"] | ||
| assert isinstance(flasher, QemuFlasher) | ||
|
|
||
| with pytest.raises(ValueError, match="OCI authentication requires both"): | ||
| async for _ in flasher.flash_oci("oci://image:tag", oci_username="user", oci_password=None): | ||
| pass | ||
|
|
||
| with pytest.raises(ValueError, match="OCI authentication requires both"): | ||
| async for _ in flasher.flash_oci("oci://image:tag", oci_username=None, oci_password="pass"): | ||
| pass |
There was a problem hiding this comment.
Cover ambient-env credential mixing for partial explicit credentials.
Line 417 and Line 421 only pass reliably when the test runner has no complementary OCI_PASSWORD / OCI_USERNAME. The implementation fills missing explicit values from env, so oci_username="user" plus ambient OCI_PASSWORD is accepted instead of rejected.
🧪 Suggested test hardening
`@pytest.mark.anyio`
async def test_flash_oci_partial_credentials_rejected():
"""Providing only username or only password should be rejected."""
driver = Qemu()
flasher = driver.children["flasher"]
assert isinstance(flasher, QemuFlasher)
- with pytest.raises(ValueError, match="OCI authentication requires both"):
- async for _ in flasher.flash_oci("oci://image:tag", oci_username="user", oci_password=None):
- pass
-
- with pytest.raises(ValueError, match="OCI authentication requires both"):
- async for _ in flasher.flash_oci("oci://image:tag", oci_username=None, oci_password="pass"):
- pass
+ with patch.dict(os.environ, {"OCI_PASSWORD": "ambient-pass"}, clear=False):
+ with pytest.raises(ValueError, match="OCI authentication requires both"):
+ async for _ in flasher.flash_oci("oci://image:tag", oci_username="user", oci_password=None):
+ pass
+
+ with patch.dict(os.environ, {"OCI_USERNAME": "ambient-user"}, clear=False):
+ with pytest.raises(ValueError, match="OCI authentication requires both"):
+ async for _ in flasher.flash_oci("oci://image:tag", oci_username=None, oci_password="pass"):
+ passThe driver should reject mixed explicit/env credential pairs, or this test should document that mixing is intentional. As per coding guidelines, "Each driver package must include comprehensive tests in driver_test.py."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py`
around lines 410 - 423, The test test_flash_oci_partial_credentials_rejected is
flaky because the implementation fills missing explicit
oci_username/oci_password from environment variables; update the test to ensure
no ambient OCI_USERNAME/OCI_PASSWORD are present before calling
QemuFlasher.flash_oci (e.g., monkeypatch or temporarily unset os.environ keys)
so the partial-credential calls reliably raise ValueError, referencing the Qemu
class, its child flasher (QemuFlasher) and the flash_oci method when locating
where to change the test.
| def test_flash_oci_convenience_method(): | ||
| """qemu.flash_oci() should delegate to flasher.flash().""" | ||
| mock_process = _create_mock_process() | ||
|
|
||
| with serve(Qemu()) as qemu: | ||
| with patch("jumpstarter_driver_qemu.driver.get_fls_binary", return_value="fls"): | ||
| with patch( | ||
| "asyncio.create_subprocess_exec", new_callable=AsyncMock, return_value=mock_process | ||
| ) as mock_exec: | ||
| qemu.flash_oci("oci://quay.io/org/image:tag", partition="bios") | ||
|
|
||
| mock_exec.assert_called_once() | ||
| assert mock_exec.call_args.args[1] == "from-url" | ||
| assert mock_exec.call_args.args[2] == "oci://quay.io/org/image:tag" |
There was a problem hiding this comment.
Assert the convenience method preserves the requested partition.
Line 606 passes partition="bios", but the test only checks the URL. It would still pass if qemu.flash_oci() ignored the partition and flashed the default root partition.
🧪 Suggested assertion
mock_exec.assert_called_once()
assert mock_exec.call_args.args[1] == "from-url"
assert mock_exec.call_args.args[2] == "oci://quay.io/org/image:tag"
+ assert Path(mock_exec.call_args.args[3]).name == "bios"As per coding guidelines, "Each driver package must include comprehensive tests in driver_test.py."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-qemu/jumpstarter_driver_qemu/driver_test.py`
around lines 597 - 610, The test test_flash_oci_convenience_method currently
only verifies the URL argument and can pass if partition is ignored; update the
test to assert the partition is forwarded to the flasher invocation by checking
mock_exec.call_args (e.g., assert mock_exec.call_args.args[3] == "bios" or
another appropriate index/value) so Qemu.flash_oci("oci://...",
partition="bios") actually results in the partition being passed through to the
subprocess call.
No description provided.