97 lines
3.5 KiB
Python
97 lines
3.5 KiB
Python
"""
|
|
附件上传 H5：token 不含账号，过期 410，拒收图片。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
if ROOT not in sys.path:
|
|
sys.path.insert(0, ROOT)
|
|
|
|
os.environ.setdefault("YTD_ENV", "test")
|
|
os.environ.setdefault("REDIS_BACKEND", "memory")
|
|
os.environ.setdefault("CHECKPOINT_BACKEND", "memory")
|
|
os.environ.setdefault("MESSAGE_STORE_BACKEND", "memory")
|
|
os.environ.setdefault("LEDGER_BACKEND", "memory")
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from agent.channel.h5 import router as h5_router
|
|
from agent.channel.h5.copy_store import build_upload_url
|
|
from agent.channel.wecom.models import InboundMessage
|
|
from agent.handlers.attachment_inquiry import handle_attachment_inquiry, lookup_upload_sender
|
|
from agent.policy.attachment_wave import reset_attach_wave_book_for_test
|
|
from agent.policy.image_wave import reset_image_wave_book_for_test
|
|
from agent.policy.air_text_flow import reset_air_text_flow_for_test
|
|
from agent.policy.land_text_flow import reset_land_text_flow_for_test
|
|
from agent.policy.sea_text_flow import reset_sea_text_flow_for_test
|
|
|
|
|
|
class AttachmentUploadH5Tests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
reset_attach_wave_book_for_test()
|
|
reset_image_wave_book_for_test()
|
|
reset_sea_text_flow_for_test()
|
|
reset_air_text_flow_for_test()
|
|
reset_land_text_flow_for_test()
|
|
app = FastAPI()
|
|
app.include_router(h5_router)
|
|
self.client = TestClient(app)
|
|
|
|
def _open_card(self) -> str:
|
|
handle_attachment_inquiry(
|
|
InboundMessage(sender_id="s1", message_id="m1", content="附件询价")
|
|
)
|
|
from agent.policy.attachment_wave import get_attach_wave_book
|
|
|
|
wave = get_attach_wave_book().active("s1")
|
|
self.assertIsNotNone(wave)
|
|
url = build_upload_url(wave.token)
|
|
self.assertIn("/upload/", url)
|
|
return wave.token
|
|
|
|
def test_open_page_ok_and_expired(self) -> None:
|
|
token = self._open_card()
|
|
self.assertEqual(lookup_upload_sender(token), "s1")
|
|
hit = self.client.get(f"/inquiry-form/upload/{token}")
|
|
self.assertEqual(hit.status_code, 200)
|
|
self.assertIn("Excel", hit.text)
|
|
self.assertNotIn("s1", hit.text)
|
|
miss = self.client.get("/inquiry-form/upload/not-a-real-token")
|
|
self.assertEqual(miss.status_code, 410)
|
|
self.assertIn("过期", miss.text)
|
|
|
|
def test_upload_page_uses_card_chrome(self) -> None:
|
|
from agent.channel.h5.upload_page import expired_upload_page, render_upload_message, render_upload_page
|
|
|
|
page = render_upload_page()
|
|
self.assertIn('class="card"', page)
|
|
self.assertIn('class="drop"', page)
|
|
self.assertIn('id="picked"', page)
|
|
self.assertIn("一次可选多个", page)
|
|
self.assertIn("Excel", page)
|
|
self.assertNotIn("s1", page)
|
|
ok = render_upload_message("上传成功", "已收到，请回企微查看识别结果。")
|
|
dead = expired_upload_page()
|
|
self.assertIn('class="card"', ok)
|
|
self.assertIn('class="card"', dead)
|
|
self.assertIn("过期", dead)
|
|
|
|
def test_post_png_rejected(self) -> None:
|
|
token = self._open_card()
|
|
hit = self.client.post(
|
|
f"/inquiry-form/upload/{token}",
|
|
files=[("files", ("shot.png", b"\x89PNG", "image/png"))],
|
|
)
|
|
self.assertEqual(hit.status_code, 400)
|
|
self.assertIn("Excel", hit.text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|