diff --git a/tests/test_libs_and_subcommands_detail.py b/tests/test_libs_and_subcommands_detail.py new file mode 100644 index 0000000..df4d514 --- /dev/null +++ b/tests/test_libs_and_subcommands_detail.py @@ -0,0 +1,221 @@ +"""Focused tests for library helpers and command edge cases.""" + +import gzip +import os +import subprocess +from unittest.mock import Mock + +import click +import pytest +from click.testing import CliRunner + +from kcidev.libs import files, git_repo, job_filters +from kcidev.libs.common import config_path, load_toml +from kcidev.subcommands import bisect, checkout, commit +from kcidev.subcommands.config import add_config, check_configuration, config +from kcidev.subcommands.mcp import mcp +from kcidev.subcommands.testretry import testretry as retry_command +from kcidev.subcommands.watch import watch + + +def test_load_toml_reads_explicit_settings(tmp_path): + settings = tmp_path / "kci-dev.toml" + settings.write_text( + '[staging]\napi = "https://api.example.org/"\n' + 'pipeline = "https://pipeline.example.org/"\n' + 'token = "secret"\ndefault_instance = "staging"\n', + encoding="utf-8", + ) + + assert load_toml(str(settings), "checkout")["staging"]["api"].startswith( + "https://api" + ) + assert config_path(str(settings)) == str(settings) + + +def test_load_toml_missing_required_config_aborts(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("HOME", str(tmp_path / "home")) + + with pytest.raises(click.Abort): + load_toml("missing.toml", "checkout") + + +def test_config_command_creates_private_example_file(tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path / "home")) + target = tmp_path / "nested" / "kci-dev.toml" + result = CliRunner().invoke( + config, + ["--file-path", str(target)], + obj={"SETTINGS": str(tmp_path / "missing.toml")}, + ) + + assert result.exit_code == 0, result.output + assert target.exists() + assert "default_instance" in target.read_text(encoding="utf-8") + assert oct(target.stat().st_mode & 0o777) == "0o600" + + +def test_add_config_rejects_directory(tmp_path): + with pytest.raises(click.Abort): + add_config(str(tmp_path)) + + +def test_check_configuration_rejects_existing_explicit_file(tmp_path): + existing = tmp_path / "kci-dev.toml" + existing.write_text("", encoding="utf-8") + + with pytest.raises(click.Abort): + check_configuration(str(existing)) + + +def test_bisect_state_round_trip(tmp_path): + path = tmp_path / "state.json" + state = dict(bisect.default_state, giturl="https://git.example/linux.git") + + assert bisect.load_state(str(path)) is None + bisect.save_state(state, str(path)) + assert bisect.load_state(str(path)) == state + + +def test_commit_find_diff_returns_latest_patch(tmp_path): + subprocess.run( + ["git", "init", "-b", "master", str(tmp_path)], + check=True, + capture_output=True, + text=True, + ) + subprocess.run( + ["git", "config", "user.email", "test@example.org"], cwd=tmp_path, check=True + ) + subprocess.run( + ["git", "config", "user.name", "Test User"], cwd=tmp_path, check=True + ) + (tmp_path / "base.txt").write_text("base\n", encoding="utf-8") + subprocess.run(["git", "add", "base.txt"], cwd=tmp_path, check=True) + subprocess.run(["git", "commit", "-m", "base"], cwd=tmp_path, check=True) + subprocess.run(["git", "checkout", "-b", "feature"], cwd=tmp_path, check=True) + (tmp_path / "feature.txt").write_text("feature\n", encoding="utf-8") + subprocess.run(["git", "add", "feature.txt"], cwd=tmp_path, check=True) + subprocess.run(["git", "commit", "-m", "feature"], cwd=tmp_path, check=True) + + diff = commit.find_diff(str(tmp_path), "feature", "master", "linux") + + assert "feature.txt" in diff + assert "feature" in diff + + +def test_checkout_retrieve_tot_commit_success(monkeypatch): + process = Mock(returncode=0) + process.communicate.return_value = (b"deadbeef\trefs/heads/main\n", b"") + popen = Mock(return_value=process) + monkeypatch.setattr(checkout.subprocess, "Popen", popen) + + assert ( + checkout.retrieve_tot_commit("https://git.example/linux.git", "main") + == "deadbeef" + ) + popen.assert_called_once() + + +def test_checkout_retrieve_tot_commit_failure(monkeypatch): + process = Mock(returncode=128) + process.communicate.return_value = (b"", b"fatal") + monkeypatch.setattr(checkout.subprocess, "Popen", Mock(return_value=process)) + + assert checkout.retrieve_tot_commit("https://git.example/linux.git", "main") is None + + +def test_files_download_logs_to_file_decompresses_and_sanitizes(tmp_path, monkeypatch): + response = Mock(content=gzip.compress(b"boot log\n")) + response.raise_for_status.return_value = None + monkeypatch.setattr(files.kcidev_session, "get", Mock(return_value=response)) + monkeypatch.chdir(tmp_path) + + url = files.download_logs_to_file("https://logs.example/log.gz", "bad/name:log.txt") + + assert url == f"file://{tmp_path / 'badnamelog.txt'}" + assert (tmp_path / "badnamelog.txt").read_bytes() == b"boot log\n" + + +def test_git_repo_repository_url_cleaner_removes_credentials_and_normalizes_scheme(): + assert ( + git_repo.repository_url_cleaner( + "ssh://user:pass@git.example.org:2222/linux.git" + ) + == "https://git.example.org:2222/linux.git" + ) + + +def test_job_filters_cover_tree_hardware_and_test_regexes(): + item = { + "tree_name": "mainline", + "environment_misc": {"platform": "qemu-arm64"}, + "environment_compatible": ["linux,dummy"], + "path": "baseline.login", + } + + assert job_filters.TreeFilter("main.*").matches(item) + assert job_filters.HardwareRegexFilter("qemu.*").matches(item) + assert job_filters.HardwareRegexFilter("linux,.*").matches(item) + assert job_filters.TestRegexFilter("baseline\\..*").matches(item) + assert not job_filters.TreeFilter("next").matches(item) + + +def test_testretry_sends_retry_and_prints_message(monkeypatch): + send = Mock(return_value={"message": "retry queued"}) + monkeypatch.setattr("kcidev.subcommands.testretry.send_jobretry", send) + result = CliRunner().invoke( + retry_command, + ["--nodeid", "node-1"], + obj={ + "CFG": {"staging": {"pipeline": "https://pipeline/", "token": "secret"}}, + "INSTANCE": "staging", + }, + ) + + assert result.exit_code == 0, result.output + assert "retry queued" in result.output + send.assert_called_once_with("https://pipeline/", "node-1", "secret") + + +def test_mcp_missing_instance_aborts(): + result = CliRunner().invoke( + mcp, + [], + obj={"CFG": {"staging": {"api": "https://api/"}}, "INSTANCE": "missing"}, + ) + + assert result.exit_code != 0 + assert ( + "Instance missing not found" in result.output + or "MCP support is not installed" in result.output + ) + + +def test_watch_forwards_filters_to_watch_jobs(monkeypatch): + watch_jobs = Mock(return_value=True) + monkeypatch.setattr( + "kcidev.subcommands.watch.maestro_get_node", + Mock(return_value={"treeid": "tree-1"}), + ) + monkeypatch.setattr("kcidev.subcommands.watch.maestro_watch_jobs", watch_jobs) + result = CliRunner().invoke( + watch, + ["--nodeid", "node-1", "--job-filter", "baseline", "--test", "login"], + obj={ + "CFG": { + "staging": { + "pipeline": "https://pipeline/", + "api": "https://api/", + "token": "secret", + } + }, + "INSTANCE": "staging", + }, + ) + + assert result.exit_code == 0, result.output + watch_jobs.assert_called_once_with( + "https://api/", "secret", "tree-1", ("baseline",), "login" + ) diff --git a/tests/test_script_coverage.py b/tests/test_script_coverage.py new file mode 100644 index 0000000..04f62d7 --- /dev/null +++ b/tests/test_script_coverage.py @@ -0,0 +1,88 @@ +"""Repository-wide smoke coverage for kcidev scripts. + +These tests intentionally walk every Python module under kcidev/subcommands and +kcidev/libs so newly added command or helper scripts cannot fall out of the +pytest chain unnoticed. +""" + +import importlib +import pkgutil + +import click +import pytest +from click.testing import CliRunner + +from kcidev.main import get_cli + +PACKAGES_UNDER_TEST = ("kcidev.subcommands", "kcidev.libs") + + +def _walk_modules(package_name): + package = importlib.import_module(package_name) + yield package.__name__ + for module_info in pkgutil.walk_packages( + package.__path__, prefix=f"{package.__name__}." + ): + yield module_info.name + + +ALL_SCRIPT_MODULES = sorted( + {module for package in PACKAGES_UNDER_TEST for module in _walk_modules(package)} +) + + +@pytest.mark.parametrize("module_name", ALL_SCRIPT_MODULES) +def test_every_kcidev_script_imports(module_name): + """Every script under kcidev/subcommands and kcidev/libs imports cleanly.""" + module = importlib.import_module(module_name) + assert module.__name__ == module_name + + +def _click_command_paths(command, prefix=()): + yield prefix, command + if isinstance(command, click.Group): + for name, child in sorted(command.commands.items()): + yield from _click_command_paths(child, prefix + (name,)) + + +CLI_COMMAND_PATHS = list(_click_command_paths(get_cli())) + + +@pytest.mark.parametrize( + "path,command", + CLI_COMMAND_PATHS, + ids=lambda value: " ".join(value) if isinstance(value, tuple) else None, +) +def test_every_registered_cli_command_has_working_help(path, command, tmp_path): + """Every registered kci-dev command and subgroup must render --help.""" + settings = tmp_path / "kci-dev.toml" + settings.write_text( + 'default_instance = "staging"\n' + '[staging]\napi = "https://api.example.org/"\n' + 'pipeline = "https://pipeline.example.org/"\n' + 'token = "secret"\n', + encoding="utf-8", + ) + runner = CliRunner() + result = runner.invoke(get_cli(), ["--settings", str(settings), *path, "--help"]) + + assert result.exit_code == 0, result.output + assert "Usage:" in result.output + + +def test_registered_cli_exposes_expected_top_level_commands(): + """Protect against accidentally dropping a kcidev subcommand from main.py.""" + cli = get_cli() + assert set(cli.commands) >= { + "bisect", + "checkout", + "commit", + "config", + "maestro", + "mcp", + "results", + "storage", + "submit", + "testretry", + "watch", + }