From dc8449a0d7c02ab51e2913e7dbce561bfd0b6ab7 Mon Sep 17 00:00:00 2001 From: Simone Date: Thu, 14 May 2026 23:23:55 +0200 Subject: [PATCH] test: cover easy missing branches in report/config/credentials/ssh/wait/job --- tests/python/test_commands_job.py | 68 ++++++++++++++++++ tests/python/test_commands_report.py | 104 +++++++++++++++++++++++++++ tests/python/test_commands_wait.py | 49 +++++++++++++ tests/python/test_config.py | 47 ++++++++++++ tests/python/test_credentials.py | 18 +++++ tests/python/test_ssh_transport.py | 85 ++++++++++++++++++++++ 6 files changed, 371 insertions(+) diff --git a/tests/python/test_commands_job.py b/tests/python/test_commands_job.py index 122565a..ff4f9f6 100644 --- a/tests/python/test_commands_job.py +++ b/tests/python/test_commands_job.py @@ -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 + ) diff --git a/tests/python/test_commands_report.py b/tests/python/test_commands_report.py index ab41de6..66b4ef0 100644 --- a/tests/python/test_commands_report.py +++ b/tests/python/test_commands_report.py @@ -106,3 +106,107 @@ def test_report_job_unknown_id(tmp_path: Path) -> None: ) assert result.exit_code == 1 assert "No JSONL log" in result.output + + +# ── helpers ──────────────────────────────────────────────────────────────── + +import ci_orchestrator.commands.report as report_module # noqa: E402 + + +def test_read_events_oserror_returns_empty(tmp_path: Path) -> None: + p = tmp_path / "d" # directory: read_text raises OSError + p.mkdir() + assert report_module._read_events(p) == [] + + +def test_read_events_skips_blank_invalid_and_non_dict(tmp_path: Path) -> None: + p = tmp_path / "x.jsonl" + p.write_text( + "\n" # blank + "{not json\n" # invalid JSON + "[1, 2]\n" # not a dict + '{"phase": "job", "status": "start", "ts": "t"}\n' + '{"phase": "job", "status": "success", "data": {"elapsedSec": 1.5, "error": "e"}}\n', + encoding="utf-8", + ) + events = report_module._read_events(p) + assert len(events) == 2 + assert events[1].elapsed_sec == 1 + assert events[1].error == "e" + + +def test_format_hms_negative_returns_question_mark() -> None: + assert report_module._format_hms(-1) == "?" + + +def test_format_hms_with_hours() -> None: + assert report_module._format_hms(3725) == "1h02m05s" + assert report_module._format_hms(45) == "0m45s" + + +def test_summarise_in_progress_when_no_terminal_event() -> None: + events = [ + report_module._Event( + phase="job", status="start", ts="t", elapsed_sec=None, error=None, + raw={"jobId": "jx"}, + ) + ] + row = report_module._summarise(events, "default") + assert row.status == "in-progress" + assert row.job_id == "jx" + + +def test_summarise_uses_default_id_when_jobid_missing() -> None: + events = [ + report_module._Event( + phase="job", status="start", ts="t", elapsed_sec=None, error=None, + raw={}, + ) + ] + row = report_module._summarise(events, "fallback") + assert row.job_id == "fallback" + + +def test_summarise_truncates_long_error() -> None: + long_err = "x" * 100 + events = [ + report_module._Event( + phase="job", status="failure", ts="t", elapsed_sec=10, error=long_err, + raw={"jobId": "j"}, + ) + ] + row = report_module._summarise(events, "d") + assert row.status == "FAILED" + assert row.error.endswith("...") + assert len(row.error) == 60 + + +def test_report_job_detail_json_emits_raw_events(tmp_path: Path) -> None: + _write_jsonl(tmp_path / "r1" / "invoke-ci.jsonl", _success_events("r1")) + result = CliRunner().invoke( + cli, ["report", "job", "--log-dir", str(tmp_path), "--job-id", "r1", "--json"] + ) + assert result.exit_code == 0 + payload = json.loads(result.output.strip()) + assert isinstance(payload, list) + assert payload[0]["phase"] == "job" + + +def test_report_job_detail_empty_jsonl(tmp_path: Path) -> None: + p = tmp_path / "empty" / "invoke-ci.jsonl" + p.parent.mkdir(parents=True) + p.write_text("") + result = CliRunner().invoke( + cli, ["report", "job", "--log-dir", str(tmp_path), "--job-id", "empty"] + ) + assert result.exit_code == 1 + assert "empty or unreadable" in result.output + + +def test_report_job_failed_filter_no_results(tmp_path: Path) -> None: + _write_jsonl(tmp_path / "r1" / "invoke-ci.jsonl", _success_events("r1")) + result = CliRunner().invoke( + cli, ["report", "job", "--log-dir", str(tmp_path), "--failed"] + ) + assert result.exit_code == 0 + assert "No matching jobs" in result.output diff --git a/tests/python/test_commands_wait.py b/tests/python/test_commands_wait.py index 3bf3420..705c271 100644 --- a/tests/python/test_commands_wait.py +++ b/tests/python/test_commands_wait.py @@ -187,3 +187,52 @@ def test_wait_ready_invalid_transport_value(monkeypatch: pytest.MonkeyPatch) -> ) assert result.exit_code != 0 assert "Unknown" in result.output or "bogus" in result.output + + +# ── helpers (wait_module._wait_running / _wait_for_ip / branches) ─────────────────── + + +from ci_orchestrator.backends.errors import BackendError # noqa: E402 + + +class _ErrorBackend: + def __init__(self) -> None: + self.calls = 0 + + def is_running(self, _h: VmHandle) -> bool: + self.calls += 1 + raise BackendError("probe boom") + + def get_ip(self, _h: VmHandle, timeout: float = 0.0) -> str | None: + raise BackendError("ip boom") + + +def test_wait_running_swallows_backend_error_and_returns_false( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(wait_module.time, "sleep", lambda _s: None) + monkeypatch.setattr(wait_module.time, "monotonic", lambda: 0.0) + backend = _ErrorBackend() + # deadline already passed after first iteration + seq = iter([0.0, 1.0]) + monkeypatch.setattr(wait_module.time, "monotonic", lambda: next(seq)) + ok = wait_module._wait_running(backend, VmHandle("x"), deadline=0.5, poll=0.001) # type: ignore[arg-type] + assert ok is False + assert backend.calls == 1 + + +def test_wait_for_ip_returns_none_on_backend_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(wait_module.time, "sleep", lambda _s: None) + seq = iter([0.0, 1.0]) + monkeypatch.setattr(wait_module.time, "monotonic", lambda: next(seq)) + backend = _ErrorBackend() + ip = wait_module._wait_for_ip(backend, VmHandle("x"), deadline=0.5, poll=0.001) # type: ignore[arg-type] + assert ip is None + + +def test_normalise_guest_os_rejects_unknown() -> None: + import click + with pytest.raises(click.BadParameter): + wait_module._normalise_guest_os("freebsd") diff --git a/tests/python/test_config.py b/tests/python/test_config.py index 5eaf57e..a37bc4e 100644 --- a/tests/python/test_config.py +++ b/tests/python/test_config.py @@ -5,6 +5,8 @@ from __future__ import annotations import os from pathlib import Path +import pytest + from ci_orchestrator.config import load_config @@ -60,3 +62,48 @@ def test_env_wins_over_toml(tmp_path: Path) -> None: env={"CI_ROOT": str(tmp_path / "fromenv")}, ) assert cfg.paths.root == tmp_path / "fromenv" + + +def test_ci_config_env_loads_toml(tmp_path: Path) -> None: + toml = tmp_path / "alt.toml" + toml.write_text( + f'[paths]\nroot = "{(tmp_path / "rootenv").as_posix()}"\n', encoding="utf-8" + ) + cfg = load_config(env={"CI_CONFIG": str(toml)}) + assert cfg.paths.root == tmp_path / "rootenv" + + +def test_default_config_toml_in_root_is_loaded( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """When no config_path/env is given, ``/config.toml`` is autoloaded.""" + import ci_orchestrator.config as cfg_mod + from ci_orchestrator.config import Paths + + fake = Paths( + root=tmp_path, + templates=tmp_path / "t", + build_vms=tmp_path / "b", + artifacts=tmp_path / "a", + keys=tmp_path / "k", + ) + monkeypatch.setattr(cfg_mod, "_platform_defaults", lambda: fake) + (tmp_path / "config.toml").write_text( + '[backend]\ntype = "workstation"\noption = "auto"\n', encoding="utf-8" + ) + cfg = load_config(env={}) + assert cfg.backend.options == {"option": "auto"} + + +def test_toml_provides_ssh_key_path_and_guest_cred_target(tmp_path: Path) -> None: + toml = tmp_path / "config.toml" + key = tmp_path / "id_ed25519" + key.write_text("k") + toml.write_text( + f'ssh_key_path = "{key.as_posix()}"\n' + 'guest_cred_target = "MyTarget"\n', + encoding="utf-8", + ) + cfg = load_config(config_path=toml, env={}) + assert cfg.ssh_key_path == key + assert cfg.guest_cred_target == "MyTarget" diff --git a/tests/python/test_credentials.py b/tests/python/test_credentials.py index a48271c..39072ed 100644 --- a/tests/python/test_credentials.py +++ b/tests/python/test_credentials.py @@ -52,6 +52,24 @@ def test_get_falls_back_to_password(monkeypatch: pytest.MonkeyPatch) -> None: assert cred.password == "tokentoken" +def test_get_raises_keyerror_when_unknown(monkeypatch: pytest.MonkeyPatch) -> None: + _install_fake_keyring(monkeypatch, credential_map={}, password_map={}) + with pytest.raises(KeyError): + KeyringCredentialStore().get("missing") + + +def test_credential_store_protocol_can_be_implemented() -> None: + """Cover the Protocol body (pass / ...).""" + from ci_orchestrator.credentials import Credential, CredentialStore + + class _Impl: + def get(self, target: str) -> Credential: + return Credential(username="u", password="p") + + store: CredentialStore = _Impl() + assert store.get("x").username == "u" + + def test_get_raises_when_missing(monkeypatch: pytest.MonkeyPatch) -> None: _install_fake_keyring(monkeypatch, credential_map={}, password_map={}) with pytest.raises(KeyError): diff --git a/tests/python/test_ssh_transport.py b/tests/python/test_ssh_transport.py index 609dddb..1382efc 100644 --- a/tests/python/test_ssh_transport.py +++ b/tests/python/test_ssh_transport.py @@ -118,6 +118,91 @@ def test_run_raises_on_nonzero(monkeypatch: pytest.MonkeyPatch) -> None: t.run("bad") +def test_close_is_idempotent_and_swallows(monkeypatch: pytest.MonkeyPatch) -> None: + class _ExplodingClient(_FakeSSHClient): + def close(self) -> None: # type: ignore[override] + raise RuntimeError("already closed") + + client = _ExplodingClient() + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", key_path="/tmp/key") + t._connect() # type: ignore[attr-defined] + t.close() + t.close() # idempotent on already-closed (no client) + + +def test_context_manager_connects_and_closes(monkeypatch: pytest.MonkeyPatch) -> None: + client = _FakeSSHClient() + _install_fake_paramiko(monkeypatch, client) + with SshTransport("10.0.0.2", key_path="/tmp/key") as t: + assert isinstance(t, SshTransport) + assert client.closed + + +def test_known_hosts_file_is_loaded( + monkeypatch: pytest.MonkeyPatch, tmp_path: Any +) -> None: + kh = tmp_path / "known_hosts" + kh.write_text("") + loaded: list[str] = [] + + class _RecordingClient(_FakeSSHClient): + def load_host_keys(self, p: str) -> None: # type: ignore[override] + loaded.append(p) + + client = _RecordingClient() + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", key_path="/tmp/key", known_hosts=str(kh)) + t.run("ls") + assert loaded == [str(kh)] + + +def test_is_ready_returns_true_on_success(monkeypatch: pytest.MonkeyPatch) -> None: + client = _FakeSSHClient() + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", key_path="/tmp/key") + assert t.is_ready() is True + + +def test_is_ready_returns_false_on_connect_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class _BoomClient(_FakeSSHClient): + def connect(self, **_kw: Any) -> None: # type: ignore[override] + raise RuntimeError("net down") + + client = _BoomClient() + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", key_path="/tmp/key") + assert t.is_ready() is False + + +def test_copy_wraps_sftp_error(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None: + class _BadSFTP(_FakeSFTP): + def put(self, _l: str, _r: str) -> None: # type: ignore[override] + raise RuntimeError("sftp boom") + + client = _FakeSSHClient() + client.sftp = _BadSFTP() + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", key_path="/tmp/key") + with pytest.raises(TransportConnectError): + t.copy("/x", "/y") + + +def test_fetch_wraps_sftp_error(monkeypatch: pytest.MonkeyPatch) -> None: + class _BadSFTP(_FakeSFTP): + def get(self, _r: str, _l: str) -> None: # type: ignore[override] + raise RuntimeError("sftp boom") + + client = _FakeSSHClient() + client.sftp = _BadSFTP() + _install_fake_paramiko(monkeypatch, client) + t = SshTransport("10.0.0.2", key_path="/tmp/key") + with pytest.raises(TransportConnectError): + t.fetch("/y", "/x") + + def test_default_policy_is_auto_add(monkeypatch: pytest.MonkeyPatch) -> None: """Mirrors permissive `StrictHostKeyChecking=no` default for ephemeral CI VMs.""" client = _FakeSSHClient()