87 lines
2.3 KiB
Python
87 lines
2.3 KiB
Python
import pytest
|
|
from datetime import datetime
|
|
from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig
|
|
from app.services.crawl_service import CrawlService
|
|
|
|
|
|
class MockGXGPSpider(BaseSpider):
|
|
name = "mock_gxgp"
|
|
source_code = "mock_gxgp"
|
|
source_name = "Mock GXGP"
|
|
|
|
async def crawl(self, **kwargs):
|
|
return CrawlResult(
|
|
source_code=self.source_code,
|
|
source_name=self.source_name,
|
|
total_count=10,
|
|
new_count=5,
|
|
announcements=[],
|
|
)
|
|
|
|
def get_pipeline_config(self):
|
|
return PipelineConfig(
|
|
filter_enabled=True,
|
|
keywords=["大化"],
|
|
notify_mode="filtered",
|
|
)
|
|
|
|
|
|
class MockDahuagovSpider(BaseSpider):
|
|
name = "mock_dahuagov"
|
|
source_code = "mock_dahuagov"
|
|
source_name = "Mock Dahuagov"
|
|
|
|
async def crawl(self, **kwargs):
|
|
return CrawlResult(
|
|
source_code=self.source_code,
|
|
source_name=self.source_name,
|
|
total_count=3,
|
|
new_count=3,
|
|
announcements=[],
|
|
)
|
|
|
|
def get_pipeline_config(self):
|
|
return PipelineConfig(
|
|
filter_enabled=False,
|
|
notify_mode="all",
|
|
mark_sent=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_crawl_service_registers_spiders():
|
|
service = CrawlService()
|
|
service.register(MockGXGPSpider())
|
|
service.register(MockDahuagovSpider())
|
|
assert len(service.spiders) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_crawl_service_run_all():
|
|
service = CrawlService()
|
|
service.register(MockGXGPSpider())
|
|
service.register(MockDahuagovSpider())
|
|
results = await service.run_all()
|
|
assert len(results) == 2
|
|
assert results[0].total_count == 10
|
|
assert results[1].total_count == 3
|
|
assert all(r.success for r in results)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_crawl_service_run_specific():
|
|
service = CrawlService()
|
|
service.register(MockGXGPSpider())
|
|
service.register(MockDahuagovSpider())
|
|
results = await service.run_spider("mock_dahuagov")
|
|
assert len(results) == 1
|
|
assert results[0].source_code == "mock_dahuagov"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_crawl_service_run_unknown():
|
|
service = CrawlService()
|
|
results = await service.run_spider("nonexistent")
|
|
assert len(results) == 1
|
|
assert results[0].success is False
|