215 lines
9.4 KiB
Python
215 lines
9.4 KiB
Python
import hashlib
|
||
import json
|
||
import os
|
||
import tempfile
|
||
import unittest
|
||
|
||
from _common import SETTINGS, agent
|
||
from fake_hub import FakeHub
|
||
|
||
CUR = agent.AGENT_VERSION
|
||
NEXT = CUR + 1
|
||
|
||
|
||
class UpdateStateTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.dir = tempfile.TemporaryDirectory()
|
||
self.path = os.path.join(self.dir.name, "update.json")
|
||
|
||
def tearDown(self):
|
||
self.dir.cleanup()
|
||
|
||
def test_write_and_read(self):
|
||
d = agent.write_update_state(self.path, "staged", 5, 6, "abc")
|
||
self.assertEqual(agent.read_update_state(self.path), d)
|
||
self.assertEqual((d["state"], d["from_version"], d["to_version"], d["sha256"], d["error"]), ("staged", 5, 6, "abc", None))
|
||
self.assertTrue(d["at"])
|
||
self.assertEqual(os.listdir(self.dir.name), ["update.json"]) # временный файл не остался
|
||
|
||
def test_read_missing(self):
|
||
self.assertIsNone(agent.read_update_state(self.path))
|
||
|
||
def test_read_update_state_garbage(self):
|
||
for text in ('{"state": "tri', "[1, 2]", ""):
|
||
with open(self.path, "w", encoding="utf-8") as f:
|
||
f.write(text)
|
||
self.assertIsNone(agent.read_update_state(self.path))
|
||
|
||
def test_confirm_trial_of_this_version(self):
|
||
u = agent.Updater(None, self.dir.name, 1)
|
||
agent.write_update_state(u.state_path, "trial", 4, agent.AGENT_VERSION, "s")
|
||
u.confirm()
|
||
st = agent.read_update_state(u.state_path)
|
||
self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("ok", 4, agent.AGENT_VERSION, "s"))
|
||
|
||
def test_confirm_ignores_other_states(self):
|
||
u = agent.Updater(None, self.dir.name, 1)
|
||
u.confirm()
|
||
self.assertIsNone(u.state()) # нет файла — не создаётся
|
||
agent.write_update_state(u.state_path, "trial", 4, agent.AGENT_VERSION + 1, "s")
|
||
u.confirm()
|
||
self.assertEqual(u.state()["state"], "trial") # пробная другая версия — не наша
|
||
agent.write_update_state(u.state_path, "rolled_back", 5, 6, "s", "код 1 через 3 с")
|
||
u.confirm()
|
||
self.assertEqual(u.state()["state"], "rolled_back")
|
||
|
||
|
||
def make_release(version, body="print('ok')\n"):
|
||
content = "AGENT_VERSION = %d\n%s" % (version, body)
|
||
return {"version": version, "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), "content": content}
|
||
|
||
|
||
class UpdaterCheckTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.dir = tempfile.TemporaryDirectory()
|
||
self.hub = FakeHub(SETTINGS)
|
||
cfg = {"hub_url": self.hub.base_url, "hub_token": self.hub.token, "timeout": 5}
|
||
self.u = agent.Updater(agent.HubClient(cfg), self.dir.name, 1)
|
||
self.selftests = []
|
||
self.code = 0
|
||
self.u.run_selftest = lambda path: (self.selftests.append(path), (self.code, "вывод selftest"))[1]
|
||
self.now = [1000.0]
|
||
self.u.clock = lambda: self.now[0]
|
||
|
||
def tearDown(self):
|
||
self.hub.stop()
|
||
self.dir.cleanup()
|
||
|
||
def ref(self, rel):
|
||
return {"version": rel["version"], "sha256": rel["sha256"]}
|
||
|
||
def test_good_version_is_staged(self):
|
||
rel = make_release(NEXT)
|
||
self.hub.release = rel
|
||
self.assertTrue(self.u.check(self.ref(rel)))
|
||
with open(self.u.new_path, encoding="utf-8") as f:
|
||
self.assertEqual(f.read(), rel["content"])
|
||
st = self.u.state()
|
||
self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("staged", CUR, NEXT, rel["sha256"]))
|
||
self.assertEqual(self.selftests, [self.u.new_path])
|
||
|
||
def test_not_newer_or_disabled_or_no_release(self):
|
||
self.hub.release = make_release(CUR)
|
||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||
self.assertFalse(self.u.check(None))
|
||
self.hub.release = make_release(NEXT)
|
||
self.u.enabled = False
|
||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||
self.assertEqual(self.hub.release_requests, 0)
|
||
|
||
def test_sha_mismatch_rejected_once(self):
|
||
rel = make_release(NEXT)
|
||
self.hub.release = dict(rel, content=rel["content"] + "# подмена\n")
|
||
ref = self.ref(rel)
|
||
self.hub.release["sha256"] = rel["sha256"]
|
||
self.assertFalse(self.u.check(ref))
|
||
st = self.u.state()
|
||
self.assertEqual((st["state"], st["to_version"], st["error"]), ("rejected", NEXT, "sha256 не сходится"))
|
||
self.assertFalse(os.path.exists(self.u.new_path))
|
||
self.assertFalse(self.u.check(ref))
|
||
self.assertEqual(self.hub.release_requests, 1) # отвергнутую версию больше не качает
|
||
|
||
def test_syntax_error_and_wrong_version_rejected(self):
|
||
bad = make_release(NEXT, body="def (:\n")
|
||
self.hub.release = bad
|
||
self.assertFalse(self.u.check(self.ref(bad)))
|
||
self.assertIn("не компилируется", self.u.state()["error"])
|
||
wrong = make_release(NEXT + 1)
|
||
wrong = dict(wrong, version=NEXT + 2)
|
||
self.hub.release = wrong
|
||
self.assertFalse(self.u.check(self.ref(wrong)))
|
||
self.assertIn("AGENT_VERSION", self.u.state()["error"])
|
||
self.assertEqual(self.selftests, [])
|
||
|
||
def test_selftest_bad_code_rejected(self):
|
||
self.hub.release = make_release(NEXT)
|
||
self.code = 1
|
||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||
st = self.u.state()
|
||
self.assertEqual(st["state"], "rejected")
|
||
self.assertTrue(st["error"].startswith("selftest: код 1"))
|
||
self.assertIn("вывод selftest", st["error"])
|
||
self.assertFalse(os.path.exists(self.u.new_path))
|
||
|
||
def test_selftest_unavailable_is_transient(self):
|
||
self.hub.release = make_release(NEXT)
|
||
self.code = 3
|
||
ref = self.ref(self.hub.release)
|
||
self.assertFalse(self.u.check(ref))
|
||
self.assertIsNone(self.u.state()) # update.json не тронут
|
||
self.assertFalse(os.path.exists(self.u.new_path))
|
||
self.now[0] += 299
|
||
self.assertFalse(self.u.check(ref))
|
||
self.assertEqual(len(self.selftests), 1) # раньше 5 минут не повторяет
|
||
self.now[0] += 2
|
||
self.code = 0
|
||
self.assertTrue(self.u.check(ref))
|
||
|
||
def test_selftest_timeout_is_transient(self):
|
||
self.hub.release = make_release(NEXT)
|
||
self.code = None
|
||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||
self.assertIsNone(self.u.state())
|
||
|
||
def test_download_failure_is_transient(self):
|
||
rel = make_release(NEXT) # в claim есть, а /agent/release — 404
|
||
self.assertFalse(self.u.check(self.ref(rel)))
|
||
self.assertIsNone(self.u.state())
|
||
|
||
def test_release_changed_between_requests_is_transient(self):
|
||
old, new = make_release(NEXT), make_release(NEXT + 1)
|
||
self.hub.release = new
|
||
self.assertFalse(self.u.check(self.ref(old)))
|
||
self.assertIsNone(self.u.state())
|
||
self.assertEqual(self.selftests, [])
|
||
|
||
def test_rolled_back_same_sha_not_retried(self):
|
||
rel = make_release(NEXT)
|
||
self.hub.release = rel
|
||
agent.write_update_state(self.u.state_path, "rolled_back", CUR, NEXT, rel["sha256"], "код 1 через 3 с")
|
||
self.assertFalse(self.u.check(self.ref(rel)))
|
||
fixed = make_release(NEXT, body="print('fixed')\n")
|
||
self.hub.release = fixed
|
||
self.assertTrue(self.u.check(self.ref(fixed))) # тот же номер, другой sha — пробуем
|
||
|
||
def test_without_launcher_only_hint(self):
|
||
self.u.launcher_version = None
|
||
self.hub.release = make_release(NEXT)
|
||
logs = []
|
||
orig = agent.log
|
||
agent.log = logs.append
|
||
try:
|
||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||
self.now[0] += 3600
|
||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||
finally:
|
||
agent.log = orig
|
||
self.assertEqual(logs.count("доступна v%d; для самообновления запускайте через run.py" % NEXT), 2)
|
||
self.assertEqual(self.hub.release_requests, 0)
|
||
|
||
def test_stop_during_selftest_is_not_reject(self):
|
||
self.hub.release = make_release(NEXT)
|
||
stop = [False]
|
||
self.u.should_stop = lambda: stop[0]
|
||
|
||
def selftest(path):
|
||
stop[0] = True # Ctrl-C пришёл и selftest упал с 1
|
||
return 1, "KeyboardInterrupt"
|
||
self.u.run_selftest = selftest
|
||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||
self.assertIsNone(self.u.state())
|
||
self.assertFalse(os.path.exists(self.u.new_path))
|
||
|
||
def test_real_selftest_runs_new_file(self):
|
||
u = agent.Updater(None, self.dir.name, 1, config_path=os.path.join(self.dir.name, "agent.json"))
|
||
path = os.path.join(self.dir.name, "agent.py.new")
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
f.write("import sys\nprint('args', sys.argv[1:])\nsys.exit(3)\n")
|
||
code, out = u._run_selftest(path)
|
||
self.assertEqual(code, 3)
|
||
self.assertIn("'--config', '%s', 'selftest'" % os.path.join(self.dir.name, "agent.json"), out)
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|