import json import os import tempfile import unittest from _common import agent 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") if __name__ == "__main__": unittest.main()