Blame for conftest.py

"""
Pytest 配置文件
提供测试 fixtures
"""

import os
import sys
from typing import Any, Generator, Iterator

import django
import pytest

from apps.core.utils.path import Path

# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent / "apiSystem"))

# 设置 Django 配置
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "apiSystem.settings")

# 排除脚本文件（非标准 pytest 测试）
collect_ignore = [
    "tests/unit/automation/test_court_document.py",
]


@pytest.fixture(scope="session")
def django_db_setup(django_db_setup: Any, django_db_blocker: Any) -> Any:
    """
    设置测试数据库

    确保测试使用独立的测试数据库，不影响生产数据库
    """
    with django_db_blocker.unblock():
        # pytest-django 会自动创建测试数据库
        # 这里只需要确保使用正确的数据库配置
        from django.conf import settings

        db_cfg = settings.DATABASES.get("default", {})
        db_engine = str(db_cfg.get("ENGINE", ""))
        db_name = str(db_cfg.get("NAME", ""))

        if db_engine == "django.db.backends.sqlite3":
            is_test_db = (
                "test_" in db_name
                or "_test" in db_name
                or ":memory:" in db_name
                or "memorydb" in db_name
                or db_name == ":memory:"
            )
        else:
            lowered_name = db_name.lower()
            is_test_db = (
                lowered_name.startswith("test_")
                or lowered_name.endswith("_test")
                or "_test_" in lowered_name
                or lowered_name == "test"
            )

        assert is_test_db, f"错误：测试正在使用生产数据库 {db_name}！测试已中止。"

        yield

        # 测试结束后，pytest-django 会自动清理测试数据库


@pytest.fixture(scope="session")
def django_db_modify_db_settings() -> None:
    """
    修改测试数据库设置

    默认使用 PostgreSQL 测试库，也支持显式切换 SQLite。
    """
    from django.conf import settings

    inferred_engine = "sqlite" if (os.environ.get("DATABASE_PATH") or os.environ.get("TEST_DB_PATH")) else "postgresql"
    test_db_engine = (os.environ.get("TEST_DB_ENGINE") or os.environ.get("DB_ENGINE") or inferred_engine).strip().lower()

    if test_db_engine in ("sqlite", "sqlite3", "django.db.backends.sqlite3"):
        test_db_path = (os.environ.get("TEST_DB_PATH") or os.environ.get("DATABASE_PATH") or ":memory:").strip()
        settings.DATABASES["default"] = {
            "ENGINE": "django.db.backends.sqlite3",
            "NAME": test_db_path,
            "ATOMIC_REQUESTS": False,
            "CONN_MAX_AGE": 0,
            "TIME_ZONE": "Asia/Shanghai",
            "OPTIONS": {
                "timeout": 20,
            },
        }
        return

    raw_test_password = os.environ.get("TEST_DB_PASSWORD")
    raw_db_password = os.environ.get("DB_PASSWORD")
    resolved_password = raw_test_password if raw_test_password is not None else raw_db_password
    if resolved_password is None:
        resolved_password = "postgres"

    settings.DATABASES["default"] = {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": (os.environ.get("TEST_DB_NAME") or "fachuan_test").strip(),
        "USER": (os.environ.get("TEST_DB_USER") or os.environ.get("DB_USER") or "postgres").strip(),
        "PASSWORD": resolved_password.strip(),
        "HOST": (os.environ.get("TEST_DB_HOST") or os.environ.get("DB_HOST") or "127.0.0.1").strip(),
        "PORT": int((os.environ.get("TEST_DB_PORT") or os.environ.get("DB_PORT") or "5432").strip()),
        "ATOMIC_REQUESTS": False,
        "CONN_MAX_AGE": 0,
        "TIME_ZONE": "Asia/Shanghai",
        "CONN_HEALTH_CHECKS": True,
    }


@pytest.fixture
def api_client() -> Any:
    """提供 API 测试客户端"""
    from django.test import Client

    return Client()


@pytest.fixture
def authenticated_client(db: Any) -> Any:
    """提供已认证的测试客户端"""
    from django.test import Client

    from apps.organization.models import LawFirm, Lawyer

    client = Client()
    firm = LawFirm.objects.create(name="测试律所")
    user = Lawyer.objects.create_user(
        username="testuser",
        password="testpass123",
        is_admin=True,
        law_firm=firm,
    )
    client.force_login(user)
    return client


@pytest.fixture
def law_firm(db: Any) -> Any:
    """提供测试律所"""
    from apps.organization.models import LawFirm

    return LawFirm.objects.create(name="Fixture测试律所")


@pytest.fixture
def lawyer(db: Any, law_firm: Any) -> Any:
    """提供测试律师"""
    from apps.organization.models import Lawyer

    return Lawyer.objects.create_user(
        username="fixturelawyer",
        password="testpass123",
        real_name="Fixture律师",
        law_firm=law_firm,
    )


@pytest.fixture
def admin_lawyer(db: Any, law_firm: Any) -> Any:
    """提供管理员律师"""
    from apps.organization.models import Lawyer

    return Lawyer.objects.create_user(
        username="adminlawyer",
        password="testpass123",
        is_admin=True,
        law_firm=law_firm,
    )


@pytest.fixture
def client_entity(db: Any) -> Any:
    """提供测试客户"""
    from apps.client.models import Client

    return Client.objects.create(
        name="Fixture测试客户",
        client_type=Client.NATURAL,
        is_our_client=True,
    )


@pytest.fixture
def contract(db: Any, lawyer: Any) -> Any:
    """提供测试合同"""
    from apps.contracts.models import Contract

    return Contract.objects.create(
        name="Fixture测试合同",
        case_type="civil",
    )


@pytest.fixture
def case(db: Any, contract: Any) -> Any:
    """提供测试案件"""
    from apps.cases.models import Case

    return Case.objects.create(
        name="Fixture测试案件",
        contract=contract,
    )


# ========== Hypothesis 配置 ==========

from hypothesis import Verbosity, settings

# 配置 Hypothesis
settings.register_profile("default", max_examples=5, verbosity=Verbosity.normal)
settings.register_profile("ci", max_examples=1000, verbosity=Verbosity.verbose)
settings.register_profile("dev", max_examples=10, verbosity=Verbosity.verbose)
settings.register_profile("debug", max_examples=10, verbosity=Verbosity.debug)

# 加载配置（默认使用 default）
profile = os.getenv("HYPOTHESIS_PROFILE", "default")
settings.load_profile(profile)


# ========== 测试工具 Fixtures ==========


@pytest.fixture
def mock_contract_service() -> Any:
    """提供 Mock 合同服务"""
    from tests.mocks import MockContractService

    return MockContractService()


@pytest.fixture
def mock_case_service() -> Any:
    """提供 Mock 案件服务"""
    from tests.mocks import MockCaseService

    return MockCaseService()


@pytest.fixture
def mock_permission_service() -> Any:
    """提供 Mock 权限服务"""
    from tests.mocks import MockPermissionService

    return MockPermissionService()


@pytest.fixture
def mock_email_service() -> Any:
    """提供 Mock 邮件服务"""
    from tests.mocks import MockEmailService

    return MockEmailService()


@pytest.fixture
def query_counter(db: Any) -> Any:
    """
    查询计数器

    用于测试数据库查询优化，检测 N+1 查询问题

    使用方法：
        with query_counter() as counter:
            # 执行操作
            cases = Case.objects.select_related('contract').all()
            list(cases)

        assert counter.count <= 1  # 断言查询次数
    """
    from django.db import connection
    from django.test.utils import CaptureQueriesContext

    class QueryCounter:
        def __init__(self) -> None:
            self.context: CaptureQueriesContext | None = None
            self.count: int = 0

        def __enter__(self) -> "QueryCounter":
            self.context = CaptureQueriesContext(connection)
            self.context.__enter__()
            return self

        def __exit__(self, *args: Any) -> None:
            assert self.context is not None
            self.context.__exit__(*args)
            self.count = len(self.context.captured_queries)

        @property
        def queries(self) -> list[dict[str, Any]]:
            """获取所有查询"""
            return self.context.captured_queries if self.context else []

    def _counter() -> QueryCounter:
        return QueryCounter()

    return _counter


@pytest.fixture
def assert_num_queries(db: Any) -> Any:
    """
    断言查询次数

    使用方法：
        with assert_num_queries(1):
            # 执行操作，应该只有 1 次查询
            cases = Case.objects.select_related('contract').all()
            list(cases)
    """
    from django.db import connection
    from django.test.utils import CaptureQueriesContext

    class AssertNumQueries:
        def __init__(self, expected_count: int) -> None:
            self.expected_count = expected_count
            self.context: CaptureQueriesContext | None = None

        def __enter__(self) -> "AssertNumQueries":
            self.context = CaptureQueriesContext(connection)
            self.context.__enter__()
            return self

        def __exit__(self, *args: Any) -> None:
            assert self.context is not None
            self.context.__exit__(*args)
            actual_count = len(self.context.captured_queries)

            if actual_count != self.expected_count:
                queries = "\n".join(f"{i + 1}. {q['sql']}" for i, q in enumerate(self.context.captured_queries))
                raise AssertionError(f"Expected {self.expected_count} queries, but got {actual_count}:\n{queries}")

    return AssertNumQueries