3a5bd9a109
По финальному ревью: M-1, M-4, M-6. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
254 lines
13 KiB
Python
254 lines
13 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()
|
||
|
||
|
||
class FetchAttachmentTest(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-1650")
|
||
|
||
def tearDown(self):
|
||
self.fake.stop()
|
||
self.hub.stop()
|
||
|
||
def fetch(self, a, key="AURORA-1650"):
|
||
return self.ex.run({"id": 7, "type": "fetch_attachment", "params": {"key": key, "attachment_id": a["id"]}})
|
||
|
||
def test_fetch_screenshot(self): # сценарий «Скачивание скриншота»
|
||
data = bytes(range(256)) * 900
|
||
a = self.fake.add_attachment("AURORA-1650", "скрин 1.png", data, mime="image/png")
|
||
out = self.fetch(a)
|
||
self.assertEqual(out["status"], "done", out)
|
||
self.assertEqual(out["result"], {"attachment_id": a["id"], "file": hashlib.sha256(data).hexdigest(),
|
||
"size": len(data)})
|
||
self.assertEqual(self.hub.uploaded, [data])
|
||
self.assertEqual(self.fake.requests[0], ("GET", "/rest/api/2/issue/AURORA-1650"))
|
||
self.assertEqual(self.fake.queries[0].get("fields"), "attachment")
|
||
self.assertEqual(self.fake.requests[1][1],
|
||
"/secure/attachment/%s/%%D1%%81%%D0%%BA%%D1%%80%%D0%%B8%%D0%%BD%%201.png" % a["id"])
|
||
self.assertEqual(self.fake.writes(), [])
|
||
|
||
def test_fetch_deleted(self):
|
||
a = self.fake.add_attachment("AURORA-1650", "a.png", b"abc")
|
||
self.fake.attachments["AURORA-1650"] = []
|
||
out = self.fetch(a)
|
||
self.assertEqual((out["status"], out["error"]),
|
||
("failed", "вложение #%s удалено из AURORA-1650 в Jira" % a["id"]))
|
||
self.assertEqual(self.fake.content_gets, 0)
|
||
|
||
def test_fetch_too_large_known(self):
|
||
a = self.fake.add_attachment("AURORA-1650", "запись.mp4", b"x")
|
||
a["size"] = 25000000
|
||
out = self.fetch(a)
|
||
self.assertEqual((out["status"], out["error"]),
|
||
("failed", "файл больше 10 МБ (23,8 МБ) — скачайте в Jira"))
|
||
self.assertEqual(self.fake.content_gets, 0)
|
||
self.assertEqual(self.hub.uploaded, [])
|
||
|
||
def test_fetch_too_large_unknown(self):
|
||
a = self.fake.add_attachment("AURORA-1650", "big.bin", b"x" * (12 * 1024 * 1024))
|
||
self.fake.hide_size.add(a["id"])
|
||
out = self.fetch(a)
|
||
self.assertEqual((out["status"], out["error"]),
|
||
("failed", "файл больше 10 МБ (больше 10,0 МБ) — скачайте в Jira"))
|
||
self.assertEqual(self.hub.uploaded, [])
|
||
|
||
def test_fetch_unknown_size_ok(self):
|
||
a = self.fake.add_attachment("AURORA-1650", "a.json", b'{"a": 1}', mime="application/json")
|
||
self.fake.hide_size.add(a["id"])
|
||
out = self.fetch(a)
|
||
self.assertEqual(out["status"], "done", out)
|
||
self.assertEqual(out["result"]["size"], 8)
|
||
|
||
def test_fetch_redirect_away(self):
|
||
a = self.fake.add_attachment("AURORA-1650", "a.png", b"abc")
|
||
self.fake.redirect_content = "http://localhost:%d/secure/moved/%s/a.png" % (self.fake.server.server_port, a["id"])
|
||
out = self.fetch(a)
|
||
self.assertEqual((out["status"], out["error"]), ("failed", "Jira перенаправила скачивание на localhost"))
|
||
self.assertEqual(self.hub.uploaded, [])
|
||
|
||
def test_fetch_truncated(self):
|
||
a = self.fake.add_attachment("AURORA-1650", "a.bin", b"abcdef")
|
||
self.fake.short_content.add(a["id"])
|
||
out = self.fetch(a)
|
||
self.assertEqual((out["status"], out["error"]), ("failed", "файл пришёл не целиком (5 из 6)"))
|
||
self.assertEqual(self.hub.uploaded, [])
|
||
|
||
def test_fetch_empty(self):
|
||
a = self.fake.add_attachment("AURORA-1650", "empty.txt", b"")
|
||
out = self.fetch(a)
|
||
self.assertEqual(out["status"], "done", out)
|
||
self.assertEqual(out["result"]["size"], 0)
|
||
self.assertEqual(self.hub.uploaded, [b""])
|
||
|
||
def test_fetch_hub_down(self):
|
||
a = self.fake.add_attachment("AURORA-1650", "a.png", b"abc")
|
||
self.hub.reject_files = "drop"
|
||
out = self.fetch(a)
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertTrue(out["retryable"])
|
||
self.assertEqual(out["peer"], "хабом")
|
||
self.assertTrue(out["error"].startswith("хаб: "))
|
||
|
||
def test_fetch_hub_422(self):
|
||
a = self.fake.add_attachment("AURORA-1650", "a.png", b"abc")
|
||
self.hub.reject_files = (422, {"error": "bad_request", "detail": "content — не base64"})
|
||
out = self.fetch(a)
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertFalse(out["retryable"])
|
||
self.assertTrue(out["error"].startswith("хаб: "), out["error"])
|
||
|
||
def test_fetch_unknown_size_cut_body_retryable(self): # M-1 финального ревью
|
||
a = self.fake.add_attachment("AURORA-1650", "a.json", b"x" * 40)
|
||
self.fake.hide_size.add(a["id"])
|
||
self.fake.lying_length.add(a["id"])
|
||
out = self.fetch(a)
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertTrue(out["retryable"])
|
||
self.assertEqual(self.hub.uploaded, [])
|