leaudit-platform-backend/tests/test_qichacha_service.py

"""企查查服务测试。"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

import pytest


class _FakeSession:
    async def commit(self):
        return None

    async def refresh(self, obj):
        return obj

    async def flush(self):
        return None


class _FakeSessionContext:
    async def __aenter__(self):
        return _FakeSession()

    async def __aexit__(self, exc_type, exc, tb):
        return None


async def _unexpected_upsert(cls, session, **fields):
    pytest.fail("fresh cache should not upsert")


class _FakeClient:
    def __init__(self):
        self.calls: list[tuple[str, str]] = []

    async def QueryCompany(self, Keyword: str):
        self.calls.append(("company", Keyword))
        return (
            {"Name": "广州测试有限公司", "CreditCode": "91440000TEST"},
            {"VerifyResult": 1, "Data": [{"Anno": "案号1"}, {"Anno": "案号2"}]},
            "91440000TEST",
            "广州测试有限公司",
        )

    async def GetEnterpriseInfo(self, Keyword: str):
        self.calls.append(("enterprise", Keyword))
        return {"Name": "广州测试有限公司", "CreditCode": "91440000TEST"}

    async def GetDishonestyInfo(self, Keyword: str):
        self.calls.append(("dishonesty", Keyword))
        return {"VerifyResult": 0, "Data": []}


@pytest.mark.asyncio
async def test_query_company_uses_fresh_cache(monkeypatch):
    """未过期缓存应直接返回，不调用企查查 API。"""
    from fastapi_modules.fastapi_leaudit.models.qichachaCompanyInfo import QichachaCompanyInfo
    from fastapi_modules.fastapi_leaudit.services.impl.qichachaServiceImpl import QichachaServiceImpl

    now = datetime.now(UTC)
    record = SimpleNamespace(
        Id=1,
        searchKey="广州测试有限公司",
        creditCode="91440000TEST",
        companyName="广州测试有限公司",
        enterprise={"Name": "广州测试有限公司", "CreditCode": "91440000TEST"},
        dishonesty={"VerifyResult": 0, "Data": []},
        created_at=now,
        updated_at=now,
    )
    fake_client = _FakeClient()

    async def fake_find(cls, session, Keyword):
        return record

    monkeypatch.setattr(
        "fastapi_modules.fastapi_leaudit.services.impl.qichachaServiceImpl.GetAsyncSession",
        lambda: _FakeSessionContext(),
    )
    monkeypatch.setattr(QichachaCompanyInfo, "FindByKeyword", classmethod(fake_find))
    monkeypatch.setattr(QichachaCompanyInfo, "Upsert", classmethod(_unexpected_upsert))

    service = QichachaServiceImpl(Client=fake_client, CacheDays=30)
    result = await service.QueryCompany(Keyword="广州测试有限公司", ForceRefresh=False)

    assert result.success is True
    assert result.data is not None
    assert result.data.companyName == "广州测试有限公司"
    assert result.data.hasDishonesty is False
    assert result.data.dishonestyCount == 0
    assert fake_client.calls == []


@pytest.mark.asyncio
async def test_query_company_refreshes_when_forced(monkeypatch):
    """强制刷新时应调用企查查 API 并写入缓存。"""
    from fastapi_modules.fastapi_leaudit.models.qichachaCompanyInfo import QichachaCompanyInfo
    from fastapi_modules.fastapi_leaudit.services.impl.qichachaServiceImpl import QichachaServiceImpl

    old_record = SimpleNamespace(
        Id=1,
        searchKey="旧名称",
        creditCode=None,
        companyName=None,
        enterprise=None,
        dishonesty=None,
        created_at=datetime.now(UTC) - timedelta(days=1),
        updated_at=datetime.now(UTC) - timedelta(days=1),
    )
    saved: dict[str, object] = {}
    fake_client = _FakeClient()

    async def fake_find(cls, session, Keyword):
        return old_record

    async def fake_upsert(cls, session, **fields):
        saved.update(fields)
        return SimpleNamespace(
            Id=1,
            searchKey=fields["SearchKey"],
            creditCode=fields["CreditCode"],
            companyName=fields["CompanyName"],
            enterprise=fields["Enterprise"],
            dishonesty=fields["Dishonesty"],
            created_at=datetime.now(UTC),
            updated_at=datetime.now(UTC),
        )

    monkeypatch.setattr(
        "fastapi_modules.fastapi_leaudit.services.impl.qichachaServiceImpl.GetAsyncSession",
        lambda: _FakeSessionContext(),
    )
    monkeypatch.setattr(QichachaCompanyInfo, "FindByKeyword", classmethod(fake_find))
    monkeypatch.setattr(QichachaCompanyInfo, "Upsert", classmethod(fake_upsert))

    service = QichachaServiceImpl(Client=fake_client, CacheDays=30)
    result = await service.QueryCompany(Keyword="广州测试有限公司", ForceRefresh=True)

    assert fake_client.calls == [("company", "广州测试有限公司")]
    assert saved["SearchKey"] == "广州测试有限公司"
    assert saved["CreditCode"] == "91440000TEST"
    assert result.data is not None
    assert result.data.hasDishonesty is True
    assert result.data.dishonestyCount == 2


@pytest.mark.asyncio
async def test_query_company_refreshes_stale_cache(monkeypatch):
    """超过缓存天数的数据应自动刷新。"""
    from fastapi_modules.fastapi_leaudit.models.qichachaCompanyInfo import QichachaCompanyInfo
    from fastapi_modules.fastapi_leaudit.services.impl.qichachaServiceImpl import QichachaServiceImpl

    stale_record = SimpleNamespace(
        Id=1,
        searchKey="广州测试有限公司",
        creditCode="OLD",
        companyName="旧公司",
        enterprise={"Name": "旧公司"},
        dishonesty={"VerifyResult": 0, "Data": []},
        created_at=datetime.now(UTC) - timedelta(days=90),
        updated_at=datetime.now(UTC) - timedelta(days=90),
    )
    fake_client = _FakeClient()

    async def fake_find(cls, session, Keyword):
        return stale_record

    async def fake_upsert(cls, session, **fields):
        return SimpleNamespace(
            Id=1,
            searchKey=fields["SearchKey"],
            creditCode=fields["CreditCode"],
            companyName=fields["CompanyName"],
            enterprise=fields["Enterprise"],
            dishonesty=fields["Dishonesty"],
            created_at=datetime.now(UTC),
            updated_at=datetime.now(UTC),
        )

    monkeypatch.setattr(
        "fastapi_modules.fastapi_leaudit.services.impl.qichachaServiceImpl.GetAsyncSession",
        lambda: _FakeSessionContext(),
    )
    monkeypatch.setattr(QichachaCompanyInfo, "FindByKeyword", classmethod(fake_find))
    monkeypatch.setattr(QichachaCompanyInfo, "Upsert", classmethod(fake_upsert))

    service = QichachaServiceImpl(Client=fake_client, CacheDays=30)
    result = await service.QueryCompany(Keyword="广州测试有限公司")

    assert fake_client.calls == [("company", "广州测试有限公司")]
    assert result.data is not None
    assert result.data.companyName == "广州测试有限公司"