import hashlib import unittest from _common import SETTINGS, agent, jira_cfg, no_sleep from fake_hub import FakeHub from fake_jira import FakeJira SHA_ABC = hashlib.sha256(b"abc").hexdigest() class EnsureAttachmentTest(unittest.TestCase): def setUp(self): self.fake = FakeJira() self.hub = FakeHub(SETTINGS) cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token) self.jira = agent.JiraClient(cfg) self.jira.sleep = no_sleep self.ex = agent.Executor(self.jira, dict(SETTINGS), agent.HubClient(cfg)) self.ex.sleep = no_sleep self.fake.add("AURORA-1") def tearDown(self): self.fake.stop() self.hub.stop() def run_cmd(self, **params): return self.ex.run({"id": 1, "type": "ensure_attachment", "params": params}) def test_upload_cyrillic_name(self): data = bytes(range(256)) * 8 sha = self.hub.put_file(data) name = "ОСВ сч. 60 за 2 кв. 2026.xlsx" out = self.run_cmd(key="AURORA-1", file=sha, filename=name, size=len(data)) self.assertEqual(out["status"], "done", out) a = self.fake.attachments["AURORA-1"][0] self.assertEqual(out["result"], {"attachment_id": a["id"], "created": True}) self.assertEqual((a["filename"], a["size"], a["data"]), (name, len(data), data)) self.assertEqual(self.fake.requests, [("GET", "/rest/api/2/issue/AURORA-1"), ("POST", "/rest/api/2/issue/AURORA-1/attachments")]) self.assertEqual(self.fake.queries[0].get("fields"), "attachment") self.assertEqual(self.hub.file_requests, 1) self.assertEqual(out["jira_calls"], 2) def test_existing_same_name_and_size(self): a = self.fake.add_attachment("AURORA-1", "ОСВ 60.xlsx", b"xyz") # другое содержимое, тот же размер out = self.run_cmd(key="AURORA-1", file=SHA_ABC, filename="ОСВ 60.xlsx", size=3) self.assertEqual(out["status"], "done") self.assertEqual(out["result"], {"attachment_id": a["id"], "existed": True}) self.assertEqual(out["jira_calls"], 1) self.assertEqual(self.hub.file_requests, 0) # файл у хаба не запрашивается self.assertEqual(len(self.fake.attachments["AURORA-1"]), 1) def test_same_name_other_size_uploads_second(self): old = self.fake.add_attachment("AURORA-1", "ОСВ 60.xlsx", "старое содержимое".encode("utf-8")) sha = self.hub.put_file(b"abc") out = self.run_cmd(key="AURORA-1", file=sha, filename="ОСВ 60.xlsx", size=3) self.assertEqual(out["status"], "done") self.assertTrue(out["result"]["created"]) self.assertNotEqual(out["result"]["attachment_id"], old["id"]) self.assertEqual([x["filename"] for x in self.fake.attachments["AURORA-1"]], ["ОСВ 60.xlsx", "ОСВ 60.xlsx"]) def test_existing_under_wire_name(self): name = 'итог "кв.2".txt' a = self.fake.add_attachment("AURORA-1", agent.wire_filename(name), b"abc") # так Jira сохранила прошлую загрузку out = self.run_cmd(key="AURORA-1", file=SHA_ABC, filename=name, size=3) self.assertEqual(out["result"], {"attachment_id": a["id"], "existed": True}) self.assertEqual(self.hub.file_requests, 0) def test_size_null_fails_without_calls(self): out = self.run_cmd(key="AURORA-1", file=SHA_ABC, filename="a.txt", size=None) self.assertEqual(out["status"], "failed") self.assertEqual(out["error"], "файл удалён с хаба, загрузите заново") self.assertNotIn("retryable", out) self.assertEqual(self.fake.requests, []) self.assertEqual(self.hub.file_requests, 0) def test_hub_404_file_gone(self): out = self.run_cmd(key="AURORA-1", file=SHA_ABC, filename="a.txt", size=3) # в хранилище хаба файла нет self.assertEqual(out["status"], "failed") self.assertEqual(out["error"], "файл удалён с хаба, загрузите заново") self.assertNotIn("retryable", out) self.assertEqual(self.hub.file_requests, 1) self.assertFalse(any(m == "POST" for m, _ in self.fake.requests)) def test_corrupted_sha_is_retryable_hub_failure(self): self.hub.files[SHA_ABC] = b"xyz" # размер тот же, содержимое другое out = self.run_cmd(key="AURORA-1", file=SHA_ABC, filename="a.txt", size=3) self.assertEqual(out["status"], "failed") self.assertTrue(out["retryable"]) self.assertEqual(out["peer"], "хабом") self.assertTrue(out["error"].startswith("хаб: "), out["error"]) self.assertNotIn("AURORA-1", self.fake.attachments) self.assertFalse(any(m == "POST" for m, _ in self.fake.requests)) def test_length_mismatch_is_retryable_hub_failure(self): self.hub.put_file(b"abc") out = self.run_cmd(key="AURORA-1", file=SHA_ABC, filename="a.txt", size=4) self.assertEqual(out["status"], "failed") self.assertTrue(out["retryable"]) self.assertEqual(out["peer"], "хабом") self.assertNotIn("AURORA-1", self.fake.attachments) def test_hub_non_json_file_no_content_in_error(self): def broken(sha): raise agent.HttpError(200, 'ответ не JSON: Unterminated string; начало: {"file": "x", "content": "U0VDUkVULUNPTlRFTlQ') self.ex.hub.file = broken out = self.run_cmd(key="AURORA-1", file=SHA_ABC, filename="a.txt", size=3) self.assertEqual(out["status"], "failed") self.assertTrue(out["retryable"]) self.assertEqual(out["peer"], "хабом") self.assertNotIn("U0VDUkVU", out["error"]) self.assertNotIn("начало", out["error"]) def test_hub_4xx_other_than_404_not_retryable(self): self.ex.hub = agent.HubClient({"hub_url": self.hub.base_url, "hub_token": "wrong-token", "timeout": 5}) out = self.run_cmd(key="AURORA-1", file=SHA_ABC, filename="a.txt", size=3) self.assertEqual(out["status"], "failed") self.assertTrue(out["error"].startswith("хаб: HTTP 401"), out["error"]) self.assertFalse(out["retryable"]) def test_jira_413_failed_with_code_and_text(self): self.fake.attachment_limit = 2 sha = self.hub.put_file(b"abc") out = self.run_cmd(key="AURORA-1", file=sha, filename="a.txt", size=3) self.assertEqual(out["status"], "failed") self.assertTrue(out["error"].startswith("HTTP 413"), out["error"]) self.assertIn("Размер вложения превышает предел", out["error"]) self.assertFalse(out["retryable"]) def test_upload_response_without_id(self): self.fake.bad_attachment_response = True sha = self.hub.put_file(b"abc") out = self.run_cmd(key="AURORA-1", file=sha, filename="a.txt", size=3) self.assertEqual(out["status"], "failed") self.assertEqual(out["error"], "Jira не вернула id вложения в ответе на загрузку") self.assertNotIn("retryable", out) if __name__ == "__main__": unittest.main()