380 lines
21 KiB
Python
380 lines
21 KiB
Python
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
|
|
from tests.fakes import ROOT
|
|
|
|
spec = importlib.util.spec_from_file_location("host_secret", os.path.join(ROOT, "tools", "host_secret.py"))
|
|
host_secret = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(host_secret)
|
|
|
|
OLD = "old-token-0123456789abcdef"
|
|
NEW = "new-token-fedcba9876543210"
|
|
OTHER = "other-key-0123456789abcdef"
|
|
|
|
|
|
def sha(value):
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
|
|
def yml(token, key_quote=""):
|
|
return f"""builds:
|
|
ai-review:
|
|
werkdock:
|
|
env:
|
|
GITEA_BOT_TOKEN: {key_quote}{token}{key_quote}
|
|
OPENROUTER_API_KEY: "{OTHER}"
|
|
WERKREVISOR_CONFIG_TOML: |
|
|
[model]
|
|
id = "m"
|
|
"""
|
|
|
|
|
|
ENV = f"GITEA_CLAUDE_CODE_TOKEN={OTHER}\nGITEA_BOT_TOKEN={OLD}\n"
|
|
|
|
|
|
class HostSecretTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.home = tempfile.mkdtemp(prefix="werkrevisor-test-")
|
|
self.ymls = {}
|
|
for repo, content in (("alpha", yml(OLD)), ("beta", yml(OLD, '"')), ("gamma", "builds: {}\n")):
|
|
directory = os.path.join(self.home, repo, ".git", "werkator")
|
|
os.makedirs(directory)
|
|
self.ymls[repo] = os.path.join(directory, ".werkator.yml")
|
|
with open(self.ymls[repo], "w") as f:
|
|
f.write(content)
|
|
os.chmod(self.ymls[repo], 0o600)
|
|
self.env_file = os.path.join(self.home, "secrets.env")
|
|
with open(self.env_file, "w") as f:
|
|
f.write(ENV)
|
|
os.chmod(self.env_file, 0o600)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.home)
|
|
|
|
def run_tool(self, *argv, payload=None, now="20261020120000"):
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with redirect_stdout(out), redirect_stderr(err):
|
|
code = host_secret.main(list(argv), io.StringIO(json.dumps(payload or {})), self.home, now)
|
|
self.assertNotIn(OLD, out.getvalue() + err.getvalue())
|
|
self.assertNotIn(NEW, out.getvalue() + err.getvalue())
|
|
return code, out.getvalue(), err.getvalue()
|
|
|
|
def read(self, path):
|
|
with open(path) as f:
|
|
return f.read()
|
|
|
|
def test_check_names_each_repository_with_the_key(self):
|
|
code, out, _ = self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)})
|
|
self.assertEqual((code, out), (0, "alpha: equal; 0 backups hold that value\n"
|
|
"beta: equal; 0 backups hold that value\n"))
|
|
code, out, _ = self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": sha(NEW)})
|
|
self.assertEqual((code, out), (1, "alpha: differs; 0 backups hold that value\n"
|
|
"beta: differs; 0 backups hold that value\n"))
|
|
|
|
def test_set_replaces_only_the_value_and_keeps_quotes_and_mode(self):
|
|
code, out, _ = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW})
|
|
self.assertEqual((code, out), (0, "alpha: replaced; backup .werkator.yml.bak-secret-GITEA_BOT_TOKEN-20261020120000\n"
|
|
"beta: replaced; backup .werkator.yml.bak-secret-GITEA_BOT_TOKEN-20261020120000\n"
|
|
f"old value: sha256 {sha(OLD)}\n"))
|
|
self.assertEqual(self.read(self.ymls["alpha"]), yml(NEW))
|
|
self.assertEqual(self.read(self.ymls["beta"]), yml(NEW, '"'))
|
|
self.assertEqual(self.read(self.ymls["gamma"]), "builds: {}\n")
|
|
self.assertEqual(self.read(self.ymls["alpha"] + ".bak-secret-GITEA_BOT_TOKEN-20261020120000"), yml(OLD))
|
|
for path in (self.ymls["alpha"], self.ymls["alpha"] + ".bak-secret-GITEA_BOT_TOKEN-20261020120000"):
|
|
self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)
|
|
self.assertEqual(self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)})[1],
|
|
"alpha: differs; 1 backups hold that value\nbeta: differs; 1 backups hold that value\n")
|
|
self.assertEqual(self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW})[1], "alpha: equal\nbeta: equal\n")
|
|
|
|
def test_set_in_an_env_file(self):
|
|
code, out, _ = self.run_tool("set", "GITEA_BOT_TOKEN", self.env_file, payload={"value": NEW}, now="1")
|
|
self.assertEqual((code, out), (0, "secrets.env: replaced; backup secrets.env.bak-secret-GITEA_BOT_TOKEN-1\n"
|
|
f"old value: sha256 {sha(OLD)}\n"))
|
|
self.assertEqual(self.read(self.env_file), ENV.replace(OLD, NEW))
|
|
self.assertEqual(stat.S_IMODE(os.stat(self.env_file).st_mode), 0o600)
|
|
|
|
def test_set_adds_a_missing_key_to_an_env_file_only(self):
|
|
code, out, _ = self.run_tool("set", "WERKREVISOR_STATS_TOKEN", self.env_file, payload={"value": NEW}, now="1")
|
|
self.assertEqual((code, out), (0, "secrets.env: added; backup secrets.env.bak-secret-WERKREVISOR_STATS_TOKEN-1\n"))
|
|
self.assertEqual(self.read(self.env_file), ENV + f"WERKREVISOR_STATS_TOKEN={NEW}\n")
|
|
code, _, err = self.run_tool("set", "WERKREVISOR_STATS_TOKEN", payload={"value": NEW})
|
|
self.assertEqual((code, err), (2, "error: no file sets WERKREVISOR_STATS_TOKEN\n"))
|
|
|
|
def test_set_refuses_a_third_value(self):
|
|
with open(self.ymls["beta"], "w") as f:
|
|
f.write(yml(OTHER))
|
|
code, _, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW})
|
|
self.assertEqual((code, err), (2, "error: the files hold 2 different old values of GITEA_BOT_TOKEN\n"))
|
|
code, _, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW, "old_sha256": sha(OLD)})
|
|
self.assertEqual((code, err), (2, "error: a file holds another value of GITEA_BOT_TOKEN than the old one\n"))
|
|
self.assertEqual(self.read(self.ymls["alpha"]), yml(OLD))
|
|
|
|
def test_set_continues_a_rotation_that_stopped_half_way(self):
|
|
with open(self.ymls["beta"], "w") as f:
|
|
f.write(yml(NEW))
|
|
code, out, _ = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW, "old_sha256": sha(OLD)}, now="1")
|
|
self.assertEqual((code, out), (0, "alpha: replaced; backup .werkator.yml.bak-secret-GITEA_BOT_TOKEN-1\n"
|
|
f"beta: equal\nold value: sha256 {sha(OLD)}\n"))
|
|
|
|
def test_set_refuses_a_value_that_would_need_quoting(self):
|
|
for value in ("", "short", "with space 0123456789", 'quote"0123456789abc', "line\nbreak0123456789"):
|
|
code, _, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": value})
|
|
self.assertEqual((code, err.startswith("error: the new value")), (2, True), value)
|
|
self.assertEqual(self.read(self.ymls["alpha"]), yml(OLD))
|
|
|
|
def test_a_file_with_two_lines_for_the_key_stops_set_before_any_change(self):
|
|
with open(self.ymls["beta"], "a") as f:
|
|
f.write(f" GITEA_BOT_TOKEN: {OLD}\n")
|
|
code, out, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW})
|
|
self.assertEqual((code, out, err), (2, "", "error: 2 lines set GITEA_BOT_TOKEN\n"))
|
|
self.assertEqual(self.read(self.ymls["alpha"]), yml(OLD))
|
|
self.assertEqual(os.listdir(os.path.dirname(self.ymls["alpha"])), [".werkator.yml"])
|
|
|
|
def test_a_line_with_a_comment_is_not_changed(self):
|
|
with open(self.ymls["alpha"], "w") as f:
|
|
f.write(yml(OLD).replace(OLD, OLD + " # rotated"))
|
|
code, _, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW})
|
|
self.assertEqual((code, "a form this tool does not change" in err), (2, True))
|
|
|
|
def test_purge_removes_the_backups_with_the_old_value_only(self):
|
|
directory = os.path.dirname(self.ymls["alpha"])
|
|
for name, content in ((".werkator.yml.bak-before-0.54.1", yml(OLD)), (".werkator.yml.pre-0.21.0", yml(OTHER + "x")),
|
|
(".werkator.yml.bak-other", "no key\n")):
|
|
with open(os.path.join(directory, name), "w") as f:
|
|
f.write(content)
|
|
code, _, err = self.run_tool("purge", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)})
|
|
self.assertEqual((code, err), (2, "error: alpha still holds that value; set the new value first\n"))
|
|
self.assertEqual(len(os.listdir(directory)), 4)
|
|
self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW}, now="2")
|
|
code, out, _ = self.run_tool("purge", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)})
|
|
self.assertEqual((code, out), (0, "alpha: removed .werkator.yml.bak-before-0.54.1, .werkator.yml.bak-secret-GITEA_BOT_TOKEN-2\n"
|
|
"beta: removed .werkator.yml.bak-secret-GITEA_BOT_TOKEN-2\n"))
|
|
self.assertEqual(sorted(os.listdir(directory)), [".werkator.yml", ".werkator.yml.bak-other", ".werkator.yml.pre-0.21.0"])
|
|
|
|
def test_purge_keeps_a_file_that_only_shares_the_prefix(self):
|
|
production = self.env_file + ".production"
|
|
with open(production, "w") as f:
|
|
f.write(ENV)
|
|
with open(self.env_file + ".bak-secret-GITEA_BOT_TOKEN-1", "w") as f:
|
|
f.write(ENV)
|
|
self.run_tool("set", "GITEA_BOT_TOKEN", self.env_file, payload={"value": NEW}, now="2")
|
|
code, out, _ = self.run_tool("purge", "GITEA_BOT_TOKEN", self.env_file, payload={"sha256": sha(OLD)})
|
|
self.assertEqual((code, out), (0, "secrets.env: removed secrets.env.bak-secret-GITEA_BOT_TOKEN-1, "
|
|
"secrets.env.bak-secret-GITEA_BOT_TOKEN-2\n"))
|
|
self.assertEqual(self.read(production), ENV)
|
|
|
|
def test_an_unreadable_machine_config_is_an_error(self):
|
|
with open(self.ymls["beta"], "wb") as f:
|
|
f.write(b"\xff\xfe broken")
|
|
for mode, payload in (("check", {"sha256": sha(OLD)}), ("purge", {"sha256": sha(NEW)}), ("set", {"value": NEW})):
|
|
code, _, err = self.run_tool(mode, "GITEA_BOT_TOKEN", payload=payload)
|
|
self.assertEqual((code, err), (2, "error: cannot read .werkator.yml: UnicodeDecodeError\n"), mode)
|
|
self.assertEqual(self.read(self.ymls["alpha"]), yml(OLD))
|
|
|
|
def test_a_backup_it_cannot_parse_may_hold_the_value(self):
|
|
directory = os.path.dirname(self.ymls["alpha"])
|
|
with open(os.path.join(directory, ".werkator.yml.bak-broken"), "w") as f:
|
|
f.write(yml(OLD).replace(OLD, OLD + " # comment"))
|
|
code, out, _ = self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)})
|
|
self.assertEqual((code, out.splitlines()[0]),
|
|
(2, "alpha: equal; 0 backups hold that value; cannot read or parse .werkator.yml.bak-broken"))
|
|
self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW}, now="2")
|
|
code, _, err = self.run_tool("purge", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)})
|
|
self.assertEqual((code, err), (2, "error: alpha: cannot read or parse .werkator.yml.bak-broken; "
|
|
"check or remove them by hand\n"))
|
|
self.assertEqual(len(os.listdir(directory)), 3)
|
|
|
|
def test_export_writes_the_one_value_as_json_only_into_a_pipe(self):
|
|
out = io.StringIO()
|
|
with redirect_stdout(out):
|
|
code = host_secret.main(["export", "GITEA_BOT_TOKEN"], io.StringIO(""), self.home)
|
|
self.assertEqual((code, json.loads(out.getvalue())), (0, {"value": OLD}))
|
|
|
|
class Terminal(io.StringIO):
|
|
def isatty(self):
|
|
return True
|
|
out, err = Terminal(), io.StringIO()
|
|
with redirect_stdout(out), redirect_stderr(err):
|
|
code = host_secret.main(["export", "GITEA_BOT_TOKEN"], io.StringIO(""), self.home)
|
|
self.assertEqual((code, out.getvalue(), err.getvalue()), (2, "", "error: export writes a value; it refuses a terminal\n"))
|
|
with open(self.ymls["beta"], "w") as f:
|
|
f.write(yml(NEW))
|
|
self.assertEqual(self.run_tool("export", "GITEA_BOT_TOKEN")[2],
|
|
"error: the files hold 2 different values of GITEA_BOT_TOKEN\n")
|
|
|
|
def test_invalid_input_prints_no_payload(self):
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with redirect_stdout(out), redirect_stderr(err):
|
|
code = host_secret.main(["set", "GITEA_BOT_TOKEN"], io.StringIO(f"{{\"value\": \"{NEW}\""), self.home)
|
|
self.assertEqual((code, err.getvalue()), (2, "error: the input is no JSON object with the needed field\n"))
|
|
self.assertEqual(self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": "abc"})[2],
|
|
"error: a SHA-256 must be 64 lower-case hex digits\n")
|
|
self.assertEqual(self.run_tool("check", "lower")[0], 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 RotateSecretScriptTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp(prefix="werkrevisor-test-")
|
|
self.home = os.path.join(self.tmp, "home")
|
|
directory = os.path.join(self.home, "repo", ".git", "werkator")
|
|
os.makedirs(directory)
|
|
self.yml = os.path.join(directory, ".werkator.yml")
|
|
with open(self.yml, "w") as f:
|
|
f.write(yml(OLD))
|
|
self.env_file = os.path.join(self.tmp, "secrets.env")
|
|
with open(self.env_file, "w") as f:
|
|
f.write(ENV)
|
|
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"), "WERKREVISOR_SECRETS_ENV": self.env_file}
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp)
|
|
|
|
def run_script(self, *args, stdin=""):
|
|
result = subprocess.run([os.path.join(ROOT, "tools", "rotate-secret"), *args], cwd=self.tmp, env=self.env,
|
|
input=stdin, capture_output=True, text=True)
|
|
self.assertNotIn(OLD, result.stdout + result.stderr)
|
|
self.assertNotIn(NEW, result.stdout + result.stderr)
|
|
return result
|
|
|
|
def ssh_calls(self):
|
|
with open(self.env["SSH_LOG"]) as f:
|
|
return f.read().splitlines()
|
|
|
|
def test_set_then_check_then_purge(self):
|
|
check = self.run_script("check", "user@host", "GITEA_BOT_TOKEN")
|
|
self.assertEqual((check.returncode, check.stdout), (0, "repo: equal; 0 backups hold that value\n"
|
|
"secrets.env: equal; 0 backups hold that value\n"))
|
|
result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=NEW + "\n")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
lines = result.stdout.splitlines()
|
|
self.assertRegex(lines[0], r"^repo: replaced; backup \.werkator\.yml\.bak-secret-GITEA_BOT_TOKEN-\d{14}$")
|
|
self.assertEqual(lines[1], f"old value: sha256 {sha(OLD)}")
|
|
self.assertRegex(lines[2], r"^secrets\.env: replaced; backup secrets\.env\.bak-secret-GITEA_BOT_TOKEN-\d{14}$")
|
|
self.assertEqual(lines[3:], ["repo: equal; 0 backups hold that value", "secrets.env: equal; 0 backups hold that value",
|
|
f"after the old value is revoked: tools/rotate-secret purge user@host GITEA_BOT_TOKEN {sha(OLD)}"])
|
|
with open(self.yml) as f:
|
|
self.assertEqual(f.read(), yml(NEW))
|
|
with open(self.env_file) as f:
|
|
self.assertEqual(f.read(), ENV.replace(OLD, NEW))
|
|
purge = self.run_script("purge", "user@host", "GITEA_BOT_TOKEN", sha(OLD))
|
|
self.assertEqual(purge.returncode, 0, purge.stderr)
|
|
self.assertEqual(sorted(os.listdir(os.path.dirname(self.yml))), [".werkator.yml"])
|
|
self.assertEqual(sorted(os.listdir(self.tmp)), ["bin", "home", "secrets.env", "ssh.log"])
|
|
calls = self.ssh_calls()
|
|
self.assertEqual(len(calls), 4)
|
|
self.assertTrue(all(call.startswith("-o BatchMode=yes user@host python3 -c ") for call in calls))
|
|
self.assertTrue(all(OLD not in call and NEW not in call for call in calls))
|
|
|
|
def test_set_takes_the_old_value_from_the_host_if_the_secrets_file_lacks_the_key(self):
|
|
with open(self.env_file, "w") as f:
|
|
f.write(ENV.replace(f"GITEA_BOT_TOKEN={OLD}\n", ""))
|
|
result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=NEW)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn(f"secrets.env: added; ", result.stdout)
|
|
self.assertTrue(result.stdout.endswith(f"purge user@host GITEA_BOT_TOKEN {sha(OLD)}\n"))
|
|
with open(self.env_file) as f:
|
|
self.assertEqual(f.read(), ENV.replace(OLD, NEW))
|
|
|
|
def test_fetch_adds_the_value_of_the_host_to_the_secrets_file(self):
|
|
without = ENV.replace(f"GITEA_BOT_TOKEN={OLD}\n", "")
|
|
with open(self.env_file, "w") as f:
|
|
f.write(without)
|
|
result = self.run_script("fetch", "user@host", "GITEA_BOT_TOKEN")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertRegex(result.stdout, r"^secrets\.env: added; backup secrets\.env\.bak-secret-GITEA_BOT_TOKEN-\d{14}\n$")
|
|
with open(self.env_file) as f:
|
|
self.assertEqual(f.read(), without + f"GITEA_BOT_TOKEN={OLD}\n")
|
|
again = self.run_script("fetch", "user@host", "GITEA_BOT_TOKEN")
|
|
self.assertEqual((again.returncode, again.stderr), (2, "secrets.env sets GITEA_BOT_TOKEN already\n"))
|
|
|
|
def test_fetch_stops_if_the_host_holds_no_single_value(self):
|
|
without = ENV.replace(f"GITEA_BOT_TOKEN={OLD}\n", "")
|
|
with open(self.env_file, "w") as f:
|
|
f.write(without)
|
|
other = os.path.join(self.home, "other", ".git", "werkator")
|
|
os.makedirs(other)
|
|
with open(os.path.join(other, ".werkator.yml"), "w") as f:
|
|
f.write(yml(NEW))
|
|
result = self.run_script("fetch", "user@host", "GITEA_BOT_TOKEN")
|
|
self.assertEqual((result.returncode, result.stderr),
|
|
(2, "error: the files hold 2 different values of GITEA_BOT_TOKEN\n"))
|
|
with open(self.env_file) as f:
|
|
self.assertEqual(f.read(), without)
|
|
|
|
def test_a_secrets_file_without_the_key_is_an_error_with_2(self):
|
|
self.assertEqual(self.run_script("probe", "WERKREVISOR_STATS_TOKEN").returncode, 2)
|
|
self.assertEqual(self.run_script("digest", "WERKREVISOR_STATS_TOKEN").returncode, 2)
|
|
|
|
def test_set_refuses_if_the_host_holds_another_value_than_the_secrets_file(self):
|
|
with open(self.yml, "w") as f:
|
|
f.write(yml(OTHER))
|
|
result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=NEW)
|
|
self.assertEqual((result.returncode, result.stderr),
|
|
(2, "error: a file holds another value of GITEA_BOT_TOKEN than the old one\n"))
|
|
with open(self.env_file) as f:
|
|
self.assertEqual(f.read(), ENV)
|
|
|
|
def test_set_stops_before_the_host_if_the_secrets_file_cannot_be_parsed(self):
|
|
with open(self.env_file, "w") as f:
|
|
f.write(ENV + f"GITEA_BOT_TOKEN={OLD}\n")
|
|
result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=NEW)
|
|
self.assertEqual((result.returncode, result.stderr), (2, "2 lines set GITEA_BOT_TOKEN\n"))
|
|
self.assertFalse(os.path.exists(self.env["SSH_LOG"]))
|
|
with open(self.yml) as f:
|
|
self.assertEqual(f.read(), yml(OLD))
|
|
|
|
def test_set_refuses_the_value_of_the_secrets_file(self):
|
|
result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=OLD)
|
|
self.assertEqual((result.returncode, result.stderr), (2, "the secrets file holds this value already\n"))
|
|
self.assertFalse(os.path.exists(self.env["SSH_LOG"]))
|
|
|
|
def test_check_reports_another_value(self):
|
|
result = self.run_script("check", "user@host", "GITEA_BOT_TOKEN", sha(NEW))
|
|
self.assertEqual((result.returncode, result.stdout), (1, "repo: differs; 0 backups hold that value\n"
|
|
"secrets.env: differs; 0 backups hold that value\n"))
|
|
|
|
def test_digest(self):
|
|
self.assertEqual(self.run_script("digest", "GITEA_BOT_TOKEN").stdout, sha(OLD))
|
|
|
|
def test_an_unreachable_host_is_an_error_with_2(self):
|
|
bin_dir = os.path.join(self.tmp, "bin")
|
|
for name, script in (("ssh", "exit 255"), ("sleep", "exit 0")):
|
|
with open(os.path.join(bin_dir, name), "w") as f:
|
|
f.write(f"#!/usr/bin/env bash\n{script}\n")
|
|
os.chmod(os.path.join(bin_dir, name), 0o755)
|
|
result = self.run_script("check", "user@host", "GITEA_BOT_TOKEN")
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertTrue(result.stderr.endswith("the host refused six connections\n"), result.stderr)
|
|
|
|
def test_usage(self):
|
|
self.assertEqual(self.run_script("--help").returncode, 0)
|
|
for args in (("set", "user@host"), ("probe", "GITEA_BOT_TOKEN", "x"), ("check", "user@host", "lower"), ("rotate",)):
|
|
self.assertEqual(self.run_script(*args).returncode, 2, args)
|
|
self.assertFalse(os.path.exists(self.env["SSH_LOG"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|