from datetime import datetime from app.crawler.parsers import ( extract_pagination, parse_dahuagov_html, parse_gxgp_api_response, ) def make_api_response(records_data): return { "success": True, "result": { "data": { "data": records_data, "total": len(records_data), "pageNo": 1, "pageSize": 100, "pages": 1, "empty": len(records_data) == 0, "hasNext": False, "hasPrevious": False, } }, } def test_parse_gxgp_single_record(): response = make_api_response([ { "title": "测试采购公告", "publishDate": 1746720000000, "purchaseName": "测试采购单位", "articleId": 12345, } ]) crawled_at = datetime(2026, 5, 9, 10, 0, 0) results = parse_gxgp_api_response( response, source_code="ZcyAnnouncement1", source_name="采购公告", crawled_at=crawled_at, category_id=66485, ) assert len(results) == 1 assert results[0]["title"] == "测试采购公告" assert results[0]["source_code"] == "ZcyAnnouncement1" assert "content_hash" in results[0] def test_parse_gxgp_empty_response(): response = make_api_response([]) crawled_at = datetime(2026, 5, 9, 10, 0, 0) results = parse_gxgp_api_response( response, "ZcyAnnouncement1", "采购公告", crawled_at, 66485 ) assert len(results) == 0 def test_parse_gxgp_missing_title(): response = make_api_response([ {"title": "", "publishDate": 1746720000000, "purchaseName": "x", "articleId": 1} ]) crawled_at = datetime(2026, 5, 9, 10, 0, 0) results = parse_gxgp_api_response( response, "ZcyAnnouncement1", "采购公告", crawled_at, 66485 ) assert len(results) == 0 def test_extract_pagination(): response = make_api_response([]) pagination = extract_pagination(response) assert pagination["total"] == 0 assert pagination["page_no"] == 1 assert pagination["has_next"] is False def test_parse_dahuagov_html(): html = """