Add coverage gap tests for command functionalities
- Implement tests for disk alert JSON output to include a message field when below threshold. - Add tests for runner status when Gitea is online but no webhook URL is provided. - Introduce tests for job report details to ensure no error line appears when there are no failure events. - Add tests to skip empty JSONL files in job report listing.
This commit is contained in:
@@ -322,3 +322,234 @@ def test_collect_linux_no_files_raises(
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "no files were transferred" in result.output
|
||||
|
||||
|
||||
# ── NEW: coverage gap tests ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
from ci_orchestrator.transport.errors import TransportError # noqa: E402
|
||||
|
||||
|
||||
def test_collect_windows_include_logs_fetch_succeeds(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Line 187 (191): log fetch succeeds → file deposited, no error message."""
|
||||
_patch(monkeypatch)
|
||||
payload = _zip_payload(tmp_path)
|
||||
|
||||
class _LogFetchOk(_win_transport_with_zip(payload)): # type: ignore[misc]
|
||||
def __init__(self, *a: Any, **kw: Any) -> None:
|
||||
super().__init__(*a, **kw)
|
||||
# The log-listing command returns one log file path.
|
||||
self.run_overrides["C:\\CI\\build"] = _FakeResult(
|
||||
stdout="C:\\CI\\build\\build.log\r\n"
|
||||
)
|
||||
|
||||
def fetch(self, remote_path: str, local_path: str) -> None:
|
||||
self.fetched.append((remote_path, str(local_path)))
|
||||
# Write something for the log file fetch too.
|
||||
Path(local_path).write_bytes(payload if local_path.endswith(".zip") else b"log content")
|
||||
|
||||
monkeypatch.setattr(artifacts_module, "WinRmTransport", _LogFetchOk)
|
||||
out_dir = tmp_path / "out"
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"artifacts", "collect",
|
||||
"--ip-address", "10.0.0.7",
|
||||
"--guest-artifact-path", "C:\\CI\\output",
|
||||
"--host-artifact-dir", str(out_dir),
|
||||
"--include-logs",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
# fetch was called for both the zip and the log
|
||||
fetched_remotes = [r for r, _ in _FakeTransport.instances[0].fetched]
|
||||
assert "C:\\CI\\build\\build.log" in fetched_remotes
|
||||
assert "fetching log" in result.output
|
||||
|
||||
|
||||
def test_collect_windows_include_logs_fetch_fails(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Lines 192-193: log fetch raises TransportError → error message printed, continues."""
|
||||
_patch(monkeypatch)
|
||||
payload = _zip_payload(tmp_path)
|
||||
|
||||
class _LogFetchFail(_win_transport_with_zip(payload)): # type: ignore[misc]
|
||||
def __init__(self, *a: Any, **kw: Any) -> None:
|
||||
super().__init__(*a, **kw)
|
||||
self.run_overrides["C:\\CI\\build"] = _FakeResult(
|
||||
stdout="C:\\CI\\build\\fail.log\r\n"
|
||||
)
|
||||
|
||||
def fetch(self, remote_path: str, local_path: str) -> None:
|
||||
if remote_path.endswith(".log"):
|
||||
raise TransportError("log fetch boom")
|
||||
# Normal zip fetch
|
||||
self.fetched.append((remote_path, str(local_path)))
|
||||
Path(local_path).write_bytes(payload)
|
||||
|
||||
monkeypatch.setattr(artifacts_module, "WinRmTransport", _LogFetchFail)
|
||||
out_dir = tmp_path / "out"
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"artifacts", "collect",
|
||||
"--ip-address", "10.0.0.7",
|
||||
"--guest-artifact-path", "C:\\CI\\output",
|
||||
"--host-artifact-dir", str(out_dir),
|
||||
"--include-logs",
|
||||
],
|
||||
)
|
||||
# Should succeed overall (error logged but not fatal)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "log fetch failed" in result.output
|
||||
|
||||
|
||||
def test_collect_windows_include_logs_no_artifacts(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Line 199: include_logs=True, no artifacts extracted → ClickException."""
|
||||
_patch(monkeypatch)
|
||||
|
||||
# Transport that returns a real-looking zip but empty after extraction,
|
||||
# and returns log listing.
|
||||
empty_zip = tmp_path / "empty.zip"
|
||||
import zipfile as _zf
|
||||
with _zf.ZipFile(empty_zip, "w"):
|
||||
pass
|
||||
empty_zip_bytes = empty_zip.read_bytes()
|
||||
|
||||
class _NoArtifacts(_FakeTransport):
|
||||
def __init__(self, *a: Any, **kw: Any) -> None:
|
||||
super().__init__(*a, **kw)
|
||||
# no log lines returned
|
||||
self.run_overrides["C:\\CI\\build"] = _FakeResult(stdout="")
|
||||
|
||||
def fetch(self, remote_path: str, local_path: str) -> None:
|
||||
self.fetched.append((remote_path, str(local_path)))
|
||||
Path(local_path).write_bytes(empty_zip_bytes)
|
||||
|
||||
monkeypatch.setattr(artifacts_module, "WinRmTransport", _NoArtifacts)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"artifacts", "collect",
|
||||
"--ip-address", "10.0.0.7",
|
||||
"--guest-artifact-path", "C:\\CI\\output",
|
||||
"--host-artifact-dir", str(tmp_path / "out"),
|
||||
"--include-logs",
|
||||
],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "no artifacts after collect" in result.output
|
||||
|
||||
|
||||
def test_collect_linux_key_from_config(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Line 261: --guest-os linux with no --ssh-key-path but config.ssh_key_path set."""
|
||||
_patch(monkeypatch)
|
||||
|
||||
# Build a real tarball for the fake transport to return.
|
||||
payload_dir = tmp_path / "payload"
|
||||
payload_dir.mkdir()
|
||||
(payload_dir / "out.txt").write_text("hi", encoding="utf-8")
|
||||
tar_local = tmp_path / "src.tar.gz"
|
||||
with tarfile.open(tar_local, "w:gz") as tf:
|
||||
for f in payload_dir.iterdir():
|
||||
tf.add(f, arcname=f.name)
|
||||
payload_bytes = tar_local.read_bytes()
|
||||
|
||||
captured_key: list[Any] = []
|
||||
|
||||
class _LinuxCapture(_FakeTransport):
|
||||
def __init__(self, *a: Any, **kw: Any) -> None:
|
||||
super().__init__(*a, **kw)
|
||||
captured_key.append(kw.get("key_path"))
|
||||
|
||||
def fetch(self, remote_path: str, local_path: str) -> None:
|
||||
self.fetched.append((remote_path, str(local_path)))
|
||||
Path(local_path).write_bytes(payload_bytes)
|
||||
|
||||
monkeypatch.setattr(artifacts_module, "SshTransport", _LinuxCapture)
|
||||
|
||||
# Patch load_config to return a config that has ssh_key_path set.
|
||||
from ci_orchestrator.config import BackendConfig, Config, Paths
|
||||
|
||||
fake_key = tmp_path / "ci_linux"
|
||||
_test_paths = Paths(
|
||||
root=tmp_path, templates=tmp_path, build_vms=tmp_path,
|
||||
artifacts=tmp_path, keys=tmp_path,
|
||||
)
|
||||
_test_cfg = Config(paths=_test_paths, backend=BackendConfig(), ssh_key_path=fake_key)
|
||||
monkeypatch.setattr(artifacts_module, "load_config", lambda: _test_cfg)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"artifacts", "collect",
|
||||
"--ip-address", "10.0.0.5",
|
||||
"--guest-os", "linux",
|
||||
"--guest-artifact-path", "/opt/ci/output",
|
||||
"--host-artifact-dir", str(tmp_path / "out"),
|
||||
# no --ssh-key-path → should come from config
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert captured_key and captured_key[0] == str(fake_key)
|
||||
|
||||
|
||||
def test_collect_transport_error_raises_click_exception(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Line 280: TransportError during collect → ClickException."""
|
||||
_patch(monkeypatch)
|
||||
|
||||
class _Boom(_FakeTransport):
|
||||
def __enter__(self) -> _Boom:
|
||||
raise TransportError("connect failed")
|
||||
|
||||
monkeypatch.setattr(artifacts_module, "WinRmTransport", _Boom)
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"artifacts", "collect",
|
||||
"--ip-address", "10.0.0.7",
|
||||
"--guest-artifact-path", "C:\\CI\\output",
|
||||
"--host-artifact-dir", str(tmp_path / "out"),
|
||||
],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "connect failed" in result.output
|
||||
|
||||
|
||||
def test_collect_windows_log_listing_skips_blank_lines(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Line 187 (continue): listing stdout has blank lines → skipped silently."""
|
||||
_patch(monkeypatch)
|
||||
payload = _zip_payload(tmp_path)
|
||||
|
||||
class _BlankLines(_win_transport_with_zip(payload)): # type: ignore[misc]
|
||||
def __init__(self, *a: Any, **kw: Any) -> None:
|
||||
super().__init__(*a, **kw)
|
||||
# Return listing with leading/trailing blank lines only — no real log paths.
|
||||
self.run_overrides["C:\\CI\\build"] = _FakeResult(stdout="\r\n \r\n")
|
||||
|
||||
monkeypatch.setattr(artifacts_module, "WinRmTransport", _BlankLines)
|
||||
out_dir = tmp_path / "out"
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"artifacts", "collect",
|
||||
"--ip-address", "10.0.0.7",
|
||||
"--guest-artifact-path", "C:\\CI\\output",
|
||||
"--host-artifact-dir", str(out_dir),
|
||||
"--include-logs",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.output
|
||||
# No "fetching log" output — all listing lines were blank and skipped.
|
||||
assert "fetching log" not in result.output
|
||||
|
||||
Reference in New Issue
Block a user