331 lines
16 KiB
Python
331 lines
16 KiB
Python
import io
|
|
import os
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
|
|
from tests.fakes import ROOT
|
|
|
|
sys.path.insert(0, os.path.join(ROOT, "tools"))
|
|
import release_rootfs # noqa: E402
|
|
|
|
DIGEST = "ab" * 32
|
|
HOST_CONFIG = '[model]\nid = "openrouter/m"\n\n[budget]\nmax_usd_per_review = 0.25\n'
|
|
|
|
|
|
def machine_config_text(rootfs, block=HOST_CONFIG):
|
|
"""A machine config with its archive in line 3 and the inline host config of the review after it."""
|
|
indented = "".join(f" {line}\n" if line else "\n" for line in block.splitlines())
|
|
return (f"werkdock:\n rootfs: {rootfs}\nbuilds:\n ai-review:\n env:\n"
|
|
f" WERKREVISOR_CONFIG_TOML: |\n{indented} OTHER: x\n")
|
|
|
|
|
|
def write(path, text):
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(text)
|
|
|
|
|
|
def read(path):
|
|
with open(path, encoding="utf-8") as f:
|
|
return f.read()
|
|
|
|
|
|
class ReleaseRootfsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.root, self.home, self.storage = (os.path.join(tmp.name, d) for d in ("repo", "home", "storage"))
|
|
os.makedirs(self.storage)
|
|
write(os.path.join(self.root, "Dockerfile"),
|
|
f"FROM docker.io/pragent/pr-agent:0.47.0@sha256:{DIGEST} AS base\nRUN true\nFROM base\nCOPY x y\n")
|
|
write(os.path.join(self.root, "tools", "pr-lens", "package-lock.json"), '{"lockfileVersion": 3}\n')
|
|
self.base = release_rootfs.base_name(self.root)
|
|
write(os.path.join(self.root, "werkrevisor", "__init__.py"), '__version__ = "0.9.1"\n')
|
|
write(os.path.join(self.root, "werkrevisor", "__pycache__", "x.pyc"), "stale")
|
|
write(os.path.join(self.root, "config", "reviewed.toml"), HOST_CONFIG)
|
|
self.reviewed = self.machine_config("reviewed",
|
|
machine_config_text(f"{self.storage}/werkrevisor-rootfs-0.9.0.tar.zst"))
|
|
self.other = self.machine_config("other", "werkdock:\n rootfs: /s/other.tar\n")
|
|
self.write_base()
|
|
|
|
def machine_config(self, repo, text):
|
|
path = os.path.join(self.home, repo, ".git", "werkator", ".werkator.yml")
|
|
write(path, "token: secret\n" + text)
|
|
return path
|
|
|
|
def write_base(self):
|
|
plain = os.path.join(self.storage, "base.tar")
|
|
with tarfile.open(plain, "w") as tar:
|
|
info = tarfile.TarInfo("usr/bin/python")
|
|
info.size = 2
|
|
tar.addfile(info, io.BytesIO(b"py"))
|
|
release_rootfs.compress(plain, os.path.join(self.storage, self.base))
|
|
os.remove(plain)
|
|
|
|
def release(self):
|
|
out = io.StringIO()
|
|
release_rootfs.release(self.storage, self.root, self.home, out)
|
|
return out.getvalue()
|
|
|
|
def archive_names(self, name):
|
|
plain = os.path.join(self.storage, "check.tar")
|
|
release_rootfs.decompress(os.path.join(self.storage, name), plain)
|
|
try:
|
|
with tarfile.open(plain) as tar:
|
|
return {m.name: m for m in tar.getmembers()}
|
|
finally:
|
|
os.remove(plain)
|
|
|
|
def test_a_new_version_is_assembled_and_installed(self):
|
|
"""Scenario#118.01"""
|
|
out = self.release()
|
|
|
|
members = self.archive_names("werkrevisor-rootfs-0.9.1.tar.zst")
|
|
self.assertEqual(set(members), {"usr/bin/python", "work", "werkrevisor", "werkrevisor/werkrevisor",
|
|
"werkrevisor/werkrevisor/__init__.py", "werkrevisor/config",
|
|
"werkrevisor/config/reviewed.toml"})
|
|
self.assertEqual((members["werkrevisor/config/reviewed.toml"].uid, members["werkrevisor/config"].mode),
|
|
(0, 0o755))
|
|
self.assertIn("werkrevisor-rootfs-0.9.1.tar.zst", read(self.reviewed))
|
|
self.assertIn("werkrevisor-rootfs-0.9.0.tar.zst", read(self.reviewed + ".bak-before-0.9.1"))
|
|
self.assertIn("reviewed: 3", out)
|
|
self.assertNotIn("secret", out)
|
|
self.assertEqual(sorted(os.listdir(os.path.dirname(self.reviewed))),
|
|
[".werkator.yml", ".werkator.yml.bak-before-0.9.1"])
|
|
|
|
def test_a_released_version_is_not_released_again(self):
|
|
"""Scenario#118.02"""
|
|
self.release()
|
|
before = read(self.reviewed + ".bak-before-0.9.1")
|
|
archive = os.path.join(self.storage, "werkrevisor-rootfs-0.9.1.tar.zst")
|
|
mtime = os.path.getmtime(archive)
|
|
|
|
out = self.release()
|
|
|
|
self.assertEqual(out, "werkrevisor-rootfs-0.9.1.tar.zst is already released\n")
|
|
self.assertEqual((os.path.getmtime(archive), read(self.reviewed + ".bak-before-0.9.1")), (mtime, before))
|
|
|
|
def test_a_missing_base_archive_fails_the_release(self):
|
|
"""Scenario#118.03"""
|
|
os.remove(os.path.join(self.storage, self.base))
|
|
before = read(self.reviewed)
|
|
|
|
with self.assertRaises(release_rootfs.ReleaseError) as raised:
|
|
self.release()
|
|
|
|
self.assertIn(self.base, str(raised.exception))
|
|
self.assertIn("tools/build-rootfs --base", str(raised.exception))
|
|
self.assertEqual((read(self.reviewed), os.listdir(self.storage)), (before, []))
|
|
|
|
def test_a_machine_config_without_a_werkrevisor_archive_stays_unchanged(self):
|
|
"""Scenario#118.04"""
|
|
before = read(self.other)
|
|
|
|
self.release()
|
|
|
|
self.assertEqual(read(self.other), before)
|
|
self.assertFalse(os.path.exists(self.other + ".bak-before-0.9.1"))
|
|
|
|
def test_a_release_archive_is_refused_as_base(self):
|
|
self.release()
|
|
os.replace(os.path.join(self.storage, "werkrevisor-rootfs-0.9.1.tar.zst"), os.path.join(self.storage, self.base))
|
|
write(os.path.join(self.root, "werkrevisor", "__init__.py"), '__version__ = "0.9.2"\n')
|
|
|
|
with self.assertRaises(release_rootfs.ReleaseError):
|
|
self.release()
|
|
|
|
self.assertFalse(os.path.exists(os.path.join(self.storage, "werkrevisor-rootfs-0.9.2.tar.zst")))
|
|
self.assertEqual(sorted(os.listdir(self.storage)), [self.base])
|
|
|
|
def test_a_new_lockfile_or_base_stage_names_a_new_base_archive(self):
|
|
write(os.path.join(self.root, "Dockerfile"), read(os.path.join(self.root, "Dockerfile")) + "ENV X=1\n")
|
|
self.assertEqual(release_rootfs.base_name(self.root), self.base)
|
|
|
|
write(os.path.join(self.root, "tools", "pr-lens", "package-lock.json"), '{"lockfileVersion": 3, "x": 1}\n')
|
|
by_lockfile = release_rootfs.base_name(self.root)
|
|
write(os.path.join(self.root, "Dockerfile"), "RUN apt-get install nodejs\n" + read(os.path.join(self.root, "Dockerfile")))
|
|
|
|
self.assertEqual(len({self.base, by_lockfile, release_rootfs.base_name(self.root)}), 3)
|
|
self.assertRegex(self.base, r"^werkrevisor-base-[0-9a-f]{12}\.tar\.zst$")
|
|
|
|
def test_install_keeps_the_mode_of_a_machine_config(self):
|
|
os.chmod(self.reviewed, 0o600)
|
|
name = "werkrevisor-rootfs-0.9.1.tar.zst"
|
|
write(os.path.join(self.storage, name), "new")
|
|
|
|
release_rootfs.install(self.storage, name, release_rootfs.sha256(os.path.join(self.storage, name)),
|
|
home=self.home, out=io.StringIO())
|
|
|
|
self.assertEqual(os.stat(self.reviewed).st_mode & 0o777, 0o600)
|
|
self.assertIn(name, read(self.reviewed))
|
|
|
|
def test_install_refuses_machine_configs_with_another_archive_directory(self):
|
|
elsewhere = self.machine_config("elsewhere", "werkdock:\n rootfs: /t/werkrevisor-rootfs-0.9.0.tar.zst\n")
|
|
name = "werkrevisor-rootfs-0.9.1.tar.zst"
|
|
write(os.path.join(self.storage, name), "new")
|
|
before = read(self.reviewed), read(elsewhere)
|
|
|
|
with self.assertRaises(release_rootfs.ReleaseError) as raised:
|
|
release_rootfs.install(self.storage, name, release_rootfs.sha256(os.path.join(self.storage, name)),
|
|
home=self.home, out=io.StringIO())
|
|
|
|
self.assertIn("/t", str(raised.exception))
|
|
self.assertEqual((read(self.reviewed), read(elsewhere)), before)
|
|
|
|
def test_install_refuses_an_archive_with_another_checksum(self):
|
|
name = "werkrevisor-rootfs-0.9.1.tar.zst"
|
|
write(os.path.join(self.storage, name), "archive")
|
|
before = read(self.reviewed)
|
|
|
|
with self.assertRaises(release_rootfs.ReleaseError):
|
|
release_rootfs.install(self.storage, name, "0" * 64, home=self.home, out=io.StringIO())
|
|
|
|
self.assertEqual(read(self.reviewed), before)
|
|
|
|
def test_install_keeps_the_two_newest_archives_and_the_one_in_use(self):
|
|
for v in ("0.8.0", "0.8.1", "0.9.0"):
|
|
write(os.path.join(self.storage, f"werkrevisor-rootfs-{v}.tar.zst"), v)
|
|
self.machine_config("pinned", f"werkdock:\n rootfs: {self.storage}/werkrevisor-rootfs-0.8.0.tar.zst\n")
|
|
name = "werkrevisor-rootfs-0.9.1.tar.zst"
|
|
write(os.path.join(self.storage, name), "new")
|
|
|
|
release_rootfs.install(self.storage, name, release_rootfs.sha256(os.path.join(self.storage, name)),
|
|
["reviewed"], self.home, io.StringIO())
|
|
|
|
self.assertEqual(sorted(n for n in os.listdir(self.storage) if n.startswith("werkrevisor-rootfs-")),
|
|
["werkrevisor-rootfs-0.8.0.tar.zst", "werkrevisor-rootfs-0.9.0.tar.zst", name])
|
|
|
|
def test_a_version_in_the_archive_directory_does_not_change_which_archives_stay(self):
|
|
storage = os.path.join(self.storage, "releases-1.0")
|
|
os.makedirs(storage)
|
|
for v in ("0.8.0", "0.9.0", "0.10.0", "0.11.0"):
|
|
write(os.path.join(storage, f"werkrevisor-rootfs-{v}.tar.zst"), v)
|
|
self.machine_config("reviewed", f"werkdock:\n rootfs: {storage}/werkrevisor-rootfs-0.10.0.tar.zst\n")
|
|
name = "werkrevisor-rootfs-0.11.0.tar.zst"
|
|
|
|
release_rootfs.install(storage, name, release_rootfs.sha256(os.path.join(storage, name)),
|
|
["reviewed"], self.home, io.StringIO())
|
|
|
|
self.assertEqual(sorted(os.listdir(storage)), ["werkrevisor-rootfs-0.10.0.tar.zst", name])
|
|
|
|
def test_the_archive_directory_comes_from_the_machine_configs(self):
|
|
self.assertEqual(release_rootfs.archive_directory(self.home), self.storage)
|
|
self.machine_config("elsewhere", "werkdock:\n rootfs: /t/werkrevisor-rootfs-0.9.0.tar.zst\n")
|
|
with self.assertRaises(release_rootfs.ReleaseError):
|
|
release_rootfs.archive_directory(self.home)
|
|
|
|
def block_of(self, path):
|
|
lines = read(path).split("\n")
|
|
start = lines.index(" WERKREVISOR_CONFIG_TOML: |") + 1
|
|
end = next(i for i in range(start, len(lines)) if lines[i].startswith(" OTHER"))
|
|
return "".join(line[8:] + "\n" for line in lines[start:end]).rstrip("\n") + "\n"
|
|
|
|
def assert_nothing_changed(self, before):
|
|
self.assertEqual({p: read(p) for p in before}, before)
|
|
self.assertEqual(os.listdir(self.storage), [self.base])
|
|
self.assertFalse(any(".bak-" in n for p in before for n in os.listdir(os.path.dirname(p))))
|
|
|
|
def test_a_stale_host_key_is_replaced_with_the_archive(self):
|
|
"""Scenario#124.01"""
|
|
old = HOST_CONFIG + '\n[doc_check]\nmodel = "sk-stale-value"\n'
|
|
self.machine_config("reviewed", machine_config_text(f"{self.storage}/werkrevisor-rootfs-0.9.0.tar.zst", old))
|
|
|
|
out = self.release()
|
|
|
|
self.assertIn("werkrevisor-rootfs-0.9.1.tar.zst", read(self.reviewed))
|
|
self.assertEqual(self.block_of(self.reviewed), HOST_CONFIG)
|
|
self.assertIn("OTHER: x", read(self.reviewed))
|
|
backup = self.reviewed + ".bak-before-0.9.1"
|
|
self.assertEqual((self.block_of(backup), "werkrevisor-rootfs-0.9.0.tar.zst" in read(backup)), (old, True))
|
|
self.assertIn("reviewed: host config from config/reviewed.toml; dropped unknown keys: doc_check.model", out)
|
|
self.assertNotIn("sk-stale-value", out)
|
|
self.assertNotIn("secret", out)
|
|
|
|
def test_a_host_setting_that_the_file_would_lose_stops_the_release(self):
|
|
"""Scenario#124.02"""
|
|
rootfs = f"{self.storage}/werkrevisor-rootfs-0.9.0.tar.zst"
|
|
write(os.path.join(self.root, "config", "first.toml"), HOST_CONFIG + "\n[size]\nmax_files = 10\n")
|
|
first = self.machine_config("first", machine_config_text(rootfs))
|
|
self.machine_config("reviewed", machine_config_text(rootfs, HOST_CONFIG.replace("openrouter/m", "sk-own-model")
|
|
+ "min_key_remaining_usd = 0.5\n"))
|
|
before = {p: read(p) for p in (first, self.reviewed, self.other)}
|
|
|
|
with self.assertRaises(release_rootfs.ReleaseError) as raised:
|
|
self.release()
|
|
|
|
self.assertIn("reviewed: the host sets keys that config/reviewed.toml lacks or sets otherwise: "
|
|
"budget.min_key_remaining_usd, model.id", str(raised.exception))
|
|
self.assertNotIn("sk-own-model", str(raised.exception))
|
|
self.assertNotIn("first", str(raised.exception))
|
|
self.assert_nothing_changed(before)
|
|
|
|
def test_a_host_config_that_cannot_be_checked_stops_the_release(self):
|
|
"""Scenario#124.03"""
|
|
rootfs = f"{self.storage}/werkrevisor-rootfs-0.9.0.tar.zst"
|
|
write(os.path.join(self.root, "config", "rejected.toml"), HOST_CONFIG + "\n[doc_check]\nmodel = \"x\"\n")
|
|
write(os.path.join(self.root, "config", "blockless.toml"), HOST_CONFIG)
|
|
write(os.path.join(self.root, "config", "broken.toml"), HOST_CONFIG)
|
|
paths = [self.machine_config("unknown", machine_config_text(rootfs)),
|
|
self.machine_config("rejected", machine_config_text(rootfs)),
|
|
self.machine_config("blockless", f"werkdock:\n rootfs: {rootfs}\n"),
|
|
self.machine_config("broken", machine_config_text(rootfs, "[model\nid = 'sk-x'\n"))]
|
|
before = {p: read(p) for p in [self.reviewed, *paths]}
|
|
|
|
with self.assertRaises(release_rootfs.ReleaseError) as raised:
|
|
self.release()
|
|
|
|
message = str(raised.exception)
|
|
self.assertIn("unknown: no config/unknown.toml", message)
|
|
self.assertIn("rejected: the new version rejects config/rejected.toml: "
|
|
"host config: [doc_check] has unknown keys: model", message)
|
|
self.assertIn("blockless: 0 WERKREVISOR_CONFIG_TOML blocks", message)
|
|
self.assertIn("broken: invalid TOML in the host block at line 1", message)
|
|
self.assertNotIn("sk-x", message)
|
|
self.assert_nothing_changed(before)
|
|
|
|
def test_a_host_block_changed_after_the_check_stops_the_install(self):
|
|
rootfs = f"{self.storage}/werkrevisor-rootfs-0.9.0.tar.zst"
|
|
self.machine_config("reviewed", machine_config_text(rootfs, HOST_CONFIG + '\n[doc_check]\nmodel = "x"\n'))
|
|
blocks = release_rootfs.host_config_blocks({"reviewed": self.reviewed}, self.root, io.StringIO())
|
|
# The operator switches the model while the archive is assembled.
|
|
changed = read(self.reviewed).replace("openrouter/m", "sk-own-model")
|
|
write(self.reviewed, changed)
|
|
name = "werkrevisor-rootfs-0.9.1.tar.zst"
|
|
write(os.path.join(self.storage, name), "new")
|
|
|
|
with self.assertRaises(release_rootfs.ReleaseError) as raised:
|
|
release_rootfs.install(self.storage, name, release_rootfs.sha256(os.path.join(self.storage, name)),
|
|
home=self.home, out=io.StringIO(), blocks=blocks)
|
|
|
|
self.assertEqual(str(raised.exception),
|
|
"reviewed: the host block changed since the check; run the release again")
|
|
self.assertEqual(read(self.reviewed), changed)
|
|
self.assertFalse(os.path.exists(self.reviewed + ".bak-before-0.9.1"))
|
|
|
|
def test_an_equal_host_block_changed_after_the_check_stops_the_install(self):
|
|
blocks = release_rootfs.host_config_blocks({"reviewed": self.reviewed}, self.root, io.StringIO())
|
|
# The operator adds a setting that the new version rejects while the archive is assembled.
|
|
stale = ' [doc_check]\n model = "x"\n'
|
|
changed = read(self.reviewed).replace(" OTHER: x", stale + " OTHER: x")
|
|
write(self.reviewed, changed)
|
|
name = "werkrevisor-rootfs-0.9.1.tar.zst"
|
|
write(os.path.join(self.storage, name), "new")
|
|
|
|
with self.assertRaises(release_rootfs.ReleaseError):
|
|
release_rootfs.install(self.storage, name, release_rootfs.sha256(os.path.join(self.storage, name)),
|
|
home=self.home, out=io.StringIO(), blocks=blocks)
|
|
|
|
self.assertEqual(read(self.reviewed), changed)
|
|
|
|
def test_a_host_block_equal_to_its_file_stays_as_it_is(self):
|
|
out = self.release()
|
|
|
|
self.assertEqual(self.block_of(self.reviewed), HOST_CONFIG)
|
|
self.assertNotIn("host config from", out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|