fuzzyworld-tools / files / tests / test_deploys.py
test_deploys.py
| 1 | """Tests for server/deploys.py (`fw deploys`: the deploy ledger and deployed apps against their repos).""" |
| 2 | import importlib.machinery |
| 3 | import importlib.util |
| 4 | import json |
| 5 | import os |
| 6 | import shutil |
| 7 | import subprocess |
| 8 | import tempfile |
| 9 | import unittest |
| 10 | from unittest import mock |
| 11 | |
| 12 | ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") |
| 13 | spec = importlib.util.spec_from_file_location("deploys", os.path.join(ROOT, "server", "deploys.py")) |
| 14 | deploys = importlib.util.module_from_spec(spec) |
| 15 | spec.loader.exec_module(deploys) |
| 16 | |
| 17 | ENV = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@x"} |
| 18 | PKG = {"name": "tools", "fuzzyworld": {"deploy": { |
| 19 | "server": {"paths": ["server"], "command": ["server/install.sh"], "description": "the server"}, |
| 20 | "laptop-runner": {"paths": ["runner"], "command": ["x"], "perMachine": True}, |
| 21 | "../evil": {"paths": ["server"], "command": ["x"]}, |
| 22 | "bad": {"paths": [], "command": ["x"]}}}} |
| 23 | |
| 24 | |
| 25 | class Deploys(unittest.TestCase): |
| 26 | def setUp(self): |
| 27 | self.tmp = tempfile.mkdtemp() |
| 28 | self.addCleanup(shutil.rmtree, self.tmp, True) |
| 29 | self.git_root, self.state, self.apps = (os.path.join(self.tmp, d) for d in ("git", "state", "apps")) |
| 30 | for d in (self.git_root, self.state, self.apps): |
| 31 | os.makedirs(d) |
| 32 | self.work = os.path.join(self.tmp, "work") |
| 33 | subprocess.run(["git", "init", "-q", self.work], check=True) |
| 34 | os.makedirs(os.path.join(self.work, "server")) |
| 35 | os.makedirs(os.path.join(self.work, "runner")) |
| 36 | self.write("package.json", json.dumps(PKG)) |
| 37 | self.write("server/a.txt", "1") |
| 38 | self.write("runner/r.txt", "1") |
| 39 | self.first = self.commit("first") |
| 40 | for name, value in [("GIT_ROOT", self.git_root), ("STATE", self.state), ("APPS", self.apps)]: |
| 41 | p = mock.patch.object(deploys, name, value) |
| 42 | p.start() |
| 43 | self.addCleanup(p.stop) |
| 44 | |
| 45 | def write(self, rel, text): |
| 46 | with open(os.path.join(self.work, rel), "w") as f: |
| 47 | f.write(text) |
| 48 | |
| 49 | def commit(self, msg): |
| 50 | subprocess.run(["git", "-C", self.work, "add", "-A"], check=True) |
| 51 | subprocess.run(["git", "-C", self.work, "commit", "-qm", msg], check=True, env=ENV) |
| 52 | sha = subprocess.run(["git", "-C", self.work, "rev-parse", "HEAD"], check=True, capture_output=True, text=True).stdout.strip() |
| 53 | self.sync("tools") |
| 54 | return sha |
| 55 | |
| 56 | def sync(self, name): |
| 57 | bare = os.path.join(self.git_root, name + ".git") |
| 58 | if os.path.isdir(bare): |
| 59 | shutil.rmtree(bare) |
| 60 | subprocess.run(["git", "clone", "-q", "--bare", self.work, bare], check=True) |
| 61 | |
| 62 | def tree(self, path): |
| 63 | return subprocess.run(["git", "-C", self.work, "rev-parse", f"HEAD:{path}"], check=True, capture_output=True, text=True).stdout.strip() |
| 64 | |
| 65 | def ledger(self, data): |
| 66 | with open(os.path.join(self.state, "deployed.json"), "w") as f: |
| 67 | json.dump(data, f) |
| 68 | |
| 69 | def target(self, data, name): |
| 70 | (repo,) = data["repos"] |
| 71 | return next(t for t in repo["targets"] if t["target"] == name) |
| 72 | |
| 73 | def test_never_current_pending_and_behind(self): |
| 74 | data = deploys.collect() |
| 75 | self.assertEqual([t["target"] for t in data["repos"][0]["targets"]], ["server", "laptop-runner"]) # bad names/paths dropped |
| 76 | self.assertEqual(self.target(data, "server")["entries"][0]["state"], "never") |
| 77 | self.ledger({"tools/server": {"hash": self.tree("server"), "commit": self.first[:7], "at": "2026-11-04 10:00"}}) |
| 78 | self.assertEqual(self.target(deploys.collect(), "server")["entries"][0]["state"], "current") |
| 79 | self.write("README.md", "docs") # unrelated change: still current, not behind for this target |
| 80 | self.commit("docs only") |
| 81 | e = self.target(deploys.collect(), "server")["entries"][0] |
| 82 | self.assertEqual((e["state"], e["behind"]["count"]), ("current", 0)) |
| 83 | self.write("server/a.txt", "2") |
| 84 | self.commit("change server") |
| 85 | e = self.target(deploys.collect(), "server")["entries"][0] |
| 86 | self.assertEqual((e["state"], e["behind"]["count"], e["behind"]["commits"][0][1]), ("pending", 1, "change server")) |
| 87 | |
| 88 | def test_uncommitted_deploy_or_unpushed_commit_is_pending_with_unknown_behind(self): |
| 89 | self.ledger({"tools/server": {"hash": "deadbeef", "commit": "abc1234+uncommitted", "at": "t"}}) |
| 90 | e = self.target(deploys.collect(), "server")["entries"][0] |
| 91 | self.assertEqual((e["state"], e["behind"]), ("pending", None)) |
| 92 | |
| 93 | def test_per_machine_target_lists_each_machine(self): |
| 94 | h = self.tree("runner") |
| 95 | self.ledger({"tools/laptop-runner@mac-a": {"hash": h, "commit": "x", "at": "1"}, |
| 96 | "tools/laptop-runner@mac-b": {"hash": "old", "commit": "y", "at": "2"}}) |
| 97 | entries = self.target(deploys.collect(), "laptop-runner")["entries"] |
| 98 | self.assertEqual([(e["machine"], e["state"]) for e in entries], [("mac-a", "current"), ("mac-b", "pending")]) |
| 99 | |
| 100 | def test_apps_against_their_repo(self): |
| 101 | for app, commit in [("one", self.first), ("two", "f" * 40), ("three", self.first)]: |
| 102 | os.makedirs(os.path.join(self.apps, app)) |
| 103 | with open(os.path.join(self.apps, app, ".fuzzyworld.json"), "w") as f: |
| 104 | json.dump({"app": app, "title": app.title(), "repo": "tools" if app != "three" else "missing", |
| 105 | "commit": commit, "deployed": "2026-11-04T10:00:00Z"}, f) |
| 106 | self.write("server/a.txt", "3") |
| 107 | self.commit("later") |
| 108 | by = {a["app"]: a for a in deploys.collect()["apps"]} |
| 109 | self.assertEqual((by["one"]["state"], by["one"]["behind"]["count"]), ("pending", 1)) |
| 110 | self.assertEqual((by["two"]["state"], by["two"]["behind"]), ("unknown", None)) # commit not in the repo |
| 111 | self.assertEqual(by["three"]["state"], "unknown") # repo not on the server |
| 112 | |
| 113 | def test_app_ignores_commits_that_only_touch_docs_and_ci_config(self): |
| 114 | os.makedirs(os.path.join(self.apps, "one")) |
| 115 | with open(os.path.join(self.apps, "one", ".fuzzyworld.json"), "w") as f: |
| 116 | json.dump({"app": "one", "repo": "tools", "commit": self.first, "deployed": "2026-11-04T10:00:00Z"}, f) |
| 117 | for rel, text in [("README.md", "x"), (".fw-ci.yml", "x"), ("docs/a.txt", "x"), ("sub/NOTES.md", "x")]: |
| 118 | os.makedirs(os.path.dirname(os.path.join(self.work, rel)) or self.work, exist_ok=True) |
| 119 | self.write(rel, text) |
| 120 | self.commit("docs " + rel) |
| 121 | app = deploys.collect()["apps"][0] |
| 122 | self.assertEqual((app["state"], app["behind"]["count"]), ("current", 0)) |
| 123 | self.write("server/a.txt", "code") |
| 124 | self.write("README.md", "also docs") |
| 125 | self.commit("real change") # touches code and a doc: counts |
| 126 | app = deploys.collect()["apps"][0] |
| 127 | self.assertEqual((app["state"], app["behind"]["count"], app["behind"]["commits"][0][1]), ("pending", 1, "real change")) |
| 128 | |
| 129 | def test_hash_equals_what_deploy_tools_records(self): |
| 130 | loader = importlib.machinery.SourceFileLoader("deploy_tools", os.path.join(ROOT, "bin", "deploy-tools")) |
| 131 | dt = importlib.util.module_from_spec(importlib.util.spec_from_loader("deploy_tools", loader)) |
| 132 | loader.exec_module(dt) |
| 133 | with mock.patch.object(dt, "ROOT", self.work): |
| 134 | recorded = dt.content_hash(["server", "runner"]) |
| 135 | self.assertEqual(deploys.tree_hash(os.path.join(self.git_root, "tools.git"), ["server", "runner"]), recorded) |
| 136 | |
| 137 | def test_commit_without_a_subject_and_odd_ledger_and_app_values(self): |
| 138 | self.write("server/a.txt", "2") |
| 139 | subprocess.run(["git", "-C", self.work, "add", "-A"], check=True) |
| 140 | subprocess.run(["git", "-C", self.work, "commit", "-q", "--allow-empty-message", "-m", ""], check=True, env=ENV) |
| 141 | self.sync("tools") |
| 142 | self.ledger({"tools/server": {"hash": "old", "commit": self.first[:7], "at": "t"}, "tools/laptop-runner@m": "not a dict"}) |
| 143 | data = deploys.collect() |
| 144 | self.assertEqual(self.target(data, "server")["entries"][0]["behind"]["commits"][0][1], "") |
| 145 | self.assertEqual(self.target(data, "laptop-runner")["entries"][0]["state"], "pending") |
| 146 | os.makedirs(os.path.join(self.apps, "odd")) |
| 147 | with open(os.path.join(self.apps, "odd", ".fuzzyworld.json"), "w") as f: |
| 148 | json.dump({"repo": "../x", "commit": "zzz"}, f) |
| 149 | self.assertEqual({a["app"]: a["state"] for a in deploys.collect()["apps"]}, {"odd": "unknown"}) |
| 150 | |
| 151 | def test_paths_must_be_written_plainly(self): |
| 152 | for ok in ("server", "a/b"): |
| 153 | self.assertTrue(deploys.clean_path(ok)) |
| 154 | for bad in ("", "./server", "server/", "/abs", "..", "../x", "a//b", None, 3): |
| 155 | self.assertFalse(deploys.clean_path(bad), bad) |
| 156 | |
| 157 | def test_text_and_nothing_to_report(self): |
| 158 | self.assertIn("never", deploys.text(deploys.collect())) |
| 159 | shutil.rmtree(self.git_root) |
| 160 | self.assertEqual(deploys.collect(), {"repos": [], "apps": []}) |
| 161 | |
| 162 | |
| 163 | if __name__ == "__main__": |
| 164 | unittest.main() |