140 lines
6.2 KiB
Python
140 lines
6.2 KiB
Python
import hashlib
|
||
import io
|
||
import os
|
||
import unittest
|
||
from contextlib import redirect_stdout
|
||
|
||
from _common import SETTINGS, agent, eva_cfg, no_sleep
|
||
from fake_eva import FakeEva
|
||
from fake_hub import FakeHub
|
||
|
||
MB = 1024 * 1024
|
||
TOO_BIG = "файл больше 10 МБ — откройте в Eva"
|
||
|
||
|
||
class WikiFetchTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.eva = FakeEva()
|
||
self.hub = FakeHub(SETTINGS)
|
||
self.addCleanup(self.eva.stop)
|
||
self.addCleanup(self.hub.stop)
|
||
cfg = eva_cfg(self.eva.base_url, hub_url=self.hub.base_url, hub_token=self.hub.token)
|
||
client = agent.EvaClient(cfg)
|
||
client.sleep = no_sleep
|
||
self.ex = agent.Executor(client, agent.HubClient(cfg))
|
||
|
||
def fetch(self, url):
|
||
return self.ex.run({"id": 1, "type": "wiki_fetch", "params": {"attachment_id": 7, "url": url}})
|
||
|
||
def file_url(self, path="/files/a.bin"):
|
||
return self.eva.base_url + path
|
||
|
||
def test_fetch_ok(self):
|
||
data = os.urandom(300 * 1024)
|
||
self.eva.add_file("/files/a.bin", data)
|
||
out = self.fetch(self.file_url())
|
||
self.assertEqual(out["status"], "done", out)
|
||
self.assertEqual(out["result"]["size"], 307200)
|
||
self.assertEqual(out["result"]["file"], hashlib.sha256(data).hexdigest())
|
||
self.assertEqual(self.hub.files[out["result"]["file"]], data)
|
||
self.assertEqual(self.eva.requests[0]["auth"], "Bearer " + self.eva.token)
|
||
|
||
def test_fetch_foreign_host(self):
|
||
out = self.fetch("http://127.0.0.1:1/files/a.bin")
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertIn("чужой хост", out["error"])
|
||
self.assertEqual(self.eva.requests, [])
|
||
self.assertEqual(self.hub.uploaded, [])
|
||
|
||
def test_fetch_foreign_lookalikes(self):
|
||
"""T5.2: похожий хост, userinfo, другой порт и схема — «чужой хост», запроса к Eva нет."""
|
||
port = self.eva.base_url.rsplit(":", 1)[1] # http://127.0.0.1:<port>
|
||
for url in ("http://127.0.0.1.evil:%s/files/a.bin" % port,
|
||
"http://127.0.0.1:%s@evil.test/files/a.bin" % port,
|
||
"http://127.0.0.1:%d/files/a.bin" % (int(port) + 1),
|
||
"https://127.0.0.1:%s/files/a.bin" % port):
|
||
with self.subTest(url=url):
|
||
out = self.fetch(url)
|
||
self.assertEqual((out["status"], out["error"]), ("failed", "чужой хост"))
|
||
self.assertEqual(self.eva.requests, [])
|
||
self.assertEqual(self.hub.uploaded, [])
|
||
|
||
def test_fetch_malformed_url(self):
|
||
out = self.fetch("http://[::1/files/a.bin")
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertIn("некорректный адрес", out["error"])
|
||
self.assertEqual(self.eva.requests, [])
|
||
|
||
def test_fetch_non_ascii_and_spaces(self):
|
||
"""I-2: адрес из HTML с пробелами и кириллицей кодируется перед запросом; %XX не кодируется повторно."""
|
||
enc = "/files/obj/x/%D0%9F%D0%9C%D0%9F%20%D1%84%D0%B0%D0%B9%D0%BB.png"
|
||
self.eva.add_file(enc, b"png")
|
||
for path in ("/files/obj/x/ПМП файл.png", enc, "/files/obj/x/ПМП%20файл.png"):
|
||
with self.subTest(path=path), redirect_stdout(io.StringIO()):
|
||
out = self.fetch(self.file_url(path))
|
||
self.assertEqual(out["status"], "done", out)
|
||
self.assertEqual(out["result"]["size"], 3)
|
||
self.assertEqual([r["path"] for r in self.eva.requests], [enc] * 3)
|
||
|
||
def test_fetch_stray_percent(self):
|
||
self.eva.add_file("/files/100%25.txt", b"ok")
|
||
with redirect_stdout(io.StringIO()):
|
||
out = self.fetch(self.file_url("/files/100%.txt"))
|
||
self.assertEqual(out["status"], "done", out)
|
||
|
||
def test_value_error_inside_is_not_foreign_host(self):
|
||
"""I-2: ValueError внутри скачивания (например, UnicodeEncodeError) не выдаётся за «чужой хост»."""
|
||
def boom(*a, **kw):
|
||
raise UnicodeEncodeError("ascii", "я", 0, 1, "ordinal not in range")
|
||
self.ex.eva.fetch_bytes = boom
|
||
with self.assertRaises(UnicodeEncodeError):
|
||
self.fetch(self.file_url())
|
||
|
||
def test_fetch_too_big_header(self):
|
||
self.eva.add_file("/files/a.bin", b"x" * 1024, length=11 * MB) # заявлено 11 МБ, тело не дойдёт
|
||
out = self.fetch(self.file_url())
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertEqual(out["error"], TOO_BIG)
|
||
self.assertEqual(self.hub.uploaded, [])
|
||
|
||
def test_fetch_too_big_streaming(self):
|
||
self.eva.add_file("/files/a.bin", b"x" * (10 * MB + 1), length=False)
|
||
out = self.fetch(self.file_url())
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertEqual(out["error"], TOO_BIG)
|
||
self.assertEqual(self.hub.uploaded, [])
|
||
|
||
def test_fetch_exactly_limit(self):
|
||
self.eva.add_file("/files/a.bin", b"x" * (10 * MB))
|
||
out = self.fetch(self.file_url())
|
||
self.assertEqual(out["status"], "done", out)
|
||
self.assertEqual(out["result"]["size"], 10 * MB)
|
||
|
||
def test_fetch_404(self):
|
||
out = self.fetch(self.file_url("/files/none.bin"))
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertEqual(out["error"], "файл не найден в Eva")
|
||
|
||
def test_fetch_empty_file(self):
|
||
self.eva.add_file("/files/a.bin", b"")
|
||
out = self.fetch(self.file_url())
|
||
self.assertEqual(out["status"], "done", out)
|
||
self.assertEqual(out["result"]["size"], 0)
|
||
self.assertEqual(self.hub.files[out["result"]["file"]], b"")
|
||
|
||
def test_fetch_hub_error_is_hub_error(self):
|
||
self.eva.add_file("/files/a.bin", b"abc")
|
||
self.hub.reject_files = (500, {"error": "boom"})
|
||
out = self.fetch(self.file_url())
|
||
self.assertEqual(out["status"], "failed")
|
||
self.assertTrue(out["error"].startswith("хаб:"), out)
|
||
self.assertTrue(out["retryable"])
|
||
|
||
def test_fetch_no_url(self):
|
||
out = self.ex.run({"id": 1, "type": "wiki_fetch", "params": {"attachment_id": 7}})
|
||
self.assertEqual(out["status"], "failed")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|