test: cover easy missing branches in report/config/credentials/ssh/wait/job
This commit is contained in:
@@ -392,3 +392,71 @@ def test_parse_extra_env_json_empty_inputs() -> None:
|
||||
def test_parse_extra_env_json_rejects_non_object() -> None:
|
||||
with pytest.raises(click.exceptions.UsageError):
|
||||
job_module._parse_extra_env_json("[]")
|
||||
|
||||
|
||||
def test_read_guest_os_handles_oserror(tmp_path: Path) -> None:
|
||||
"""Unreadable VMX (e.g. a directory) returns the safe ``windows`` default."""
|
||||
p = tmp_path / "is_a_dir.vmx"
|
||||
p.mkdir()
|
||||
assert job_module._read_guest_os_from_vmx(p) == "windows"
|
||||
|
||||
|
||||
def test_apply_vmx_overrides_noop_when_both_zero(tmp_path: Path) -> None:
|
||||
vmx = tmp_path / "x.vmx"
|
||||
vmx.write_text('numvcpus = "1"\n', encoding="utf-8")
|
||||
job_module._apply_vmx_overrides(vmx, cpu=0, memory_mb=0)
|
||||
assert vmx.read_text(encoding="utf-8") == 'numvcpus = "1"\n'
|
||||
|
||||
|
||||
def test_parse_extra_env_json_invalid_json_raises() -> None:
|
||||
with pytest.raises(click.exceptions.UsageError):
|
||||
job_module._parse_extra_env_json("{not json")
|
||||
|
||||
|
||||
def test_parse_extra_env_json_coerces_value_types() -> None:
|
||||
out = job_module._parse_extra_env_json('{"A":null,"B":42,"C":"x"}')
|
||||
assert out == {"A": "", "B": "42", "C": "x"}
|
||||
|
||||
|
||||
def test_parse_extra_env_json_rejects_non_string_key() -> None:
|
||||
# JSON keys are always strings, so coerce via raw call to guarantee branch.
|
||||
with pytest.raises(click.exceptions.UsageError):
|
||||
job_module._parse_extra_env_json('{"1bad":"x"}')
|
||||
|
||||
|
||||
def test_job_wait_running_swallows_backend_error(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
from ci_orchestrator.backends.errors import BackendError
|
||||
from ci_orchestrator.backends.protocol import VmHandle
|
||||
|
||||
class _Boom:
|
||||
def is_running(self, _h: VmHandle) -> bool:
|
||||
raise BackendError("nope")
|
||||
|
||||
monkeypatch.setattr(job_module.time, "sleep", lambda _s: None)
|
||||
seq = iter([0.0, 1.0])
|
||||
monkeypatch.setattr(job_module, "_now", lambda: next(seq))
|
||||
assert (
|
||||
job_module._wait_running(_Boom(), VmHandle("x"), deadline=0.5, poll=0.001) # type: ignore[arg-type]
|
||||
is False
|
||||
)
|
||||
|
||||
|
||||
def test_job_wait_for_ip_returns_none_on_backend_error(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
from ci_orchestrator.backends.errors import BackendError
|
||||
from ci_orchestrator.backends.protocol import VmHandle
|
||||
|
||||
class _Boom:
|
||||
def get_ip(self, _h: VmHandle, timeout: float = 0.0) -> str | None:
|
||||
raise BackendError("ip nope")
|
||||
|
||||
monkeypatch.setattr(job_module.time, "sleep", lambda _s: None)
|
||||
seq = iter([0.0, 1.0])
|
||||
monkeypatch.setattr(job_module, "_now", lambda: next(seq))
|
||||
assert (
|
||||
job_module._wait_for_ip(_Boom(), VmHandle("x"), deadline=0.5, poll=0.001) # type: ignore[arg-type]
|
||||
is None
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user