72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
from app.crawler.gxgp_spider import GXGPSpider
|
|
from app.crawler.base import PipelineConfig
|
|
|
|
|
|
@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 False
|
|
|
|
|
|
@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
|