You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
96 lines
4.2 KiB
96 lines
4.2 KiB
import unittest
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
import pandas as pd
|
|
import asyncio
|
|
from main import search_bilibili_videos, get_cid, get_danmaku, ai_count_danmakus, cloud
|
|
|
|
class TestDanmakuProcessing(unittest.IsolatedAsyncioTestCase):
|
|
|
|
@patch('main.aiohttp.ClientSession.get', new_callable=AsyncMock)
|
|
async def test_search_no_results(self, mock_get):
|
|
# 模拟API返回无结果
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {'data': {'result': []}}
|
|
mock_get.return_value = mock_response
|
|
|
|
result = await search_bilibili_videos('不存在的关键词')
|
|
self.assertEqual(result, [])
|
|
|
|
@patch('main.aiohttp.ClientSession.get', new_callable=AsyncMock)
|
|
async def test_get_cid_failure(self, mock_get):
|
|
# 模拟API返回错误的CID信息
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {'data': []}
|
|
mock_get.return_value = mock_response
|
|
|
|
cid = await get_cid(123456, MagicMock())
|
|
self.assertIsNone(cid)
|
|
|
|
@patch('main.aiohttp.ClientSession.get', new_callable=AsyncMock)
|
|
async def test_get_danmaku_empty(self, mock_get):
|
|
# 模拟弹幕API返回空数据
|
|
mock_response = MagicMock()
|
|
mock_response.text.return_value = ''
|
|
mock_get.return_value = mock_response
|
|
danmakus = await get_danmaku(654321, MagicMock())
|
|
self.assertEqual(danmakus, [])
|
|
|
|
@patch('main.pd.read_csv')
|
|
def test_ai_count_danmakus_no_ai(self, mock_read_csv):
|
|
# 模拟读取无AI相关的弹幕数据
|
|
mock_read_csv.return_value = pd.DataFrame({'danmaku': ['今天天气真好', '我喜欢这首歌', '哈哈哈']})
|
|
ai_danmakus = ai_count_danmakus('path/to/danmaku.csv')
|
|
self.assertEqual(len(ai_danmakus), 0)
|
|
|
|
@patch('main.pd.read_csv')
|
|
def test_ai_count_danmakus_with_ai(self, mock_read_csv):
|
|
# 模拟读取包含AI相关的弹幕数据
|
|
mock_read_csv.return_value = pd.DataFrame({'danmaku': ['人工智能很有趣', '机器学习改变世界', '深度学习']})
|
|
ai_danmakus = ai_count_danmakus('path/to/danmaku.csv')
|
|
self.assertEqual(len(ai_danmakus), 3)
|
|
|
|
@patch('main.pd.read_csv')
|
|
def test_cloud_generation_empty(self, mock_read_csv):
|
|
# 模拟读取无数据的CSV文件
|
|
mock_read_csv.return_value = pd.DataFrame({'danmaku': []})
|
|
with self.assertRaises(ValueError):
|
|
cloud('path/to/danmaku.csv') # 假设 cloud 函数在无数据时抛出 ValueError
|
|
|
|
@patch('main.pd.read_csv')
|
|
def test_cloud_generation_normal(self, mock_read_csv):
|
|
# 模拟读取正常数据的CSV文件
|
|
mock_read_csv.return_value = pd.DataFrame({'danmaku': ['人工智能', '机器学习', '深度学习']})
|
|
try:
|
|
cloud('path/to/danmaku.csv')
|
|
except Exception as e:
|
|
self.fail(f"词云生成失败，出现异常：{e}")
|
|
|
|
@patch('main.pd.read_csv')
|
|
def test_large_danmaku_processing(self, mock_read_csv):
|
|
# 模拟读取大量弹幕数据的CSV文件
|
|
mock_read_csv.return_value = pd.DataFrame({'danmaku': ['AI'] * 100000})
|
|
try:
|
|
ai_danmakus = ai_count_danmakus('danmaku.csv')
|
|
# 验证返回的列表的长度是否与预期一致
|
|
self.assertEqual(len(ai_danmakus), 1) # 因为所有弹幕都是相同的，所以结果应该包含一个元素
|
|
self.assertEqual(ai_danmakus[0], 'AI: 100000 次') # 验证弹幕及其出现次数是否正确
|
|
except Exception as e:
|
|
self.fail(f"处理大量弹幕时出现异常：{e}")
|
|
|
|
@patch('main.pd.read_csv')
|
|
def test_special_characters_handling(self, mock_read_csv):
|
|
# 模拟读取包含特殊字符的弹幕数据
|
|
mock_read_csv.return_value = pd.DataFrame({'danmaku': ['😀', '人工智能❤️', '机器学习🚀']})
|
|
ai_danmakus = ai_count_danmakus('danmaku.csv')
|
|
self.assertEqual(len(ai_danmakus), 2)
|
|
|
|
@patch('main.pd.read_csv')
|
|
def test_file_io_error(self, mock_read_csv):
|
|
# 模拟文件读取权限错误
|
|
mock_read_csv.side_effect = PermissionError("无权限读取文件")
|
|
with self.assertRaises(PermissionError):
|
|
pd.read_csv('danmaku.csv')
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|