ctms/backend/tests/test_study_lock_dependency.py

import uuid

import pytest
from fastapi import HTTPException
from starlette.requests import Request

from app.core import deps


class _Study:
    def __init__(self, study_id: uuid.UUID, is_locked: bool = False):
        self.id = study_id
        self.is_locked = is_locked


class _Category:
    def __init__(self, study_id: uuid.UUID | None):
        self.study_id = study_id


class _FaqItem:
    def __init__(self, study_id: uuid.UUID | None):
        self.study_id = study_id


class _ContractFee:
    def __init__(self, study_id: uuid.UUID | None):
        self.study_id = study_id


class _ContractPayment:
    def __init__(self, contract_fee_id: uuid.UUID):
        self.contract_fee_id = contract_fee_id


def _make_request(*, method: str = "POST", path: str = "/", query_string: bytes = b"", json_body: bytes = b"", path_params=None):
    async def receive():
        return {"type": "http.request", "body": json_body, "more_body": False}

    scope = {
        "type": "http",
        "http_version": "1.1",
        "method": method,
        "path": path,
        "raw_path": path.encode(),
        "query_string": query_string,
        "headers": [(b"content-type", b"application/json")],
        "path_params": path_params or {},
    }
    return Request(scope, receive)


@pytest.mark.asyncio
async def test_require_study_not_locked_accepts_study_id_from_request_body(monkeypatch):
    study_id = uuid.uuid4()

    async def fake_get(_db, requested_study_id):
        return _Study(requested_study_id, is_locked=False)

    monkeypatch.setattr("app.crud.study.get", fake_get)

    dependency = deps.require_study_not_locked()
    request = _make_request(json_body=f'{{"study_id":"{study_id}"}}'.encode())

    study = await dependency(request=request, db=object())

    assert study.id == study_id


@pytest.mark.asyncio
async def test_require_study_not_locked_resolves_study_id_from_category_path(monkeypatch):
    study_id = uuid.uuid4()
    category_id = uuid.uuid4()

    async def fake_get_category(_db, requested_category_id):
        assert requested_category_id == category_id
        return _Category(study_id)

    async def fake_get_study(_db, requested_study_id):
        return _Study(requested_study_id, is_locked=False)

    monkeypatch.setattr("app.crud.faq_category.get_category", fake_get_category)
    monkeypatch.setattr("app.crud.study.get", fake_get_study)

    dependency = deps.require_study_not_locked()
    request = _make_request(path=f"/api/v1/faqs/categories/{category_id}", path_params={"category_id": str(category_id)})

    study = await dependency(request=request, db=object())

    assert study.id == study_id


@pytest.mark.asyncio
async def test_require_study_not_locked_resolves_study_id_from_contract_path(monkeypatch):
    study_id = uuid.uuid4()
    contract_id = uuid.uuid4()

    async def fake_get_contract(_db, requested_contract_id):
        assert requested_contract_id == contract_id
        return _ContractFee(study_id)

    async def fake_get_study(_db, requested_study_id):
        return _Study(requested_study_id, is_locked=False)

    monkeypatch.setattr("app.crud.contract_fee.get_contract_fee", fake_get_contract)
    monkeypatch.setattr("app.crud.study.get", fake_get_study)

    dependency = deps.require_study_not_locked()
    request = _make_request(
        path=f"/api/v1/fees/contracts/{contract_id}",
        path_params={"contract_id": str(contract_id)},
    )

    study = await dependency(request=request, db=object())

    assert study.id == study_id


@pytest.mark.asyncio
async def test_require_study_not_locked_resolves_study_id_from_contract_payment_path(monkeypatch):
    study_id = uuid.uuid4()
    contract_id = uuid.uuid4()
    payment_id = uuid.uuid4()

    async def fake_get_payment(_db, requested_payment_id):
        assert requested_payment_id == payment_id
        return _ContractPayment(contract_id)

    async def fake_get_contract(_db, requested_contract_id):
        assert requested_contract_id == contract_id
        return _ContractFee(study_id)

    async def fake_get_study(_db, requested_study_id):
        return _Study(requested_study_id, is_locked=False)

    monkeypatch.setattr("app.crud.contract_fee_payment.get_payment", fake_get_payment)
    monkeypatch.setattr("app.crud.contract_fee.get_contract_fee", fake_get_contract)
    monkeypatch.setattr("app.crud.study.get", fake_get_study)

    dependency = deps.require_study_not_locked()
    request = _make_request(
        path=f"/api/v1/fees/payments/{payment_id}",
        path_params={"payment_id": str(payment_id)},
    )

    study = await dependency(request=request, db=object())

    assert study.id == study_id


@pytest.mark.asyncio
async def test_require_study_not_locked_rejects_locked_study_resolved_from_item_path(monkeypatch):
    study_id = uuid.uuid4()
    item_id = uuid.uuid4()

    async def fake_get_item(_db, requested_item_id):
        assert requested_item_id == item_id
        return _FaqItem(study_id)

    async def fake_get_study(_db, requested_study_id):
        return _Study(requested_study_id, is_locked=True)

    monkeypatch.setattr("app.crud.faq_item.get_item", fake_get_item)
    monkeypatch.setattr("app.crud.study.get", fake_get_study)

    dependency = deps.require_study_not_locked()
    request = _make_request(path=f"/api/v1/faqs/{item_id}", path_params={"item_id": str(item_id)})

    with pytest.raises(HTTPException) as exc_info:
        await dependency(request=request, db=object())

    assert exc_info.value.status_code == 403
    assert "项目已锁定" in exc_info.value.detail