RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
120 lines
4.7 KiB
Python
120 lines
4.7 KiB
Python
"""Bilibili OpenAPI command tests; no real network or message delivery."""
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
BASE = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(BASE / "bot-utility"), str(BASE)]
|
|
|
|
import bilibili
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, status=200, content_type="video/mp4", chunks=(b"video",), text=""):
|
|
self.status_code = status
|
|
self.headers = {"Content-Type": content_type}
|
|
self._chunks = chunks
|
|
self.text = text
|
|
self.closed = False
|
|
|
|
def iter_content(self, size):
|
|
yield from self._chunks
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
class FakeChannel:
|
|
name = "tg"
|
|
|
|
@staticmethod
|
|
def is_admin(uid):
|
|
return False
|
|
|
|
|
|
class BilibiliTests(unittest.IsolatedAsyncioTestCase):
|
|
def test_sanitize_bv(self):
|
|
self.assertEqual(bilibili.sanitize_bv("bv1uT4y1P7CX"), "BV1uT4y1P7CX")
|
|
self.assertEqual(bilibili.sanitize_bv(" BV1uT4y1P7CX "), "BV1uT4y1P7CX")
|
|
self.assertIsNone(bilibili.sanitize_bv("1uT4y1P7CX"))
|
|
self.assertIsNone(bilibili.sanitize_bv("https://www.bilibili.com/video/BV1uT4y1P7CX"))
|
|
self.assertIsNone(bilibili.sanitize_bv("BV1uT4y1P7CX extra"))
|
|
|
|
async def test_handle_sends_one_file_and_cleans_temp(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
video = Path(directory) / "video.mp4"
|
|
video.write_bytes(b"video")
|
|
sender = AsyncMock(return_value=True)
|
|
with patch("channel.send_message", new=sender), \
|
|
patch.object(bilibili, "download_video", return_value=video):
|
|
await bilibili.handle_bilibili(
|
|
FakeChannel(), "private", 10001, 1, ".bilibili BV1uT4y1P7CX"
|
|
)
|
|
self.assertEqual(sender.await_count, 2)
|
|
segments = sender.await_args.args[3]
|
|
self.assertEqual(len(segments), 1)
|
|
self.assertEqual(segments[0]["type"], "file")
|
|
self.assertTrue(segments[0]["data"]["file"].startswith("file://"))
|
|
self.assertIn("bv=BV1uT4y1P7CX", segments[0]["data"]["manual"])
|
|
self.assertFalse(video.exists())
|
|
|
|
async def test_invalid_input_does_not_call_openapi(self):
|
|
sender = AsyncMock(return_value=True)
|
|
with patch("channel.send_message", new=sender), \
|
|
patch.object(bilibili, "download_video") as download:
|
|
await bilibili.handle_bilibili(FakeChannel(), "private", 10001, 1, ".bilibili nope")
|
|
download.assert_not_called()
|
|
self.assertIn("格式", sender.await_args.args[3])
|
|
|
|
async def test_openapi_error_is_reported_and_temp_is_cleaned(self):
|
|
sender = AsyncMock(return_value=True)
|
|
with patch("channel.send_message", new=sender), \
|
|
patch.object(bilibili, "download_video", side_effect=RuntimeError("OpenAPI HTTP 502")):
|
|
await bilibili.handle_bilibili(
|
|
FakeChannel(), "private", 10001, 1, ".bilibili BV1uT4y1P7CX"
|
|
)
|
|
self.assertIn("OpenAPI HTTP 502", sender.await_args.args[3])
|
|
|
|
def test_download_writes_stream_and_closes_response(self):
|
|
response = FakeResponse(chunks=(b"abc", b"def"))
|
|
with tempfile.TemporaryDirectory() as directory, \
|
|
patch.object(bilibili, "CACHE_DIR", Path(directory)), \
|
|
patch("requests.get", return_value=response) as get:
|
|
path = bilibili.download_video("BV1uT4y1P7CX")
|
|
try:
|
|
self.assertEqual(path.read_bytes(), b"abcdef")
|
|
self.assertEqual(path.suffix, ".mp4")
|
|
get.assert_called_once_with(
|
|
bilibili.OPENAPI_URL,
|
|
params={"bv": "BV1uT4y1P7CX"},
|
|
stream=True,
|
|
timeout=bilibili.HTTP_TIMEOUT,
|
|
)
|
|
self.assertTrue(response.closed)
|
|
finally:
|
|
bilibili.cleanup_file(path)
|
|
|
|
def test_download_http_error_includes_detail_and_closes_response(self):
|
|
response = FakeResponse(
|
|
status=502,
|
|
content_type="application/json",
|
|
chunks=(),
|
|
text='{"error":"download failed"}',
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory, \
|
|
patch.object(bilibili, "CACHE_DIR", Path(directory)), \
|
|
patch("requests.get", return_value=response):
|
|
with self.assertRaisesRegex(RuntimeError, "download failed"):
|
|
bilibili.download_video("BV1uT4y1P7CX")
|
|
self.assertTrue(response.closed)
|
|
self.assertEqual(list(Path(directory).iterdir()), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|