RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
362 lines
16 KiB
Python
362 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""旧业务模块回归测试；网络、邮件、签到及生产文件路径全部隔离。"""
|
|
import asyncio
|
|
import queue
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
BASE = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(BASE / "bot-utility"), str(BASE)]
|
|
|
|
# The minimal audit container lacks aiohttp. Tests replace every network object,
|
|
# so a tiny import stub is sufficient there; production uses the real dependency.
|
|
try:
|
|
import aiohttp # noqa: F401
|
|
except ModuleNotFoundError:
|
|
aiohttp_stub = types.ModuleType("aiohttp")
|
|
|
|
class ClientTimeout:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
aiohttp_stub.ClientTimeout = ClientTimeout
|
|
aiohttp_stub.ClientSession = None
|
|
sys.modules["aiohttp"] = aiohttp_stub
|
|
|
|
import for_admin_komari_agent as komari # noqa: E402
|
|
import ginkgostu # noqa: E402
|
|
import identity_cmd # noqa: E402
|
|
import jm # noqa: E402
|
|
import live_monitor # noqa: E402
|
|
import mail # noqa: E402
|
|
import migrate_chat_data # noqa: E402
|
|
import pixiv # noqa: E402
|
|
import setu # noqa: E402
|
|
|
|
|
|
class GinkgoRegressionTests(unittest.TestCase):
|
|
URL = "https://ginkgostu.dufe.edu.cn/course/detail/1K/batch/2/signin/9/2"
|
|
|
|
def test_retry_status_confirms_success_without_another_commit(self):
|
|
response = MagicMock()
|
|
response.json.return_value = {
|
|
"data": {"signInModel": {"type": 2, "title": "t"}, "signStatus": 1}
|
|
}
|
|
session = MagicMock()
|
|
with patch.object(ginkgostu, "call_api", return_value=response), \
|
|
patch.object(ginkgostu, "commit_sign_in") as commit:
|
|
result = ginkgostu.fetch_sign_in_info(
|
|
"u", "p", self.URL, sess=session, token="token",
|
|
avoid_duplicate=True)
|
|
self.assertTrue(result["commit"]["success"])
|
|
commit.assert_not_called()
|
|
session.close.assert_not_called() # 调用方所有的缓存会话不能关闭
|
|
|
|
def test_owned_signin_session_closes_on_query_failure(self):
|
|
session = MagicMock()
|
|
with patch.object(ginkgostu, "login", return_value=(session, "token", "n")), \
|
|
patch.object(ginkgostu, "call_api",
|
|
side_effect=ginkgostu.UpstreamError("bad response")):
|
|
with self.assertRaises(ginkgostu.UpstreamError):
|
|
ginkgostu.fetch_sign_in_info("u", "p", self.URL)
|
|
session.close.assert_called_once_with()
|
|
|
|
def test_login_failure_closes_sso_session(self):
|
|
session = MagicMock()
|
|
with patch.object(ginkgostu, "new_session", return_value=session), \
|
|
patch.object(ginkgostu, "get_unify_config",
|
|
side_effect=ginkgostu.UpstreamError("failed")):
|
|
with self.assertRaises(ginkgostu.UpstreamError):
|
|
ginkgostu.login("u", "p")
|
|
session.close.assert_called_once_with()
|
|
|
|
def test_refresh_closes_replaced_cached_session(self):
|
|
old = MagicMock()
|
|
new = MagicMock()
|
|
cached = ginkgostu.GinkgoSession("u", "p")
|
|
cached.req = old
|
|
with patch.object(ginkgostu, "login", return_value=(new, "new-token", "n")):
|
|
cached.refresh()
|
|
old.close.assert_called_once_with()
|
|
self.assertIs(cached.req, new)
|
|
|
|
def test_signin_api_forwards_retry_status_check(self):
|
|
with patch.object(ginkgostu, "_call_sign_in_local",
|
|
return_value=(True, "ok", {}, False)) as local:
|
|
ginkgostu.call_sign_in_api("u", "p", self.URL, check_status=True)
|
|
self.assertTrue(local.call_args.kwargs["check_status"])
|
|
|
|
|
|
class MailRegressionTests(unittest.TestCase):
|
|
def test_transport_result_unknown_is_not_a_plain_send_failure(self):
|
|
user = {"addresses": ["sender@rrrosmontis.icu"], "default": "",
|
|
"push": {"enabled": False, "last_seen": {}}}
|
|
with patch.object(mail, "_load_user", return_value=user), \
|
|
patch.object(mail, "_ComSession", side_effect=RuntimeError("lost response")):
|
|
result = asyncio.run(mail.send_with_attachments(
|
|
42, "sender@rrrosmontis.icu", "to@example.net", "s", "b", []))
|
|
self.assertFalse(result["ok"])
|
|
self.assertEqual(result["error_code"], "send_unknown")
|
|
|
|
def test_explicit_bridge_rejection_remains_send_failed(self):
|
|
user = {"addresses": ["sender@rrrosmontis.icu"], "default": "",
|
|
"push": {"enabled": False, "last_seen": {}}}
|
|
app = MagicMock()
|
|
app.post.return_value = {"accepted": False}
|
|
with patch.object(mail, "_load_user", return_value=user), \
|
|
patch.object(mail, "_ComSession") as session:
|
|
session.return_value.__enter__.return_value = app
|
|
result = asyncio.run(mail.send_with_attachments(
|
|
42, "sender@rrrosmontis.icu", "to@example.net", "s", "b", []))
|
|
self.assertEqual(result["error_code"], "send_failed")
|
|
|
|
|
|
class LiveMonitorRegressionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client = live_monitor.app.test_client()
|
|
live_monitor._live_sessions.clear()
|
|
self.records = {"follows": {"42": [{
|
|
"chat_type": "group", "chat_id": 7, "channel": "qq", "site": "1",
|
|
}]}}
|
|
self.secret = patch.object(live_monitor, "SRS_CALLBACK_SECRET", "test-secret")
|
|
self.load = patch.object(live_monitor, "_load_records", return_value=self.records)
|
|
self.notifications = patch.object(live_monitor, "notification_queue", queue.Queue())
|
|
self.secret.start()
|
|
self.load.start()
|
|
self.notifications.start()
|
|
|
|
def tearDown(self):
|
|
self.notifications.stop()
|
|
self.load.stop()
|
|
self.secret.stop()
|
|
live_monitor._live_sessions.clear()
|
|
|
|
def test_missing_srs_secret_fails_closed(self):
|
|
with patch.object(live_monitor, "SRS_CALLBACK_SECRET", ""):
|
|
response = self.client.post(
|
|
"/remote_srs/on_publish", json={"stream": "room_42"})
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertTrue(live_monitor.notification_queue.empty())
|
|
|
|
def test_srs_auth_dedupe_and_unpublish_rearm(self):
|
|
payload = {"stream": "room_42"}
|
|
denied = self.client.post("/remote_srs/on_publish", json=payload)
|
|
self.assertEqual(denied.status_code, 403)
|
|
self.assertTrue(live_monitor.notification_queue.empty())
|
|
|
|
for _ in range(2):
|
|
accepted = self.client.post(
|
|
"/remote_srs/on_publish?token=test-secret", json=payload)
|
|
self.assertEqual(accepted.status_code, 200)
|
|
self.assertEqual(live_monitor.notification_queue.qsize(), 1)
|
|
|
|
denied_end = self.client.post("/remote_srs/on_unpublish", json=payload)
|
|
self.assertEqual(denied_end.status_code, 403)
|
|
ended = self.client.post(
|
|
"/remote_srs/on_unpublish", json=payload,
|
|
headers={"X-SRS-Token": "test-secret"})
|
|
self.assertEqual(ended.status_code, 200)
|
|
self.client.post("/remote_srs/on_publish?token=test-secret", json=payload)
|
|
self.assertEqual(live_monitor.notification_queue.qsize(), 2)
|
|
|
|
|
|
class KomariAndIdentityTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_recent_exception_preserves_state_without_offline_notice(self):
|
|
komari._last_state = {"node-1": {"name": "one", "offline": False}}
|
|
session = MagicMock()
|
|
session.__aenter__ = AsyncMock(return_value=session)
|
|
session.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
async def get_json(_session, path, token=None):
|
|
if path == "/api/nodes":
|
|
return {"status": "success", "data": [{"uuid": "node-1", "name": "one"}]}
|
|
raise RuntimeError("temporary recent failure")
|
|
|
|
with patch.object(komari, "_login", AsyncMock(return_value="key")), \
|
|
patch.object(komari.aiohttp, "ClientSession", return_value=session), \
|
|
patch.object(komari, "_get_json", side_effect=get_json), \
|
|
patch.object(komari, "_notify_admins", AsyncMock()) as notify:
|
|
await komari._check_servers_once()
|
|
self.assertEqual(komari._last_state["node-1"]["offline"], False)
|
|
notify.assert_not_awaited()
|
|
|
|
async def test_notification_send_false_is_not_reported_as_sent(self):
|
|
channel = object()
|
|
with patch.object(identity_cmd, "get_channel", return_value=channel), \
|
|
patch.object(identity_cmd, "send_message", AsyncMock(return_value=False)):
|
|
self.assertFalse(await identity_cmd._notify("qq", 42, "text"))
|
|
with patch.object(identity_cmd, "get_channel", return_value=channel), \
|
|
patch.object(identity_cmd, "send_message", AsyncMock(return_value=True)):
|
|
self.assertTrue(await identity_cmd._notify("qq", 42, "text"))
|
|
|
|
|
|
class QuarantineRegressionTests(unittest.TestCase):
|
|
def test_same_name_quarantine_files_never_overwrite(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
source = root / "source"
|
|
dest = root / "dest"
|
|
first = source / "a" / "same.txt"
|
|
second = source / "b" / "same.txt"
|
|
first.parent.mkdir(parents=True)
|
|
second.parent.mkdir(parents=True)
|
|
migrator = migrate_chat_data.ChatMigrator.__new__(
|
|
migrate_chat_data.ChatMigrator)
|
|
migrator.source = source
|
|
migrator.dest = dest
|
|
migrator.report = {"quarantine": []}
|
|
|
|
first.write_bytes(b"first")
|
|
second.write_bytes(b"second")
|
|
migrator._quarantine(first, "history", "one")
|
|
migrator._quarantine(second, "history", "two")
|
|
first.write_bytes(b"changed later")
|
|
migrator._quarantine(first, "history", "three")
|
|
|
|
saved = sorted(p.read_bytes() for p in (dest / "quarantine" / "history").iterdir())
|
|
self.assertEqual(saved, [b"changed later", b"first", b"second"])
|
|
|
|
|
|
class JmRegressionTests(unittest.TestCase):
|
|
def test_local_pack_import_and_stale_workdir_cleanup(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
album = "350234"
|
|
stale = root / album / "old-chapter" / "stale.jpg"
|
|
stale.parent.mkdir(parents=True)
|
|
stale.write_bytes(b"stale")
|
|
plan = ("title", [(101,)], 1, 1, "title.zip")
|
|
|
|
class Downloader:
|
|
def __init__(self, _option):
|
|
pass
|
|
|
|
def download_photo(self, photo_id):
|
|
target = root / album / str(photo_id)
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
(target / "image.webp").write_bytes(b"not-an-image-is-copied")
|
|
|
|
class Options:
|
|
@staticmethod
|
|
def construct(_config):
|
|
return object()
|
|
|
|
class Config:
|
|
option_class = Options
|
|
|
|
@staticmethod
|
|
def downloader_class():
|
|
return Downloader
|
|
|
|
jmcomic_stub = types.ModuleType("jmcomic")
|
|
jmcomic_stub.JmModuleConfig = Config
|
|
with patch.object(jm, "JM_DIR", str(root)), \
|
|
patch.dict(sys.modules, {"jmcomic": jmcomic_stub}):
|
|
title, zip_path, count, ok, failed = jm._download_and_pack(
|
|
album, 1, 1, plan=plan)
|
|
self.assertEqual((title, count, ok, failed), ("title", 1, 1, 0))
|
|
self.assertTrue(zip_path.is_file())
|
|
self.assertFalse(stale.exists())
|
|
|
|
|
|
class ImageCleanupTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_pixiv_cancellation_cleans_cache_and_releases_lock(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
pid = "424242"
|
|
local = Path(tmp) / (pid + ".jpg")
|
|
local.write_bytes(b"image")
|
|
lock = pixiv.get_pid_lock(pid)
|
|
with patch.object(pixiv, "SEND_IMAGE", 1), \
|
|
patch.object(pixiv, "RETAIN_MODE", "oss"), \
|
|
patch.object(pixiv, "IS_NAPCAT_SERVER_OSS_AVAILABLE", False), \
|
|
patch.object(pixiv, "OSS_SAVE_DIR", str(Path(tmp) / "oss")), \
|
|
patch.object(pixiv, "_load_proxy_ports", return_value=[]), \
|
|
patch.object(pixiv, "download_image", return_value=local), \
|
|
patch.object(pixiv, "image_base64", return_value="base64://x"), \
|
|
patch("channel.send_message", AsyncMock(return_value=True)), \
|
|
patch.object(pixiv.asyncio, "sleep",
|
|
AsyncMock(side_effect=asyncio.CancelledError)):
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await pixiv.handle_pixiv(MagicMock(name="qq"), "private", 1, 2,
|
|
".pixiv " + pid)
|
|
self.assertFalse(local.exists())
|
|
self.assertFalse(lock.locked())
|
|
|
|
async def test_setu_cancellation_cleans_cache_and_releases_lock(self):
|
|
class Response:
|
|
status = 200
|
|
|
|
async def json(self):
|
|
return {"data": [{"pid": 77, "urls": {
|
|
"original": "https://i.pixiv.re/img/77.jpg"}}]}
|
|
|
|
class Context:
|
|
async def __aenter__(self):
|
|
return Response()
|
|
|
|
async def __aexit__(self, *args):
|
|
return None
|
|
|
|
class Session:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *args):
|
|
return None
|
|
|
|
def get(self, *args, **kwargs):
|
|
return Context()
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
local = Path(tmp) / "77.jpg"
|
|
local.write_bytes(b"image")
|
|
lock = pixiv.get_pid_lock(77)
|
|
with patch.object(setu, "SEND_IMAGE", 1), \
|
|
patch.object(setu, "RETAIN_MODE", "oss"), \
|
|
patch.object(setu, "IS_NAPCAT_SERVER_OSS_AVAILABLE", False), \
|
|
patch.object(setu, "OSS_SAVE_DIR", str(Path(tmp) / "oss")), \
|
|
patch.object(setu, "_load_proxy_ports", return_value=[]), \
|
|
patch.object(setu, "count_oss_files", return_value=0), \
|
|
patch.object(setu.aiohttp, "ClientSession", Session), \
|
|
patch.object(pixiv, "download_image", return_value=local), \
|
|
patch.object(pixiv, "image_base64", return_value="base64://x"), \
|
|
patch("channel.send_message", AsyncMock(return_value=True)), \
|
|
patch.object(setu.asyncio, "sleep",
|
|
AsyncMock(side_effect=asyncio.CancelledError)):
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await setu.handle_setu([], None, None, "private", 1, MagicMock(name="qq"))
|
|
self.assertFalse(local.exists())
|
|
self.assertFalse(lock.locked())
|
|
|
|
async def test_interrupted_download_removes_partial_file(self):
|
|
class Response:
|
|
status_code = 200
|
|
headers = {"Content-Type": "image/jpeg"}
|
|
|
|
def iter_content(self, _size):
|
|
yield b"partial"
|
|
raise RuntimeError("connection lost")
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
response = Response()
|
|
with tempfile.TemporaryDirectory() as tmp, \
|
|
patch.object(pixiv, "PIXIV_CACHE_DIR", tmp), \
|
|
patch("requests.get", return_value=response):
|
|
with self.assertRaises(RuntimeError):
|
|
pixiv.download_image("88")
|
|
self.assertEqual(list(Path(tmp).iterdir()), [])
|
|
self.assertTrue(response.closed)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|