5bd44774b9
- 22 tests covering SearchRecJobs, GetBrandDetail, SearchBrandJobs, GetJobDetail, BossClient - Uses MagicMock (requests_go not compatible with respx) - Covers success responses, HTTP errors, biz errors, Traceid injection - All 22 tests pass (0.08s)
272 lines
11 KiB
Python
272 lines
11 KiB
Python
"""
|
||
Boss 直聘 HTTP 层 mock 测试（QUAL-03）
|
||
|
||
使用 unittest.mock.MagicMock 替代真实 HTTP 客户端，
|
||
覆盖正常响应和错误响应场景，无网络依赖。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from unittest.mock import MagicMock
|
||
|
||
from spiderJobs.platforms.boss.api import (
|
||
GetBrandDetail,
|
||
GetJobDetail,
|
||
SearchBrandJobs,
|
||
SearchRecJobs,
|
||
_parse_boss_response,
|
||
)
|
||
from spiderJobs.platforms.boss.client import BossClient
|
||
from crawler_core.base import Result
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────
|
||
# 1. _parse_boss_response 纯函数测试
|
||
# ─────────────────────────────────────────────────────────
|
||
|
||
class TestParseBossResponse:
|
||
|
||
def test_http_error_returns_failure(self):
|
||
result = _parse_boss_response(500, {})
|
||
assert result.success is False
|
||
assert result.status_code == 500
|
||
|
||
def test_non_dict_raw_returns_failure(self):
|
||
result = _parse_boss_response(200, "not a dict")
|
||
assert result.success is False
|
||
|
||
def test_biz_error_code_35_returns_failure(self):
|
||
result = _parse_boss_response(200, {"code": 35, "message": "IP地址存在异常"})
|
||
assert result.success is False
|
||
assert result.status_code == 35
|
||
assert "IP" in result.error
|
||
|
||
def test_joblist_payload_parsed_correctly(self):
|
||
raw = {
|
||
"code": 0,
|
||
"zpData": {
|
||
"jobList": [{"title": "Python工程师"}],
|
||
"hasMore": True,
|
||
},
|
||
}
|
||
result = _parse_boss_response(200, raw)
|
||
assert result.success is True
|
||
assert len(result.list) == 1
|
||
assert result.list[0]["title"] == "Python工程师"
|
||
assert result.is_end_page is False # hasMore=True → is_end_page=False
|
||
|
||
def test_joblist_no_more_pages(self):
|
||
raw = {
|
||
"code": 0,
|
||
"zpData": {"jobList": [{"title": "测试"}], "hasMore": False},
|
||
}
|
||
result = _parse_boss_response(200, raw)
|
||
assert result.is_end_page is True
|
||
|
||
def test_detail_payload(self):
|
||
raw = {"code": 0, "zpData": {"companyName": "测试公司"}}
|
||
result = _parse_boss_response(200, raw)
|
||
assert result.success is True
|
||
assert result.data == {"companyName": "测试公司"}
|
||
|
||
def test_list_field_payload_parsed(self):
|
||
raw = {
|
||
"code": 0,
|
||
"zpData": {"list": [{"jobName": "运营岗"}], "hasMore": True},
|
||
}
|
||
result = _parse_boss_response(200, raw)
|
||
assert result.success is True
|
||
assert len(result.list) == 1
|
||
assert result.is_end_page is False
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────
|
||
# 2. SearchRecJobs
|
||
# ─────────────────────────────────────────────────────────
|
||
|
||
class TestSearchRecJobs:
|
||
|
||
def _make_mock_client(self, return_value):
|
||
mock_client = MagicMock()
|
||
mock_client.get.return_value = return_value
|
||
return mock_client
|
||
|
||
def test_search_success(self):
|
||
raw = {
|
||
"code": 0,
|
||
"zpData": {
|
||
"jobList": [{"title": "测试职位1"}, {"title": "测试职位2"}],
|
||
"hasMore": False,
|
||
},
|
||
}
|
||
searcher = SearchRecJobs(city_code="101010100", client=self._make_mock_client((200, raw)))
|
||
result = searcher.search(page_index=1)
|
||
|
||
assert result.success is True
|
||
assert len(result.list) == 2
|
||
assert result.is_end_page is True
|
||
|
||
def test_search_http_error(self):
|
||
searcher = SearchRecJobs(client=self._make_mock_client((403, {})))
|
||
result = searcher.search(page_index=1)
|
||
assert result.success is False
|
||
assert result.status_code == 403
|
||
|
||
def test_search_biz_error(self):
|
||
raw = {"code": 35, "message": "IP地址存在异常"}
|
||
searcher = SearchRecJobs(client=self._make_mock_client((200, raw)))
|
||
result = searcher.search(page_index=1)
|
||
assert result.success is False
|
||
|
||
def test_search_builds_city_code_param(self):
|
||
mock_client = MagicMock()
|
||
mock_client.get.return_value = (200, {"code": 0, "zpData": {"jobList": [], "hasMore": False}})
|
||
searcher = SearchRecJobs(city_code="101280600", page_size=10, client=mock_client)
|
||
searcher.search(page_index=2)
|
||
|
||
# 验证 get 方法被调用
|
||
assert mock_client.get.called
|
||
call_args = mock_client.get.call_args
|
||
# 第二个参数是 params dict
|
||
params = call_args[0][1] if len(call_args[0]) > 1 else call_args[1].get("params")
|
||
assert params["cityCode"] == "101280600"
|
||
assert params["page"] == 2
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────
|
||
# 3. GetBrandDetail
|
||
# ─────────────────────────────────────────────────────────
|
||
|
||
class TestGetBrandDetail:
|
||
|
||
def test_fetch_success(self):
|
||
mock_client = MagicMock()
|
||
mock_client.get.return_value = (200, {
|
||
"code": 0,
|
||
"zpData": {"brandName": "测试公司", "brandId": "abc123"},
|
||
})
|
||
fetcher = GetBrandDetail(brand_id="abc123", client=mock_client)
|
||
result = fetcher.fetch()
|
||
|
||
assert result.success is True
|
||
assert result.data["brandName"] == "测试公司"
|
||
|
||
def test_fetch_404(self):
|
||
mock_client = MagicMock()
|
||
mock_client.get.return_value = (404, {})
|
||
fetcher = GetBrandDetail(brand_id="notexist", client=mock_client)
|
||
result = fetcher.fetch()
|
||
|
||
assert result.success is False
|
||
assert result.status_code == 404
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────
|
||
# 4. SearchBrandJobs
|
||
# ─────────────────────────────────────────────────────────
|
||
|
||
class TestSearchBrandJobs:
|
||
|
||
def test_search_success_has_more(self):
|
||
mock_client = MagicMock()
|
||
mock_client.get.return_value = (200, {
|
||
"code": 0,
|
||
"zpData": {"list": [{"jobName": "测试岗位"}], "hasMore": True},
|
||
})
|
||
searcher = SearchBrandJobs(brand_id="abc123", client=mock_client)
|
||
result = searcher.search(page_index=1)
|
||
|
||
assert result.success is True
|
||
assert len(result.list) == 1
|
||
assert result.is_end_page is False
|
||
|
||
def test_search_success_no_more(self):
|
||
mock_client = MagicMock()
|
||
mock_client.get.return_value = (200, {
|
||
"code": 0,
|
||
"zpData": {"list": [], "hasMore": False},
|
||
})
|
||
searcher = SearchBrandJobs(brand_id="abc123", client=mock_client)
|
||
result = searcher.search(page_index=1)
|
||
|
||
assert result.is_end_page is True
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────
|
||
# 5. GetJobDetail（batch 接口路径）
|
||
# ─────────────────────────────────────────────────────────
|
||
|
||
class TestGetJobDetail:
|
||
|
||
def test_fetch_success_merges_sub_requests(self):
|
||
mock_client = MagicMock()
|
||
mock_client.batch.return_value = (200, {
|
||
"code": 0,
|
||
"zpData": {
|
||
"/wapi/zpgeek/miniapp/job/detail.json": {
|
||
"zpData": {"jobName": "数据工程师"}
|
||
},
|
||
"/wapi/zpgeek/miniapp/jobdetail/improvement/query.json": {
|
||
"zpData": {"tags": ["Python", "大数据"]}
|
||
},
|
||
},
|
||
})
|
||
fetcher = GetJobDetail(security_id="sid123", job_id="jid456", client=mock_client)
|
||
result = fetcher.fetch()
|
||
|
||
assert result.success is True
|
||
assert result.data["detail"]["jobName"] == "数据工程师"
|
||
assert "Python" in result.data["improvement"]["tags"]
|
||
|
||
def test_fetch_biz_error(self):
|
||
mock_client = MagicMock()
|
||
mock_client.batch.return_value = (200, {"code": 35, "message": "IP地址存在异常"})
|
||
fetcher = GetJobDetail(security_id="sid", job_id="jid", client=mock_client)
|
||
result = fetcher.fetch()
|
||
assert result.success is False
|
||
|
||
def test_fetch_exception_handled(self):
|
||
mock_client = MagicMock()
|
||
mock_client.batch.side_effect = ConnectionError("连接超时")
|
||
fetcher = GetJobDetail(security_id="sid", job_id="jid", client=mock_client)
|
||
result = fetcher.fetch()
|
||
assert result.success is False
|
||
assert "连接超时" in result.error
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────
|
||
# 6. BossClient — Traceid/mpt/wt2 请求头注入
|
||
# ─────────────────────────────────────────────────────────
|
||
|
||
class TestBossClientHeaders:
|
||
|
||
def test_get_injects_traceid(self):
|
||
"""每次请求头包含 Traceid（以 M-W 开头）"""
|
||
client = BossClient()
|
||
headers = client._boss_headers()
|
||
assert "Traceid" in headers
|
||
assert headers["Traceid"].startswith("M-W")
|
||
|
||
def test_traceid_length_is_valid(self):
|
||
"""Traceid 格式长度正确（前缀 + 22 字符）"""
|
||
client = BossClient()
|
||
traceid = client._boss_headers()["Traceid"]
|
||
# 格式：M-W (3) + uuid_19 + checksum_3 = 25
|
||
assert len(traceid) >= 20
|
||
|
||
def test_mpt_wt2_in_headers(self):
|
||
"""signer 的 mpt/wt2 注入到请求头"""
|
||
from crawler_core.boss.sign import BossSign
|
||
signer = BossSign(mpt="test_mpt_value", wt2="test_wt2_value")
|
||
client = BossClient(signer=signer)
|
||
headers = client._boss_headers()
|
||
assert headers["mpt"] == "test_mpt_value"
|
||
assert headers["wt2"] == "test_wt2_value"
|
||
|
||
def test_default_signer_empty_mpt_wt2(self):
|
||
"""默认 signer（无登录）mpt/wt2 为空字符串"""
|
||
client = BossClient()
|
||
headers = client._boss_headers()
|
||
assert headers["mpt"] == ""
|
||
assert headers["wt2"] == ""
|