jira-agent/tests/test_clients.py

import unittest
import urllib.parse

from _common import agent, jira_cfg, no_sleep
from fake_jira import FakeJira


class ClientsTest(unittest.TestCase):
    def setUp(self):
        self.fake = FakeJira()
        self.jira = agent.JiraClient(jira_cfg(self.fake))
        self.jira.sleep = no_sleep

    def tearDown(self):
        self.fake.stop()

    def test_get_and_counters(self):
        self.fake.add("AURORA-1")
        data = self.jira.get("/rest/api/2/issue/AURORA-1", fields="summary")
        self.assertEqual(data["fields"]["summary"], "Тема AURORA-1")
        self.assertEqual(self.jira.calls, 1)
        self.assertIsNotNone(self.jira.last_call_at)
        self.assertEqual(self.fake.requests[-1], ("GET", "/rest/api/2/issue/AURORA-1"))

    def test_multipart_file_bytes(self):
        body = agent.multipart_file("b0", "Отчёт.txt", b"x\r\ny")
        q = urllib.parse.quote("Отчёт.txt", safe="")
        expected = ('--b0\r\nContent-Disposition: form-data; name="file"; filename="Отчёт.txt"; '
                    "filename*=UTF-8''%s\r\nContent-Type: application/octet-stream\r\n\r\n" % q).encode("utf-8") \
            + b"x\r\ny" + b"\r\n--b0--\r\n"
        self.assertEqual(body, expected)

    def test_wire_filename(self):
        self.assertEqual(agent.wire_filename('а "б"\r\n.txt'), "а %22б%22%0D%0A.txt")
        self.assertEqual(agent.wire_filename("ОСВ 60.xlsx"), "ОСВ 60.xlsx")

    def test_upload_cyrillic_name(self):
        self.fake.add("AURORA-1")
        name = "ОСВ сч. 60 за 2 кв. 2026.xlsx"
        data = bytes(range(256)) * 40
        resp = self.jira.upload("AURORA-1", name, data)
        self.assertIsInstance(resp, list)
        a = self.fake.attachments["AURORA-1"][0]
        self.assertEqual((resp[0]["id"], a["filename"], a["size"], a["data"]), (a["id"], name, len(data), data))
        up = self.fake.uploads[-1]
        self.assertEqual(up["token"], "no-check")
        self.assertTrue(up["content_type"].startswith("multipart/form-data; boundary="))
        self.assertEqual(len(up["content_type"].split("boundary=")[1]), 32)        # uuid4().hex
        self.assertEqual(up["headers"]["content-type"], "application/octet-stream")
        self.assertEqual(up["disposition"]["filename*"], name)
        self.assertEqual(self.jira.calls, 1)
        self.assertEqual(self.fake.requests[-1], ("POST", "/rest/api/2/issue/AURORA-1/attachments"))

    def test_upload_filename_with_quote_and_newline(self):
        self.fake.add("AURORA-1")
        name = 'итог "кв.2"\r\n.txt'
        self.jira.upload("AURORA-1", name, b"abc")
        up = self.fake.uploads[-1]
        self.assertEqual(up["disposition"]["filename"], "итог %22кв.2%22%0D%0A.txt")
        self.assertEqual(up["disposition"]["filename*"], name)
        self.assertEqual(len(self.fake.attachments["AURORA-1"]), 1)
        self.assertEqual(self.fake.attachments["AURORA-1"][0]["data"], b"abc")

    def test_upload_needs_xsrf_header(self):
        self.fake.add("AURORA-1")
        self.jira.headers = {k: v for k, v in self.jira.headers.items() if k != "X-Atlassian-Token"}
        with self.assertRaises(agent.HttpError) as cm:
            self.jira.upload("AURORA-1", "a.txt", b"abc")
        self.assertEqual(cm.exception.code, 403)
        self.assertNotIn("AURORA-1", self.fake.attachments)

    def test_upload_retries_429(self):
        self.fake.add("AURORA-1")
        self.fake.fail_next.append((429, {"Retry-After": "0"}))
        self.jira.upload("AURORA-1", "a.txt", b"abc")
        self.assertEqual(len(self.fake.attachments["AURORA-1"]), 1)
        self.assertEqual(self.jira.rate_limited, 1)
        self.assertEqual(self.jira.calls, 2)

    def test_upload_timeout_at_least_120(self):
        self.fake.add("AURORA-1")
        seen = []
        orig = self.jira.send

        def spy(*a, **kw):
            seen.append(kw.get("timeout"))
            return orig(*a, **kw)

        self.jira.send = spy
        self.jira.upload("AURORA-1", "a.txt", b"abc")
        self.assertEqual(seen, [120])                         # timeout в конфиге тестов — 5 с

    def _content_path(self, a):
        return "/secure/attachment/%s/%s" % (a["id"], urllib.parse.quote(a["filename"], safe=""))

    def test_download_bytes(self):
        self.fake.add("AURORA-1")
        a = self.fake.add_attachment("AURORA-1", "скрин 1.png", b"\x89PNG\x00\x01", mime="image/png")
        data = self.jira.download(self._content_path(a), 100)
        self.assertEqual(data, b"\x89PNG\x00\x01")
        self.assertEqual(self.jira.calls, 1)
        self.assertTrue(self.fake.last_headers.get("authorization", "").startswith("Basic "))
        self.assertEqual(self.fake.content_gets, 1)

    def test_download_redirect_same_host(self):
        self.fake.add("AURORA-1")
        a = self.fake.add_attachment("AURORA-1", "a.bin", b"abc")
        self.fake.redirect_content = "%s/secure/moved/%s/a.bin" % (self.fake.base_url, a["id"])
        self.assertEqual(self.jira.download(self._content_path(a), 100), b"abc")

    def test_download_redirect_other_host(self):
        self.fake.add("AURORA-1")
        a = self.fake.add_attachment("AURORA-1", "a.bin", b"abc")
        port = self.fake.server.server_port
        self.fake.redirect_content = "http://localhost:%d/secure/moved/%s/a.bin" % (port, a["id"])
        with self.assertRaises(agent.RedirectAway) as e:
            self.jira.download(self._content_path(a), 100)
        self.assertEqual(e.exception.host, "localhost")
        self.assertEqual(self.fake.content_gets, 0)
        self.assertFalse(any(r[1].startswith("/secure/moved/") for r in self.fake.requests))

    def test_download_limit(self):
        self.fake.add("AURORA-1")
        a = self.fake.add_attachment("AURORA-1", "a.bin", b"x" * 20)
        with self.assertRaises(agent.TooBig):
            self.jira.download(self._content_path(a), 10)
        self.assertEqual(self.jira.download(self._content_path(a), 20), b"x" * 20)

    def test_download_429_retry(self):
        self.fake.add("AURORA-1")
        a = self.fake.add_attachment("AURORA-1", "a.bin", b"abc")
        self.fake.fail_next.append((429, {"Retry-After": "0"}))
        self.assertEqual(self.jira.download(self._content_path(a), 100), b"abc")
        self.assertEqual(self.jira.rate_limited, 1)

    def test_download_cut_body_is_connection_error(self):       # M-1 финального ревью
        self.fake.add("AURORA-1")
        a = self.fake.add_attachment("AURORA-1", "a.bin", b"x" * 40)
        self.fake.lying_length.add(a["id"])
        with self.assertRaises(ConnectionError):
            self.jira.download(self._content_path(a), 1000)

    def test_json_call_redirect_other_host_refused(self):      # M-4 финального ревью
        self.fake.add("AURORA-1")
        port = self.fake.server.server_port
        self.fake.redirects["/rest/api/2/issue/AURORA-1"] = "http://localhost:%d/rest/api/2/issue/AURORA-2" % port
        with self.assertRaises(agent.HttpError) as e:
            self.jira.get("/rest/api/2/issue/AURORA-1")
        self.assertIn("localhost", e.exception.text)
        self.assertFalse(any(r[1] == "/rest/api/2/issue/AURORA-2" for r in self.fake.requests))

    def test_json_call_redirect_same_host_followed(self):
        self.fake.add("AURORA-2")
        self.fake.redirects["/rest/api/2/issue/AURORA-1"] = self.fake.base_url + "/rest/api/2/issue/AURORA-2"
        self.assertEqual(self.jira.get("/rest/api/2/issue/AURORA-1")["key"], "AURORA-2")

    def test_retry_on_429_with_retry_after(self):
        self.fake.add("AURORA-1")
        self.fake.fail_next.append((429, {"Retry-After": "0"}))
        self.fake.fail_next.append((503, {}))
        data = self.jira.get("/rest/api/2/issue/AURORA-1")
        self.assertEqual(data["key"], "AURORA-1")
        self.assertEqual(self.jira.rate_limited, 2)
        self.assertEqual(self.jira.calls, 3)   # считаются все полученные ответы, включая 429/503

    def test_http_error_not_retried(self):
        with self.assertRaises(agent.HttpError) as cm:
            self.jira.get("/rest/api/2/issue/NOPE-1")
        self.assertEqual(cm.exception.code, 404)
        self.assertIn("Does Not Exist", cm.exception.text)
        self.assertEqual(self.jira.calls, 1)
        self.assertIsNotNone(self.jira.last_call_at)

    def test_gives_up_after_five(self):
        self.fake.add("AURORA-1")
        self.fake.fail_next.extend([(429, {"Retry-After": "0"})] * 5)
        with self.assertRaises(agent.HttpError):
            self.jira.get("/rest/api/2/issue/AURORA-1")
        self.assertEqual(self.jira.rate_limited, 5)
        self.assertEqual(self.jira.calls, 5)

    def test_basic_auth_header_and_no_secret_in_repr(self):
        self.assertTrue(self.jira.headers["Authorization"].startswith("Basic "))
        self.assertNotIn("secret", repr(self.jira))

    def test_retry_after_lowercase_header(self):
        waits = []
        self.jira.sleep = waits.append
        self.fake.add("AURORA-1")
        self.fake.fail_next.append((429, {"retry-after": "7"}))
        self.jira.get("/rest/api/2/issue/AURORA-1")
        self.assertEqual(waits, [7])

    def test_retry_after_not_a_number(self):
        waits = []
        self.jira.sleep = waits.append
        self.fake.add("AURORA-1")
        self.fake.fail_next.append((503, {"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"}))
        self.jira.get("/rest/api/2/issue/AURORA-1")
        self.assertEqual(waits, [60])

    def test_retry_after_negative(self):
        waits = []
        self.jira.sleep = waits.append
        self.fake.add("AURORA-1")
        self.fake.fail_next.append((429, {"Retry-After": "-5"}))
        self.jira.get("/rest/api/2/issue/AURORA-1")
        self.assertEqual(waits, [0])

    def test_error_body_read_failure_is_connection_error(self):
        import io
        import urllib.error
        from unittest import mock

        class Broken(io.BytesIO):
            def read(self, *a):
                raise TimeoutError("timed out")

        err = urllib.error.HTTPError("http://x", 500, "boom", {}, Broken())
        with mock.patch("urllib.request.OpenerDirector.open", side_effect=err):
            with self.assertRaises(ConnectionError):
                self.jira.get("/x")
        self.assertEqual(self.jira.calls, 1)

    def test_incomplete_read_is_connection_error(self):
        import http.client
        from unittest import mock
        with mock.patch("urllib.request.OpenerDirector.open", side_effect=http.client.IncompleteRead(b"ab", 10)):
            with self.assertRaises(ConnectionError):
                self.jira.get("/x")

    def test_incomplete_read_inside_response_body(self):
        import socket
        import threading
        srv = socket.socket()
        srv.bind(("127.0.0.1", 0))
        srv.listen(1)

        def serve():
            c, _ = srv.accept()
            c.recv(4096)
            c.sendall(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 100\r\n\r\n{\"a\":")
            c.close()

        t = threading.Thread(target=serve, daemon=True)
        t.start()
        cfg = jira_cfg(self.fake)
        cfg["jira"]["base_url"] = "http://127.0.0.1:%d" % srv.getsockname()[1]
        try:
            with self.assertRaises(ConnectionError):
                agent.JiraClient(cfg).get("/x")
        finally:
            t.join(2)
            srv.close()

    def test_bearer_header(self):
        cfg = jira_cfg(self.fake)
        cfg["jira"]["auth"] = {"type": "bearer", "token": "tok123"}
        self.assertEqual(agent.JiraClient(cfg).headers["Authorization"], "Bearer tok123")

    def test_unreachable_jira_raises_connection_error(self):
        cfg = jira_cfg(self.fake)
        cfg["jira"]["base_url"] = "http://127.0.0.1:1"
        with self.assertRaises(ConnectionError):
            agent.JiraClient(cfg).get("/rest/api/2/myself")

    def test_non_json_200_becomes_http_error(self):
        import threading
        from http.server import BaseHTTPRequestHandler, HTTPServer

        class H(BaseHTTPRequestHandler):
            def log_message(self, *a):
                pass

            def do_GET(self):
                body = b"<html>\xff\xfe not json"
                self.send_response(200)
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

        srv = HTTPServer(("127.0.0.1", 0), H)
        threading.Thread(target=srv.serve_forever, daemon=True).start()
        try:
            cfg = jira_cfg(self.fake)
            cfg["jira"]["base_url"] = "http://127.0.0.1:%d" % srv.server_port
            jira = agent.JiraClient(cfg)
            with self.assertRaises(agent.HttpError) as cm:
                jira.get("/x")
            self.assertEqual(cm.exception.code, 200)
            self.assertEqual(jira.calls, 1)
        finally:
            srv.shutdown()
            srv.server_close()

    def test_secret_not_in_log_output(self):
        import io
        from contextlib import redirect_stdout
        buf = io.StringIO()
        self.fake.add("AURORA-1")
        self.fake.fail_next.append((429, {"Retry-After": "0"}))
        with redirect_stdout(buf):
            self.jira.get("/rest/api/2/issue/AURORA-1")
            agent.log("проверка")
        out = buf.getvalue()
        self.assertNotIn("secret", out)
        self.assertNotIn(self.jira.headers["Authorization"], out)

    def test_load_config_requires_fields(self):
        import json
        import os
        import tempfile
        with tempfile.TemporaryDirectory() as d:
            path = os.path.join(d, "agent.json")
            with open(path, "w", encoding="utf-8") as f:
                json.dump({"hub_url": "http://x", "hub_token": "t"}, f)
            with self.assertRaises(SystemExit):
                agent.load_config(path)
            with open(path, "w", encoding="utf-8") as f:
                json.dump(jira_cfg(self.fake), f)
            cfg = agent.load_config(path)
            self.assertEqual(cfg["poll_seconds"], 0)

    def test_check_command(self):
        import json
        import os
        import tempfile
        with tempfile.TemporaryDirectory() as d:
            path = os.path.join(d, "agent.json")
            with open(path, "w", encoding="utf-8") as f:
                json.dump(jira_cfg(self.fake), f)
            # хаб недоступен (порт 1), Jira доступен — check печатает оба результата и возвращает 1
            self.assertEqual(agent.main(["--config", path, "check"]), 1)


    def test_selftest_command(self):
        import contextlib
        import io
        import json
        import os
        import tempfile
        with tempfile.TemporaryDirectory() as d:
            path = os.path.join(d, "agent.json")
            with open(path, "w", encoding="utf-8") as f:
                json.dump(jira_cfg(self.fake), f)
            out = io.StringIO()
            with contextlib.redirect_stdout(out):
                # хаб недоступен (порт 1), Jira доступен — selftest печатает версию и возвращает 3, а не 1
                self.assertEqual(agent.main(["--config", path, "selftest"]), 3)
        self.assertIn("jira-agent v%d" % agent.AGENT_VERSION, out.getvalue())
        self.assertIn("Jira: доступен", out.getvalue())

if __name__ == "__main__":
    unittest.main()