180 lines
7.9 KiB
Python
180 lines
7.9 KiB
Python
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()
|