import io
import unittest
from contextlib import redirect_stderr, redirect_stdout

from tests import _common as c
from tests import fake_eva
from tests.fake_eva import FakeEva

agent = c.agent


class EvaClientTest(unittest.TestCase):
    def setUp(self):
        self.eva = FakeEva()
        self.addCleanup(self.eva.stop)
        self.cli = agent.EvaClient(c.eva_cfg(self.eva.base_url, timeout=10))
        self.sleeps = []
        self.cli.sleep = self.sleeps.append

    def test_call_ok(self):
        self.eva.add_doc(1, "DOC-1", "Корень", project="P")
        doc = self.cli.get_doc("DOC-1", ["code", "name"])
        self.assertEqual(doc, {"code": "DOC-1", "name": "Корень"})
        self.assertIsNone(self.cli.get_doc("DOC-404", ["code"]))
        req = self.eva.api_calls[0]
        self.assertEqual(req["path"], "/api/?m=CmfDocument.get")
        self.assertEqual(req["auth"], "Bearer eva-secret")
        self.assertEqual(req["body"]["jsonrpc"], "2.2")
        self.assertEqual(req["body"]["kwargs"]["filter"], [["code", "==", "DOC-1"]])
        self.assertEqual(self.cli.calls, 2)

    def test_retry_429_retry_after(self):
        self.eva.fail_next(429, 1, retry_after=7)
        self.assertEqual(self.cli.call("CmfDocument.list", filter=[], fields=["code"]), [])
        self.assertEqual(self.sleeps, [7])
        self.assertEqual(self.cli.rate_limited, 1)

    def test_retry_after_capped(self):
        self.eva.fail_next(429, 1, retry_after=600)
        self.cli.call("CmfDocument.list", filter=[], fields=["code"])
        self.assertEqual(self.sleeps, [60])

    def test_retry_503_then_ok(self):
        self.eva.fail_next(503, 2)
        self.cli.call("CmfDocument.list", filter=[], fields=["code"])
        self.assertEqual(self.sleeps, [1, 2])
        self.assertEqual(self.cli.rate_limited, 0)

    def test_gives_up_after_4(self):
        self.eva.fail_next(503, 10)
        with self.assertRaises(agent.HttpError) as cm:
            self.cli.call("CmfDocument.list", filter=[], fields=["code"])
        self.assertEqual(cm.exception.code, 503)
        self.assertEqual(len(self.eva.api_calls), 4)
        self.assertEqual(self.sleeps, [1, 2, 4])

    def test_no_retry_on_401(self):
        self.cli.headers["Authorization"] = "Bearer wrong"
        with self.assertRaises(agent.HttpError):
            self.cli.call("CmfDocument.list", filter=[], fields=["code"])
        self.assertEqual(len(self.eva.api_calls), 1)

    def test_rpc_error(self):
        self.eva.rpc_error_next("нет прав")
        with self.assertRaises(agent.EvaError) as cm:
            self.cli.call("CmfDocument.get", filter=[], fields=["code"])
        self.assertEqual(str(cm.exception), "CmfDocument.get: нет прав")

    def test_list_pages_200(self):
        for i in range(450):
            self.eva.add_doc(i + 1, "DOC-%d" % (i + 1), project="P")
        docs = self.cli.list_docs([["project.code", "==", "P"]], ["code"])
        self.assertEqual(len(docs), 450)
        self.assertEqual(len(self.eva.api_calls), 3)
        self.assertEqual([r["body"]["kwargs"]["slice"] for r in self.eva.api_calls],
                         [[0, 200], [200, 400], [400, 600]])

    def test_list_slice_ignored_returns_all(self):
        """M-4: Eva не поддерживает slice и отдаёт весь список каждый раз — второй ответ без новых записей, стоп."""
        orig = self.eva.rpc
        self.eva.rpc = lambda m, kw: orig(m, dict(kw, slice=None))
        for i in range(450):
            self.eva.add_doc(i + 1, "DOC-%d" % (i + 1), project="P")
        out = io.StringIO()
        with redirect_stdout(out):
            docs = self.cli.list_docs([["project.code", "==", "P"]], ["id", "code"])
        self.assertEqual(len(docs), 450)
        self.assertEqual(len(self.eva.api_calls), 2)
        self.assertIn("slice не соблюдается", out.getvalue())
        self.assertIn("CmfDocument.list повторяет записи", out.getvalue())

    def test_list_slice_ignored_same_page(self):
        """M-4: Eva всегда отдаёт первые 200 — без защиты цикл бесконечен."""
        orig = self.eva.rpc
        self.eva.rpc = lambda m, kw: orig(m, dict(kw, slice=[0, 200]))
        for i in range(450):
            self.eva.add_doc(i + 1, "DOC-%d" % (i + 1), project="P")
        with redirect_stdout(io.StringIO()):
            docs = self.cli.list_docs([["project.code", "==", "P"]], ["id", "code"])
        self.assertEqual([d["code"] for d in docs], ["DOC-%d" % (i + 1) for i in range(200)])
        self.assertEqual(len(self.eva.api_calls), 2)

    def test_list_unwraps_dict(self):
        orig = self.eva.rpc
        self.eva.rpc = lambda m, kw: {"items": orig(m, kw)}
        self.eva.add_doc(1, "DOC-1")
        self.assertEqual(self.cli.list_docs([], ["code"]), [{"code": "DOC-1"}])

    def test_tree_parent_filter(self):
        self.eva.add_doc(1, "DOC-1")
        self.eva.add_doc(2, "DOC-2", tree_parent="DOC-1")
        self.eva.add_doc(3, "DOC-3", tree_parent="DOC-2")
        got = self.cli.list_docs([["tree_parent.code", "==", "DOC-1"]], ["code"])
        self.assertEqual(got, [{"code": "DOC-2"}])

    def test_pacing_between_calls(self):
        self.cli.pacing = 0.5
        self.cli.call("CmfDocument.list", filter=[], fields=["code"])
        self.assertEqual(self.sleeps, [])              # перед первым вызовом паузы нет
        self.cli.call("CmfDocument.list", filter=[], fields=["code"])
        self.assertEqual(len(self.sleeps), 1)
        self.assertTrue(0 < self.sleeps[0] <= 0.5)

    def test_download_ok(self):
        self.eva.add_file("/files/a.bin", b"abc")
        self.assertEqual(self.cli.download(self.eva.base_url + "/files/a.bin", 10), b"abc")
        self.assertEqual(self.eva.requests[-1]["auth"], "Bearer eva-secret")

    def test_download_foreign_host(self):
        with self.assertRaises(agent.ForeignHost):
            self.cli.download("http://example.invalid/files/a.bin", 10)
        self.assertEqual(self.eva.requests, [])

    def test_download_lookalikes(self):
        """T5.2: похожий хост, userinfo, другой порт и схема — ForeignHost до запроса (адрес Eva — https://eva.test)."""
        cli = agent.EvaClient(c.eva_cfg("https://eva.test"))
        cli.fetch_bytes = lambda *a, **kw: self.fail("запроса быть не должно")
        for url in ("https://eva.test.evil/files/a.bin", "https://eva.test@evil/files/a.bin",
                    "https://eva.test:8443/files/a.bin", "http://eva.test/files/a.bin",
                    "https://eva.testx/files/a.bin", "https://user@eva.test/files/a.bin"):
            with self.subTest(url=url):
                with self.assertRaises(agent.ForeignHost) as cm:
                    cli.download(url, 10)
                self.assertEqual(str(cm.exception), "чужой хост")
        with self.assertRaises(agent.ForeignHost) as cm:
            cli.download("https://eva.test:abc/files/a.bin", 10)
        self.assertEqual(str(cm.exception), "некорректный адрес")

    def test_download_limit(self):
        limit = 10 * 1024 * 1024
        self.eva.add_file("/files/big.bin", b"x" * (limit + 1), length=False)
        self.eva.add_file("/files/ok.bin", b"x" * limit, length=False)
        with self.assertRaises(agent.TooBig):
            self.cli.download(self.eva.base_url + "/files/big.bin", limit)
        self.assertEqual(len(self.cli.download(self.eva.base_url + "/files/ok.bin", limit)), limit)

    def test_download_404(self):
        with self.assertRaises(agent.HttpError) as cm:
            self.cli.download(self.eva.base_url + "/files/none", 10)
        self.assertEqual(cm.exception.code, 404)

    def test_token_not_logged(self):
        out, err = io.StringIO(), io.StringIO()
        texts = []
        with redirect_stdout(out), redirect_stderr(err):
            self.eva.fail_next(503, 8)
            for call in (lambda: self.cli.call("CmfDocument.list", filter=[], fields=["code"]),
                         lambda: self.cli.download(self.eva.base_url + "/files/none", 10)):
                try:
                    call()
                except Exception as e:  # noqa: BLE001
                    texts.append(str(e))
            self.eva.rpc_error_next("сбой eva-secret")
            try:
                self.cli.call("CmfDocument.get", filter=[], fields=["code"])
            except agent.EvaError as e:
                texts.append(str(e))
            dead = agent.EvaClient(c.eva_cfg("http://127.0.0.1:1"))
            dead.sleep = lambda s: None
            try:
                dead.call("CmfDocument.get", filter=[], fields=["code"])
            except ConnectionError as e:
                texts.append(str(e))
        self.assertEqual(len(texts), 4)
        for t in texts + [out.getvalue(), err.getvalue()]:
            self.assertNotIn("eva-secret", t)
        self.assertNotIn("eva-secret", repr(self.cli))

    def test_list_versions_filter_and_fields(self):
        self.eva.add_version("CmfDocument:1", "DOCH-1")
        self.eva.add_version("CmfDocument:1", "DOCH-2", has_published=False)
        self.eva.add_version("CmfDocument:2", "DOCH-9")
        got = self.cli.list_versions("CmfDocument:1")
        self.assertEqual([v["code"] for v in got], ["DOCH-1", "DOCH-2"])
        self.assertEqual(got[0], {"code": "DOCH-1", "cmf_created_at": "2026-10-27T17:59:00.123456+03:00",
                                  "has_published": True, "cmf_author": fake_eva.AUTHOR})
        req = self.eva.api_calls[0]
        self.assertEqual(req["path"], "/api/?m=CmfDocumentHistory.list")
        self.assertEqual(req["body"]["kwargs"]["filter"], [["parent_id", "==", "CmfDocument:1"]])
        self.assertEqual(req["body"]["kwargs"]["fields"], ["code", "cmf_created_at", "has_published", "cmf_author"])
        self.assertEqual(req["body"]["kwargs"]["slice"], [0, 200])

    def test_list_versions_pages_200(self):
        for i in range(450):
            self.eva.add_version("CmfDocument:1", "DOCH-%d" % i)
        self.assertEqual(len(self.cli.list_versions("CmfDocument:1")), 450)
        self.assertEqual([r["body"]["kwargs"]["slice"] for r in self.eva.api_calls], [[0, 200], [200, 400], [400, 600]])

    def test_list_versions_slice_ignored(self):
        orig = self.eva.rpc
        self.eva.rpc = lambda m, kw: orig(m, dict(kw, slice=None))
        for i in range(450):
            self.eva.add_version("CmfDocument:1", "DOCH-%d" % i)
        out = io.StringIO()
        with redirect_stdout(out):
            got = self.cli.list_versions("CmfDocument:1")
        self.assertEqual(len(got), 450)
        self.assertEqual(len(self.eva.api_calls), 2)
        self.assertIn("CmfDocumentHistory.list повторяет записи", out.getvalue())

    def test_get_version(self):
        self.eva.add_version("CmfDocument:1", "DOCH-1", text="<p>v1</p>")
        self.assertEqual(self.cli.get_version("DOCH-1"), {"code": "DOCH-1", "text": "<p>v1</p>"})
        self.assertIsNone(self.cli.get_version("DOCH-404"))
        req = self.eva.api_calls[0]
        self.assertEqual(req["path"], "/api/?m=CmfDocumentHistory.get")
        self.assertEqual(req["body"]["kwargs"]["filter"], [["code", "==", "DOCH-1"]])
        self.assertEqual(req["body"]["kwargs"]["fields"], ["code", "text"])
        self.assertEqual(self.cli.calls, 2)

    def test_rpc_error_on_method(self):
        self.eva.add_doc(1, "DOC-1", project="P")
        self.eva.rpc_error_on("CmfDocumentHistory.list", "нет прав")
        self.assertIsNotNone(self.cli.get_doc("DOC-1", ["code"]))
        with self.assertRaises(agent.EvaError) as cm:
            self.cli.list_versions("1")
        self.assertEqual(str(cm.exception), "CmfDocumentHistory.list: нет прав")
        self.assertEqual(self.cli.list_versions("1"), [])     # ошибка разовая


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