174 lines
7.8 KiB
Python
174 lines
7.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""t-261005135853: входящие documents/photo → на диск, путь в событии.
|
||
|
||
Покрытие:
|
||
1. unit: _file_ext — имя/mime/мусор/пусто;
|
||
2. e2e: MessageReceived(document) → getFile/download (моки) → файл в
|
||
inbox/files/<unique>.pdf, file_path (абсолютный) + file_name + size
|
||
в проекции inbox;
|
||
3. e2e: photo (3 размера) → .jpg самого большого размера;
|
||
4. идемпотентность: повторный апдейт не перекачивает (счётчик getFile);
|
||
5. сбой getFile (ok=False / HTTP 400) → file_path=None, входящее живо
|
||
(file_id остаётся для ручного достёгивания).
|
||
|
||
Запуск: ./venv/bin/python tests/test_tg_documents.py | pytest tests/test_tg_documents.py
|
||
Telegram-сеть мокнута (нет реальных file_id), стор — свой в /tmp.
|
||
"""
|
||
import io
|
||
import os
|
||
import sys
|
||
import urllib.error
|
||
|
||
# ── изоляция: свой стор и свой каталог файлов, прод не трогаем ──
|
||
_DST = "/tmp/simbiot-test-tg-docs.db"
|
||
_FILES = "/tmp/simbiot-test-tg-docs-files"
|
||
for p in (_DST, _DST + "-wal", _DST + "-shm"):
|
||
if os.path.exists(p):
|
||
os.unlink(p)
|
||
os.environ["SIMBIOT_DB"] = _DST
|
||
os.environ["SIMBIOT_INBOX_FILES_DIR"] = _FILES
|
||
sys.path.insert(0, "/root/simbiot-harness")
|
||
|
||
from harness import es, tg_inbox
|
||
|
||
fails = []
|
||
_calls = {"getFile": 0}
|
||
|
||
|
||
def check(name, cond, extra=""):
|
||
print(("PASS" if cond else "FAIL"), name, (f"| {extra}" if not cond else ""), flush=True)
|
||
if not cond:
|
||
fails.append(name)
|
||
|
||
|
||
# ── моки Telegram: getFile → file_path, urlopen → байты файла ──
|
||
class _Resp(io.BytesIO):
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *a):
|
||
return False
|
||
|
||
|
||
def _fake_api(method, params, timeout=35):
|
||
if method == "getFile":
|
||
_calls["getFile"] += 1
|
||
return {"ok": True, "result": {
|
||
"file_id": params["file_id"],
|
||
"file_path": f"documents/{params['file_id']}.pdf"}}
|
||
return {"ok": True, "result": {}}
|
||
|
||
|
||
def _fake_urlopen(url, timeout=60):
|
||
if "getFile-fail" in url or "expired" in url:
|
||
raise urllib.error.HTTPError(url, 400, "Bad Request", {}, io.BytesIO(b""))
|
||
return _Resp(b"%PDF-1.4 fake-test-blob")
|
||
|
||
|
||
tg_inbox._api = _fake_api
|
||
tg_inbox._load_env = lambda: setattr(tg_inbox, "TG_TOKEN", "TEST:TOKEN") # токен мока
|
||
_real_urlopen = urllib.request.urlopen
|
||
urllib.request.urlopen = lambda url, timeout=60: _fake_urlopen(url, timeout)
|
||
|
||
es.init_db()
|
||
|
||
|
||
# ────────────────────────────── 1. _file_ext ────────────────────────────────
|
||
|
||
def test_file_ext():
|
||
check("ext из имени", tg_inbox._file_ext("Alfa-Bank_выписка.PDF") == "PDF"
|
||
or tg_inbox._file_ext("Alfa-Bank_выписка.PDF").lower() == "pdf",
|
||
tg_inbox._file_ext("Alfa-Bank_выписка.PDF"))
|
||
check("ext из mime", tg_inbox._file_ext("", "image/jpeg") == "jpg",
|
||
tg_inbox._file_ext("", "image/jpeg"))
|
||
check("мусор вырезан", tg_inbox._file_ext("a.tar.gz/../../x", "") in ("gz", "x", "bin"),
|
||
tg_inbox._file_ext("a.tar.gz/../../x"))
|
||
check("пусто → bin", tg_inbox._file_ext("", "") == "bin")
|
||
|
||
|
||
# ─────────────────────── 2. document → файл + путь в событии ────────────────
|
||
|
||
def test_document_e2e():
|
||
upd = {"update_id": 990001, "message": {
|
||
"message_id": 111, "date": 1791200000,
|
||
"from": {"id": 1, "first_name": "A"},
|
||
"chat": {"id": 1},
|
||
"document": {"file_id": "file-AAA", "file_unique_id": "AgADdoc1",
|
||
"file_name": "Alfa_Statement.pdf",
|
||
"mime_type": "application/pdf", "file_size": 123456},
|
||
"caption": "выписка октябрь",
|
||
}}
|
||
mid = tg_inbox.receive_message(upd)
|
||
p = f"{_FILES}/AgADdoc1.pdf"
|
||
check("файл в inbox/files", os.path.exists(p), p)
|
||
check("содержимое скачано", os.path.exists(p) and open(p, "rb").read() == b"%PDF-1.4 fake-test-blob")
|
||
m = es.run_projections()["inbox"][mid]
|
||
check("file_path в событии (абс)", m.get("file_path") == p, m.get("file_path"))
|
||
check("путь открывается напрямую", os.path.exists(m.get("file_path") or ""))
|
||
check("file_name в событии", m.get("file_name") == "Alfa_Statement.pdf")
|
||
check("file_size в событии", m.get("file_size") == 123456)
|
||
check("kind=document", m.get("kind") == "document")
|
||
check("caption в text", m.get("text") == "выписка октябрь")
|
||
return upd
|
||
|
||
|
||
# ───────────────────────────── 3. photo → .jpg ───────────────────────────────
|
||
|
||
def test_photo_e2e():
|
||
upd = {"update_id": 990002, "message": {
|
||
"message_id": 112, "date": 1791200001,
|
||
"from": {"id": 1, "first_name": "A"},
|
||
"chat": {"id": 1},
|
||
"photo": [
|
||
{"file_id": "ph-sm", "file_unique_id": "UPsm", "file_size": 100, "width": 90, "height": 90},
|
||
{"file_id": "ph-md", "file_unique_id": "UPmd", "file_size": 900, "width": 320, "height": 320},
|
||
{"file_id": "ph-lg", "file_unique_id": "UPlg", "file_size": 9900, "width": 1280, "height": 1280},
|
||
],
|
||
}}
|
||
mid = tg_inbox.receive_message(upd)
|
||
p = f"{_FILES}/UPlg.jpg"
|
||
check("фото: самый большой размер сохранён", os.path.exists(p), p)
|
||
m = es.run_projections()["inbox"][mid]
|
||
check("photo file_path", m.get("file_path") == p, m.get("file_path"))
|
||
check("photo kind", m.get("kind") == "photo")
|
||
check("photo file_size", m.get("file_size") == 9900)
|
||
return upd
|
||
|
||
|
||
# ─────────────────── 4. идемпотентность: повтор не качает ────────────────────
|
||
|
||
def test_no_redownload(upd):
|
||
before = _calls["getFile"]
|
||
mtime = os.path.getmtime(f"{_FILES}/AgADdoc1.pdf")
|
||
tg_inbox.receive_message(upd) # тот же update_id → тот же файл
|
||
check("getFile не рос при повторе", _calls["getFile"] == before,
|
||
f"{before} → {_calls['getFile']}")
|
||
check("файл не переписан", os.path.getmtime(f"{_FILES}/AgADdoc1.pdf") == mtime)
|
||
|
||
|
||
# ──────────────────── 5. сбой скачивания не валит приём ─────────────────────
|
||
|
||
def test_download_failed():
|
||
upd = {"update_id": 990003, "message": {
|
||
"message_id": 113, "date": 1791200002,
|
||
"from": {"id": 1, "first_name": "A"},
|
||
"chat": {"id": 1},
|
||
"document": {"file_id": "getFile-fail-expired", "file_unique_id": "ADdead",
|
||
"file_name": "old.pdf", "mime_type": "application/pdf"},
|
||
}}
|
||
mid = tg_inbox.receive_message(upd)
|
||
m = es.run_projections()["inbox"][mid]
|
||
check("file_path=None при сбое", m.get("file_path") is None, m.get("file_path"))
|
||
check("file_id сохранился (ручной путь)", m.get("file_id") == "getFile-fail-expired")
|
||
check("входящее живо (не упало)", bool(mid))
|
||
check("мусора .part не осталось", not any(f.endswith(".part") for f in os.listdir(_FILES)))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
test_file_ext()
|
||
upd = test_document_e2e()
|
||
test_photo_e2e()
|
||
test_no_redownload(upd)
|
||
test_download_failed()
|
||
print(f"\n{'ALL PASS' if not fails else 'FAILURES: ' + ', '.join(fails)}")
|
||
sys.exit(1 if fails else 0)
|