142 lines
7.2 KiB
Python
142 lines
7.2 KiB
Python
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()
|