v6ole
d851dafea9
- Pipeline 新增 _exclude_sent() 方法，通知前查询数据库跳过 is_sent=True 的公告 - GXGP Spider mark_sent 改为 True，配合去重逻辑防止每次爬取重复通知
74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.crawler.base import PipelineConfig
|
|
from app.crawler.gxgp_spider import GXGPSpider
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gxgp_spider_attributes():
|
|
spider = GXGPSpider()
|
|
assert spider.name == "gxgp"
|
|
assert spider.source_code == "gxgp"
|
|
assert spider.source_name == "广西政府采购网"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gxgp_spider_pipeline_config():
|
|
spider = GXGPSpider()
|
|
config = spider.get_pipeline_config()
|
|
assert isinstance(config, PipelineConfig)
|
|
assert config.filter_enabled is True
|
|
assert config.notify_mode == "filtered"
|
|
assert config.mark_sent is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gxgp_spider_crawl_empty():
|
|
spider = GXGPSpider()
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"success": True,
|
|
"result": {"data": {"data": [], "total": 0, "pageNo": 1, "pageSize": 100,
|
|
"pages": 0, "empty": True, "hasNext": False, "hasPrevious": False}},
|
|
}
|
|
|
|
with patch("httpx.AsyncClient.post", AsyncMock(return_value=mock_response)):
|
|
result = await spider.crawl(sources=["ZcyAnnouncement1"])
|
|
assert result.total_count == 0
|
|
assert len(result.announcements) == 0
|
|
assert result.success is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gxgp_spider_crawl_with_data():
|
|
spider = GXGPSpider()
|
|
|
|
mock_data = {
|
|
"success": True,
|
|
"result": {
|
|
"data": {
|
|
"data": [{
|
|
"title": "测试采购公告",
|
|
"publishDate": 1746720000000,
|
|
"purchaseName": "测试单位",
|
|
"articleId": 12345,
|
|
}],
|
|
"total": 1, "pageNo": 1, "pageSize": 100,
|
|
"pages": 1, "empty": False, "hasNext": False, "hasPrevious": False,
|
|
}
|
|
},
|
|
}
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = mock_data
|
|
|
|
with patch("httpx.AsyncClient.post", AsyncMock(return_value=mock_response)):
|
|
result = await spider.crawl(sources=["ZcyAnnouncement1"])
|
|
assert result.total_count >= 0
|
|
assert result.success is True
|