medical_training/test/test_case_negative.py

"""病例域负向测试：字段校验、越权、限流、事务回滚、AI Schema 违规。"""

from unittest.mock import patch

from django.core.cache import cache
from django.db import IntegrityError
from rest_framework.test import APIClient

from apps.case.models import CaseBase, ScoringRule
from apps.user.throttling import PdfParseUserThrottle
from config.exceptions import AppError
from .conftest import (
    CacheTestCase, CacheTransactionTestCase,
    CASE_PARSE_URL, CASE_FULL_CREATE_URL,
    case_full_url, create_test_user, get_auth_client, ensure_department,
    build_traditional_payload, make_fake_pdf,
)


class CaseFieldValidationTest(CacheTestCase):
    """病例字段校验负向测试。"""

    def setUp(self):
        super().setUp()
        self.user = create_test_user(phone='13800002001', password='CaseNeg1')
        self.client = get_auth_client(self.user)
        ensure_department('儿科')

    def test_invalid_case_type_400(self):
        """N10: case_type='invalid' → 400 CASE_TYPE_NOT_SUPPORTED"""
        payload = build_traditional_payload()
        payload['case_type'] = 'invalid'
        resp = self.client.post(CASE_FULL_CREATE_URL, payload, format='json')
        self.assertEqual(resp.status_code, 400, resp.content)
        self.assertEqual(resp.json()['code'], 'CASE_TYPE_NOT_SUPPORTED')

    def test_empty_scoring_rules_400(self):
        """N11: scoring_rules=[] → 400 CASE_VALIDATION_ERROR"""
        payload = build_traditional_payload()
        payload['scoring_rules'] = []
        resp = self.client.post(CASE_FULL_CREATE_URL, payload, format='json')
        self.assertEqual(resp.status_code, 400, resp.content)
        self.assertEqual(resp.json()['code'], 'CASE_VALIDATION_ERROR')

    def test_subtable_conflict_400(self):
        """N12: 同时传 traditional + teaching → 400 CASE_SUBTYPE_CONFLICT"""
        payload = build_traditional_payload()
        payload['teaching'] = {
            'teaching_goal': '不应该出现',
            'discussion_questions': '冲突',
        }
        resp = self.client.post(CASE_FULL_CREATE_URL, payload, format='json')
        self.assertEqual(resp.status_code, 400, resp.content)
        self.assertEqual(resp.json()['code'], 'CASE_SUBTYPE_CONFLICT')

    def test_duplicate_exam_items_deduped_on_create(self):
        """同一病例重复 item_code：归一化后只保留一条并成功创建。"""
        payload = build_traditional_payload(with_exam_items=True)
        self.assertIn('exam_items', payload)
        payload['exam_items'].append({
            **payload['exam_items'][0],
            'item_name': '重复血常规',
            'result_text': '应被忽略',
        })
        resp = self.client.post(CASE_FULL_CREATE_URL, payload, format='json')
        self.assertEqual(resp.status_code, 201, resp.content)
        self.assertEqual(len(resp.json()['exam_items']), 1)

    def test_missing_subtable_400(self):
        """N13: case_type=traditional 但无 traditional 子表 → 400 CASE_SUBTYPE_REQUIRED"""
        payload = build_traditional_payload()
        del payload['traditional']
        resp = self.client.post(CASE_FULL_CREATE_URL, payload, format='json')
        self.assertEqual(resp.status_code, 400, resp.content)
        self.assertEqual(resp.json()['code'], 'CASE_SUBTYPE_REQUIRED')

    def test_patch_published_case_400(self):
        """N16: PATCH 已发布病例 → 400 CASE_NOT_EDITABLE"""
        # 先创建病例
        payload = build_traditional_payload()
        resp = self.client.post(CASE_FULL_CREATE_URL, payload, format='json')
        self.assertEqual(resp.status_code, 201, resp.content)
        case_id = resp.json()['case']['id']

        # 发布
        case = CaseBase.objects.get(id=case_id)
        case.publish_status = 1
        case.save(update_fields=['publish_status'])

        # PATCH → 应被拒绝
        resp = self.client.patch(case_full_url(case_id), {
            'title': '不应该成功',
        }, format='json')
        self.assertEqual(resp.status_code, 400, resp.content)
        self.assertEqual(resp.json()['code'], 'CASE_NOT_EDITABLE')


class CaseAuthorizationTest(CacheTestCase):
    """病例越权测试。"""

    def setUp(self):
        super().setUp()
        ensure_department('儿科')

    def test_unauth_full_create_401(self):
        """N14: 未登录 POST full-create → 401"""
        client = APIClient()
        payload = build_traditional_payload()
        resp = client.post(CASE_FULL_CREATE_URL, payload, format='json')
        self.assertEqual(resp.status_code, 401, resp.content)

    def test_view_other_draft_403(self):
        """N15: 用户 B 看用户 A 的草稿 → 403"""
        # 用户 A 创建草稿
        user_a = create_test_user(phone='13800002010', password='UserAPass1')
        client_a = get_auth_client(user_a)
        payload = build_traditional_payload()
        resp = client_a.post(CASE_FULL_CREATE_URL, payload, format='json')
        self.assertEqual(resp.status_code, 201, resp.content)
        case_id = resp.json()['case']['id']

        # 验证是草稿 (publish_status=0)
        self.assertEqual(resp.json()['case']['publish_status'], 0)

        # 用户 B 尝试访问
        user_b = create_test_user(phone='13800002011', password='UserBPass1')
        client_b = get_auth_client(user_b)
        resp = client_b.get(case_full_url(case_id))
        self.assertEqual(resp.status_code, 403, resp.content)
        self.assertEqual(resp.json()['code'], 'CASE_PERMISSION_DENIED')


class CaseRateLimitTest(CacheTestCase):
    """病例限流测试。"""

    def setUp(self):
        super().setUp()
        self.user = create_test_user(phone='13800002020', password='RateTest1')
        self.client = get_auth_client(self.user)

    def test_rate_limit_pdf_parse_429(self):
        """N17: PDF 解析限流 → 429"""
        with (
            patch.object(PdfParseUserThrottle, 'allow_request', return_value=False),
            patch.object(PdfParseUserThrottle, 'wait', return_value=60),
        ):
            fake_pdf = make_fake_pdf()
            resp = self.client.post(
                CASE_PARSE_URL,
                {'files': fake_pdf, 'case_type': 'traditional'},
                format='multipart',
            )
        self.assertEqual(resp.status_code, 429, resp.content)
        self.assertEqual(resp.json()['code'], 'SYS_RATE_LIMIT')


class CaseTransactionRollbackTest(CacheTransactionTestCase):
    """N18: 事务回滚测试 — 必须用 TransactionTestCase。"""

    def setUp(self):
        super().setUp()
        self.user = create_test_user(phone='13800002030', password='TxTest1')
        self.client = get_auth_client(self.user)
        ensure_department('儿科')

    def test_transaction_rollback(self):
        """N18: bulk_create 失败 → CaseBase 应回滚"""
        initial_count = CaseBase.objects.count()

        with patch.object(
            ScoringRule.objects, 'bulk_create',
            side_effect=IntegrityError('mocked DB error'),
        ):
            payload = build_traditional_payload()
            resp = self.client.post(CASE_FULL_CREATE_URL, payload, format='json')

        # 请求应失败（500 或被异常处理捕获）
        self.assertGreaterEqual(resp.status_code, 400)

        # CaseBase 应回滚到初始状态
        self.assertEqual(
            CaseBase.objects.count(), initial_count,
            'CaseBase 应因事务回滚而未增加',
        )


class CaseAISchemaTest(CacheTestCase):
    """AI Schema 违规测试。"""

    def setUp(self):
        super().setUp()
        self.user = create_test_user(phone='13800002040', password='AITest1')
        self.client = get_auth_client(self.user)

    @patch('apps.case.services.case_importer.extract_text_from_pdfs',
           return_value='虚拟文本内容')
    def test_ai_bad_json_500(self, mock_pdf):
        """N19: DeepSeek 返回非法 JSON → 500 AI_BAD_JSON"""
        with (
            patch(
                'apps.case.services.deepseek_client.call_deepseek',
                side_effect=AppError('AI_BAD_JSON', 'AI 返回非合法 JSON', status_code=500),
            ),
            patch.object(PdfParseUserThrottle, 'allow_request', return_value=True),
        ):
            fake_pdf = make_fake_pdf()
            resp = self.client.post(
                CASE_PARSE_URL,
                {'files': fake_pdf, 'case_type': 'traditional'},
                format='multipart',
            )
        self.assertEqual(resp.status_code, 500, resp.content)
        self.assertEqual(resp.json()['code'], 'AI_BAD_JSON')

    @patch('apps.case.services.case_importer.extract_text_from_pdfs',
           return_value='虚拟文本内容')
    def test_ai_schema_violation_500(self, mock_pdf):
        """N20: DeepSeek 输出不符合 JSON Schema → 500 AI_SCHEMA_VIOLATION"""
        # 返回缺少 title 和 case_type 的数据 → jsonschema 校验失败
        bad_result = {
            'data': {
                'wrong_field': 'no title here',
            },
            'usage': {'prompt_tokens': 10, 'completion_tokens': 20},
        }
        with (
            patch('apps.case.services.deepseek_client.call_deepseek', return_value=bad_result),
            patch.object(PdfParseUserThrottle, 'allow_request', return_value=True),
        ):
            fake_pdf = make_fake_pdf()
            resp = self.client.post(
                CASE_PARSE_URL,
                {'files': fake_pdf, 'case_type': 'traditional'},
                format='multipart',
            )
        self.assertEqual(resp.status_code, 500, resp.content)
        self.assertEqual(resp.json()['code'], 'AI_SCHEMA_VIOLATION')