JobData/tests/zhilian/test_zhilian_client.py

"""
智联招聘 HTTP 层 mock 测试（ARCH-05 / QUAL-03）

使用 unittest.mock.MagicMock 替代真实 HTTP 客户端，无网络依赖。
"""

from __future__ import annotations

from unittest.mock import MagicMock

from crawler_core.zhilian.sign import ZhilianSign
from spiderJobs.platforms.zhilian.api import (
    GetCompanyDetail,
    GetCompanyExtDetail,
    GetPositionDetail,
    SearchCompanyPositions,
    SearchPositions,
)
from spiderJobs.platforms.zhilian.client import ZhilianClient
from crawler_core.base import Result


# ─────────────────────────────────────────────────────────
# 1. SearchPositions（POST cgate）
# ─────────────────────────────────────────────────────────

class TestSearchPositions:

    def _make_client(self, status_code=200, data=None):
        mock_client = MagicMock()
        mock_client.post.return_value = (status_code, data or {})
        return mock_client

    def test_search_success_returns_list(self):
        data = {
            "data": {
                "list": [{"number": "CC123", "name": "Python 工程师"}],
                "numFound": 1,
            },
        }
        searcher = SearchPositions(
            keyword="Python", city_code=538,
            client=self._make_client(200, data),
        )
        result = searcher.search(page_index=1)
        assert result.success is True

    def test_search_http_403(self):
        searcher = SearchPositions(client=self._make_client(403, {}))
        result = searcher.search(page_index=1)
        assert result.success is False
        assert result.status_code == 403

    def test_search_http_500(self):
        searcher = SearchPositions(client=self._make_client(500, {}))
        result = searcher.search(page_index=1)
        assert result.success is False

    def test_search_builds_keyword_param(self):
        mock_client = MagicMock()
        mock_client.post.return_value = (200, {"data": {"list": []}})
        searcher = SearchPositions(keyword="Java", city_code=530, client=mock_client)
        searcher.search(page_index=1)
        assert mock_client.post.called
        call_kwargs = mock_client.post.call_args
        body = call_kwargs[0][1] if len(call_kwargs[0]) > 1 else None
        if body:
            assert "Java" in str(body)


# ─────────────────────────────────────────────────────────
# 2. GetPositionDetail（GET cgate）
# ─────────────────────────────────────────────────────────

class TestGetPositionDetail:

    def test_fetch_success(self):
        mock_client = MagicMock()
        mock_client.get.return_value = (200, {
            "data": {"number": "CC123", "jobName": "高级工程师"},
        })
        fetcher = GetPositionDetail(number="CC123", client=mock_client)
        result = fetcher.fetch()
        assert result.success is True

    def test_fetch_404(self):
        mock_client = MagicMock()
        mock_client.get.return_value = (404, {})
        fetcher = GetPositionDetail(number="notexist", client=mock_client)
        result = fetcher.fetch()
        assert result.success is False
        assert result.status_code == 404


# ─────────────────────────────────────────────────────────
# 3. GetCompanyExtDetail（GET cgate）
# ─────────────────────────────────────────────────────────

class TestGetCompanyExtDetail:

    def test_fetch_success(self):
        mock_client = MagicMock()
        mock_client.get.return_value = (200, {
            "data": {"companyName": "智联测试公司"},
        })
        fetcher = GetCompanyExtDetail(
            company_name="智联测试公司",
            company_number="CZ123",
            client=mock_client,
        )
        result = fetcher.fetch()
        assert result.success is True


# ─────────────────────────────────────────────────────────
# 4. GetCompanyDetail（GET cgate）
# ─────────────────────────────────────────────────────────

class TestGetCompanyDetail:

    def test_fetch_success(self):
        mock_client = MagicMock()
        mock_client.get.return_value = (200, {
            "data": {"companyNumber": "CZ123", "name": "智联公司"},
        })
        fetcher = GetCompanyDetail(number="CZ123", client=mock_client)
        result = fetcher.fetch()
        assert result.success is True

    def test_fetch_http_error(self):
        mock_client = MagicMock()
        mock_client.get.return_value = (500, {})
        fetcher = GetCompanyDetail(number="CZ123", client=mock_client)
        result = fetcher.fetch()
        assert result.success is False


# ─────────────────────────────────────────────────────────
# 5. SearchCompanyPositions（GET capi）— 验证 sign_params 被调用
# ─────────────────────────────────────────────────────────

class TestSearchCompanyPositions:

    def test_search_success_calls_sign_params(self):
        mock_signer = MagicMock(spec=ZhilianSign)
        mock_signer.sign_params.return_value = {"at": "", "rt": ""}
        mock_client = MagicMock()
        mock_client.signer = mock_signer
        mock_client.get.return_value = (200, {
            "data": {"list": [{"jobName": "测试岗位"}]},
            "pageInfo": {},
        })
        searcher = SearchCompanyPositions(company_id="CZ123", client=mock_client)
        result = searcher.search(page_index=1)
        assert result.success is True
        assert mock_signer.sign_params.called  # 确认 sign_params 被调用

    def test_search_http_error(self):
        mock_signer = MagicMock(spec=ZhilianSign)
        mock_signer.sign_params.return_value = {}
        mock_client = MagicMock()
        mock_client.signer = mock_signer
        mock_client.get.return_value = (403, {})
        searcher = SearchCompanyPositions(company_id="CZ123", client=mock_client)
        result = searcher.search(page_index=1)
        assert result.success is False


# ─────────────────────────────────────────────────────────
# 6. ZhilianClient — 签名头注入
# ─────────────────────────────────────────────────────────

class TestZhilianClientHeaders:

    def test_sign_headers_injects_at_rt(self):
        signer = ZhilianSign(at="test_at", rt="test_rt")
        client = ZhilianClient(signer=signer)
        headers = client.signer.sign_headers()
        assert headers["x-zp-at"] == "test_at"
        assert headers["x-zp-rt"] == "test_rt"

    def test_sign_headers_has_required_keys(self):
        client = ZhilianClient()
        headers = client.signer.sign_headers()
        for key in ["x-zp-at", "x-zp-rt", "x-zp-action-id", "x-zp-device-id"]:
            assert key in headers, f"缺少头信息: {key}"

    def test_default_signer_empty_tokens(self):
        client = ZhilianClient()
        headers = client.signer.sign_headers()
        assert headers["x-zp-at"] == ""
        assert headers["x-zp-rt"] == ""

    def test_sign_params_has_required_keys(self):
        client = ZhilianClient()
        params = client.signer.sign_params()
        for key in ["at", "rt", "channel", "platform", "version", "d"]:
            assert key in params, f"缺少签名参数: {key}"