import os
import tempfile
import shutil
from django.test import TestCase, override_settings
from django.core.management import call_command
from django.core.management.base import CommandError
from django.utils import timezone
from django.core.files.uploadedfile import SimpleUploadedFile
from unittest.mock import patch, MagicMock
from .models import Website, Article
from .utils import process_article, download_media, is_valid_url, full_site_crawler
from .tasks import crawl_website, crawl_all_websites, health_check
class WebsiteModelTest(TestCase):
"""网站模型测试"""
def setUp(self):
self.website = Website.objects.create(
name='测试网站',
base_url='https://test.com',
description='测试描述'
)
def test_website_creation(self):
"""测试网站创建"""
self.assertEqual(self.website.name, '测试网站')
self.assertEqual(self.website.base_url, 'https://test.com')
self.assertTrue(self.website.enabled)
def test_website_str(self):
"""测试网站字符串表示"""
self.assertEqual(str(self.website), '测试网站')
class ArticleModelTest(TestCase):
"""文章模型测试"""
def setUp(self):
self.website = Website.objects.create(
name='测试网站',
base_url='https://test.com'
)
self.article = Article.objects.create(
website=self.website,
title='测试文章',
url='https://test.com/article/1',
content='测试内容
',
media_files=['image1.jpg', 'image2.jpg']
)
def test_article_creation(self):
"""测试文章创建"""
self.assertEqual(self.article.title, '测试文章')
self.assertEqual(self.article.url, 'https://test.com/article/1')
self.assertEqual(len(self.article.media_files), 2)
def test_article_str(self):
"""测试文章字符串表示"""
self.assertEqual(str(self.article), '测试文章')
class UtilsTest(TestCase):
"""工具函数测试"""
def setUp(self):
self.website = Website.objects.create(
name='测试网站',
base_url='https://test.com'
)
self.temp_dir = tempfile.mkdtemp()
def tearDown(self):
shutil.rmtree(self.temp_dir)
def test_is_valid_url(self):
"""测试URL验证"""
from .utils import is_valid_url
# 有效URL
self.assertTrue(is_valid_url('https://test.com/article', 'test.com'))
self.assertTrue(is_valid_url('http://test.com/article', 'test.com'))
# 无效URL
self.assertFalse(is_valid_url('https://other.com/article', 'test.com'))
self.assertFalse(is_valid_url('ftp://test.com/article', 'test.com'))
self.assertFalse(is_valid_url('invalid-url', 'test.com'))
@patch('core.utils.requests.get')
def test_download_media(self, mock_get):
"""测试媒体下载"""
# 模拟响应
mock_response = MagicMock()
mock_response.content = b'fake image content'
mock_response.headers = {'content-type': 'image/jpeg'}
mock_get.return_value = mock_response
# 测试下载
result = download_media('https://test.com/image.jpg', self.temp_dir)
self.assertIsNotNone(result)
self.assertTrue(os.path.exists(result))
@patch('core.utils.requests.get')
@patch('core.utils.download_media')
def test_process_article_success(self, mock_download_media, mock_get):
"""测试文章处理成功"""
# 模拟HTML响应
html_content = '''
测试文章
测试文章标题
测试文章内容
'''
mock_response = MagicMock()
mock_response.text = html_content
mock_response.encoding = 'utf-8'
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
# 模拟媒体下载
mock_download_media.return_value = '/tmp/test_image.jpg'
# 测试文章处理
process_article('https://test.com/article/1', self.website)
# 验证文章是否保存
article = Article.objects.filter(url='https://test.com/article/1').first()
self.assertIsNotNone(article)
self.assertEqual(article.title, '测试文章标题')
class ManagementCommandsTest(TestCase):
"""管理命令测试"""
def setUp(self):
self.website = Website.objects.create(
name='测试网站',
base_url='https://test.com'
)
@patch('core.management.commands.crawl_all_media.call_command')
def test_crawl_all_media_command(self, mock_call_command):
"""测试批量爬取命令"""
# 模拟命令执行
mock_call_command.return_value = None
# 执行命令
call_command('crawl_all_media', media='rmrb,xinhua')
# 验证命令被调用
mock_call_command.assert_called()
class CeleryTasksTest(TestCase):
"""Celery任务测试"""
def setUp(self):
self.website = Website.objects.create(
name='测试网站',
base_url='https://test.com'
)
@patch('core.tasks.full_site_crawler')
def test_crawl_website_task(self, mock_crawler):
"""测试单个网站爬取任务"""
# 模拟爬虫函数
mock_crawler.return_value = None
# 执行任务
result = crawl_website(self.website.id)
# 验证结果
self.assertEqual(result['website_id'], self.website.id)
self.assertEqual(result['website_name'], '测试网站')
self.assertEqual(result['status'], 'success')
def test_crawl_website_task_invalid_id(self):
"""测试无效网站ID的任务"""
# 执行任务
with self.assertRaises(Exception):
crawl_website(99999)
@patch('core.tasks.crawl_website.delay')
def test_crawl_all_websites_task(self, mock_delay):
"""测试批量爬取任务"""
# 模拟子任务
mock_result = MagicMock()
mock_result.id = 'task-123'
mock_delay.return_value = mock_result
# 执行任务
result = crawl_all_websites()
# 验证结果
self.assertEqual(result['total_websites'], 1)
self.assertEqual(result['status'], 'started')
def test_health_check_task(self):
"""测试健康检查任务"""
# 执行任务
result = health_check()
# 验证结果
self.assertEqual(result['database'], 'ok')
self.assertEqual(result['website_count'], 1)
self.assertEqual(result['article_count'], 0)
class IntegrationTest(TestCase):
"""集成测试"""
def setUp(self):
self.website = Website.objects.create(
name='集成测试网站',
base_url='https://integration-test.com'
)
def test_full_workflow(self):
"""测试完整工作流程"""
# 1. 创建网站
self.assertEqual(Website.objects.count(), 1)
# 2. 创建文章
article = Article.objects.create(
website=self.website,
title='集成测试文章',
url='https://integration-test.com/article/1',
content='集成测试内容
'
)
# 3. 验证关联关系
self.assertEqual(article.website, self.website)
self.assertEqual(self.website.article_set.count(), 1)
# 4. 验证数据完整性
self.assertIsNotNone(article.created_at)
self.assertIsInstance(article.media_files, list)
@override_settings(MEDIA_ROOT=tempfile.mkdtemp())
class MediaHandlingTest(TestCase):
"""媒体文件处理测试"""
def setUp(self):
self.website = Website.objects.create(
name='媒体测试网站',
base_url='https://media-test.com'
)
def test_media_files_field(self):
"""测试媒体文件字段"""
article = Article.objects.create(
website=self.website,
title='媒体测试文章',
url='https://media-test.com/article/1',
content='测试内容
',
media_files=['image1.jpg', 'video1.mp4']
)
# 验证媒体文件列表
self.assertEqual(len(article.media_files), 2)
self.assertIn('image1.jpg', article.media_files)
self.assertIn('video1.mp4', article.media_files)
class ErrorHandlingTest(TestCase):
"""错误处理测试"""
def test_duplicate_url_handling(self):
"""测试重复URL处理"""
website = Website.objects.create(
name='错误测试网站',
base_url='https://error-test.com'
)
# 创建第一篇文章
article1 = Article.objects.create(
website=website,
title='第一篇文章',
url='https://error-test.com/article/1',
content='内容1
'
)
# 尝试创建相同URL的文章
with self.assertRaises(Exception):
Article.objects.create(
website=website,
title='第二篇文章',
url='https://error-test.com/article/1', # 相同URL
content='内容2
'
)
def test_invalid_website_data(self):
"""测试无效网站数据"""
# 测试重复名称（unique约束）
Website.objects.create(
name='测试网站1',
base_url='https://test1.com'
)
with self.assertRaises(Exception):
Website.objects.create(
name='测试网站1', # 重复名称
base_url='https://test2.com'
)