fuzzyworld

fuzzyworld-tools / files / tests / test_deploys.py

test_deploys.py

Blame: the commit that last changed each line · file · history

258a9da8 2026-11-06 Pavel Safronov1"""Tests for server/deploys.py (`fw deploys`: the deploy ledger and deployed apps against their repos)."""
2import importlib.machinery
3import importlib.util
4import json
5import os
6import shutil
7import subprocess
8import tempfile
9import unittest
10from unittest import mock
11
12ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
13spec = importlib.util.spec_from_file_location("deploys", os.path.join(ROOT, "server", "deploys.py"))
14deploys = importlib.util.module_from_spec(spec)
15spec.loader.exec_module(deploys)
16
17ENV = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@x"}
18PKG = {"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
25class 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
163if __name__ == "__main__":
164 unittest.main()