import io import unittest from contextlib import redirect_stderr, redirect_stdout from tests import _common as c 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_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)) if __name__ == "__main__": unittest.main()