import importlib.util import io import json import os import shutil import stat import subprocess import tempfile import unittest from tests.fakes import ROOT spec = importlib.util.spec_from_file_location("host_config_block", os.path.join(ROOT, "tools", "host_config_block.py")) host_config_block = importlib.util.module_from_spec(spec) spec.loader.exec_module(host_config_block) TOKEN = "secret-token-123" CONFIG = '[model]\nid = "m"\n\n[categories]\ndocs = ["**/*.md", "docs/**"]\n' YML = f"""builds: ai-review: buildCommand: python -s -m werkrevisor werkdock: env: GITEA_BOT_TOKEN: {TOKEN} WERKREVISOR_CONFIG_TOML: | [model] id = "m" [categories] docs = ["**/*.md", "docs/**"] statusContext: ai-review/gate """ class HostConfigBlockTest(unittest.TestCase): def setUp(self): self.home = tempfile.mkdtemp(prefix="werkrevisor-test-") self.dir = os.path.join(self.home, "repo", ".git", "werkator") os.makedirs(self.dir) self.yml = os.path.join(self.dir, ".werkator.yml") with open(self.yml, "w") as f: f.write(YML) os.chmod(self.yml, 0o600) def run_tool(self, mode, config, now="20261013120000"): out = io.StringIO() from contextlib import redirect_stdout with redirect_stdout(out): code = host_config_block.main([mode], io.StringIO(json.dumps({"repo": config})), self.home, now) return code, out.getvalue() def read(self): with open(self.yml) as f: return f.read() def test_an_equal_block(self): self.assertEqual(self.run_tool("check", CONFIG), (0, "repo: equal\n")) def test_check_prints_only_line_numbers_and_key_names(self): changed = CONFIG.replace('id = "m"', 'id = "n"').replace('"docs/**"', '"doc/**"') code, out = self.run_tool("check", changed) self.assertEqual((code, out), (1, "repo: differs: lines 2, 5; changed: categories.docs, model.id\n")) self.assertNotIn(TOKEN, out) self.assertEqual(self.read(), YML) def test_apply_replaces_the_block_and_keeps_every_other_line(self): changed = CONFIG.replace('"docs/**"', '"doc/**"') + "\n[size]\nmax_files = 30\n" code, out = self.run_tool("apply", changed) self.assertEqual(code, 0) self.assertEqual(out, "repo: replaced: lines 5; changed: categories.docs; only in the file: size.max_files; " "backup .werkator.yml.bak-config-20261013120000\n") expected = YML.replace('"docs/**"]\n', '"doc/**"]\n\n [size]\n max_files = 30\n') self.assertEqual(self.read(), expected) self.assertEqual(stat.S_IMODE(os.stat(self.yml).st_mode), 0o600) with open(os.path.join(self.dir, ".werkator.yml.bak-config-20261013120000")) as f: self.assertEqual(f.read(), YML) self.assertEqual(self.run_tool("check", changed), (0, "repo: equal\n")) def test_a_changed_comment_changes_no_key(self): code, out = self.run_tool("check", "# new comment\n" + CONFIG) self.assertEqual((code, out), (1, "repo: differs: lines after 0; no key, comments or layout only\n")) def test_a_key_only_on_the_host(self): code, out = self.run_tool("check", CONFIG.replace('id = "m"\n', "")) self.assertEqual((code, out), (1, "repo: differs: lines 2; only on the host: model.id\n")) def test_a_block_with_a_chomping_indicator(self): for indicator in ("-", "+"): with self.subTest(indicator=indicator): with open(self.yml, "w") as f: f.write(YML.replace("WERKREVISOR_CONFIG_TOML: |", f"WERKREVISOR_CONFIG_TOML: |{indicator}")) self.assertEqual(self.run_tool("check", CONFIG), (0, "repo: equal\n")) def test_a_machine_config_without_the_block_stays_unchanged(self): with open(self.yml, "w") as f: f.write("builds:\n ai-review:\n buildCommand: true\n") self.assertEqual(self.run_tool("apply", CONFIG), (2, "repo: no block: 0 WERKREVISOR_CONFIG_TOML blocks\n")) self.assertEqual(self.read(), "builds:\n ai-review:\n buildCommand: true\n") self.assertEqual(os.listdir(self.dir), [".werkator.yml"]) def test_a_missing_machine_config(self): os.remove(self.yml) self.assertEqual(self.run_tool("check", CONFIG), (2, "repo: no block: no machine config\n")) def test_only_the_two_newest_backups_stay(self): for now in ("20261011000000", "20261012000000"): open(os.path.join(self.dir, host_config_block.BACKUP + now), "w").close() open(os.path.join(self.dir, ".werkator.yml.bak-before-0.40.0"), "w").close() self.run_tool("apply", CONFIG.replace('"m"', '"n"')) # A backup of deploy-rootfs is not one of this tool. self.assertEqual(sorted(os.listdir(self.dir)), [".werkator.yml", ".werkator.yml.bak-before-0.40.0", ".werkator.yml.bak-config-20261012000000", ".werkator.yml.bak-config-20261013120000"]) def test_an_unknown_mode_changes_nothing(self): from contextlib import redirect_stderr with redirect_stderr(io.StringIO()): code = host_config_block.main(["sync"], io.StringIO("{}"), self.home) self.assertEqual(code, 2) # Runs the command that ssh would run on the host here, with HOME as the home of the host. FAKE_SSH = """#!/usr/bin/env bash echo "$@" >> "$SSH_LOG" exec bash -c "${@: -1}" """ class SyncHostConfigScriptTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="werkrevisor-test-") self.home = os.path.join(self.tmp, "home") os.makedirs(os.path.join(self.home, "repo", ".git", "werkator")) self.yml = os.path.join(self.home, "repo", ".git", "werkator", ".werkator.yml") with open(self.yml, "w") as f: f.write(YML) self.work = os.path.join(self.tmp, "work") os.makedirs(os.path.join(self.work, "config")) with open(os.path.join(self.work, "config", "repo.toml"), "w") as f: f.write(CONFIG.replace('"docs/**"', '"doc/**"')) bin_dir = os.path.join(self.tmp, "bin") os.makedirs(bin_dir) with open(os.path.join(bin_dir, "ssh"), "w") as f: f.write(FAKE_SSH) os.chmod(os.path.join(bin_dir, "ssh"), 0o755) self.env = {**os.environ, "PATH": f"{bin_dir}:{os.environ['PATH']}", "HOME": self.home, "SSH_LOG": os.path.join(self.tmp, "ssh.log")} def tearDown(self): shutil.rmtree(self.tmp) def run_script(self, *args): return subprocess.run([os.path.join(ROOT, "tools", "sync-host-config"), *args], cwd=self.work, env=self.env, capture_output=True, text=True) def test_check_and_apply_over_one_connection_each(self): check = self.run_script("user@host", "repo") self.assertEqual((check.returncode, check.stdout), (1, "repo: differs: lines 5; changed: categories.docs\n")) applied = self.run_script("--apply", "user@host", "repo") self.assertEqual(applied.returncode, 0, applied.stderr) self.assertEqual(self.run_script("user@host", "repo").stdout, "repo: equal\n") with open(self.env["SSH_LOG"]) as f: calls = f.read().splitlines() self.assertEqual(len(calls), 3) self.assertTrue(all(call.startswith("-o BatchMode=yes user@host python3 -c ") for call in calls)) def test_a_repository_without_a_config_file_opens_no_connection(self): result = self.run_script("user@host", "repo", "other") self.assertEqual((result.returncode, result.stderr), (2, "no config/other.toml\n")) self.assertFalse(os.path.exists(self.env["SSH_LOG"])) def test_usage(self): self.assertEqual(self.run_script("--help").returncode, 0) self.assertEqual(self.run_script("user@host").returncode, 2) if __name__ == "__main__": unittest.main()