""" 单元测试 - 儿童心理筛查器 测试场景：使用虚构的儿童对话上下文，验证筛查函数能否正确识别并标注心理问题 运行方式： pytest tests/test_screener.py -v """ import pytest import json from unittest.mock import patch, MagicMock from psycho_screener.screener import ( PsychoScreener, ScreeningResult, ConcernCategory, ) # ============================================================================ # Mock API 响应（用于测试） # ============================================================================ MOCK_RESPONSES = { # 场景1：霸凌/同伴冲突 "bullying": json.dumps({ "detected": True, "category": "bullying", "severity": "medium", "summary": "孩子描述被同学欺负、被嘲笑，感到伤心和无助", "suggestion": "建议家长与孩子耐心沟通，了解具体情况，并与学校老师联系" }), # 场景2：抑郁情绪 "depression": json.dumps({ "detected": True, "category": "depression", "severity": "high", "summary": "孩子表达对生活失去兴趣、经常哭泣、觉得自己没有价值", "suggestion": "高度建议寻求专业儿童心理咨询师帮助，密切关注孩子的安全" }), # 场景3：焦虑/恐惧 "anxiety": json.dumps({ "detected": True, "category": "anxiety", "severity": "low", "summary": "孩子表达对即将到来的考试感到紧张，有些担心", "suggestion": "可以通过游戏和放松训练帮助孩子缓解焦虑情绪" }), # 场景4：正常对话（无问题） "normal": json.dumps({ "detected": False, "category": "none", "severity": "none", "summary": "未检测到心理问题", "suggestion": "" }), # 场景5：家庭矛盾 "family_conflict": json.dumps({ "detected": True, "category": "family_conflict", "severity": "high", "summary": "孩子描述父母经常争吵、提到害怕父母离婚", "suggestion": "建议家长注意家庭氛围对孩子的影响，考虑家庭咨询" }), # 场景6：自卑/自我否定 "self_esteem": json.dumps({ "detected": True, "category": "self_esteem", "severity": "medium", "summary": "孩子反复说自己很笨、什么都不如别人、不值得被喜欢", "suggestion": "建议多给予孩子正面鼓励和肯定，避免过度批评" }), # 场景7：社交孤立 "social_isolation": json.dumps({ "detected": True, "category": "social_isolation", "severity": "medium", "summary": "孩子表达在学校没有朋友、午餐时间独自一人、感到孤独", "suggestion": "建议帮助孩子学习社交技巧，与学校合作创造更多交友机会" }), } # ============================================================================ # 测试用例：虚构儿童对话上下文 # ============================================================================ TEST_CASES = [ # ---- 霸凌/同伴冲突 ---- { "case_id": "bullying_01", "scenario": "同伴冲突", "context": ( "孩子：今天在学校，小明又打我了，他抢了我的玩具还不让我告诉老师，" "他说如果我告诉老师就会打我。我很害怕，我现在不想去学校了。\n" "孩子：每次下课我都躲起来，因为我怕再被他欺负。\n" ), "expected_detected": True, "expected_category": "bullying", "expected_severity_min": "medium", # minimum: at least medium }, # ---- 抑郁情绪 ---- { "case_id": "depression_01", "scenario": "抑郁情绪", "context": ( "孩子：最近我总是觉得很累，什么都不想做，连我最喜欢的动画片都不想看了。\n" "孩子：我觉得我是一个很糟糕的人，没有人喜欢我。\n" "孩子：有时候我想，要是我不在了就好了。\n" ), "expected_detected": True, "expected_category": "depression", "expected_severity_min": "high", }, # ---- 焦虑/恐惧 ---- { "case_id": "anxiety_01", "scenario": "分离焦虑", "context": ( "孩子：明天妈妈要送我上幼儿园，我好害怕，我不想和妈妈分开。\n" "孩子：我担心在幼儿园里会做噩梦，害怕一个人睡觉。\n" ), "expected_detected": True, "expected_category": "anxiety", "expected_severity_min": "low", }, # ---- 正常对话 ---- { "case_id": "normal_01", "scenario": "正常对话", "context": ( "孩子：今天我画了一幅画，是一只大恐龙！\n" "孩子：晚上吃的是我最喜欢的红烧肉，好开心呀！\n" "孩子：明天我想和好朋友一起去公园玩。\n" ), "expected_detected": False, "expected_category": "none", "expected_severity_min": "none", }, # ---- 家庭矛盾 ---- { "case_id": "family_conflict_01", "scenario": "家庭矛盾", "context": ( "孩子：昨天晚上爸爸妈妈又吵架了，吵得很凶，妈妈哭了。\n" "孩子：我很害怕，怕他们会离婚。\n" "孩子：我觉得是我的错，是因为我表现不好他们才吵架的。\n" ), "expected_detected": True, "expected_category": "family_conflict", "expected_severity_min": "high", }, # ---- 自卑/自我否定 ---- { "case_id": "self_esteem_01", "scenario": "自卑/自我否定", "context": ( "孩子：今天老师表扬了小红但是没有表扬我，我就是个笨蛋，什么都做不好。\n" "孩子：班上的同学都不喜欢我，没有人想和我坐同桌。\n" "孩子：我什么都学不会，我好笨啊。\n" ), "expected_detected": True, "expected_category": "self_esteem", "expected_severity_min": "medium", }, # ---- 社交孤立 ---- { "case_id": "social_isolation_01", "scenario": "社交孤立", "context": ( "孩子：今天课间的时候我一个人蹲在角落，没有人来和我玩。\n" "孩子：同学们都有自己的朋友，只有我是一个人。\n" "孩子：我不想去学校了，那里好孤单。\n" ), "expected_detected": True, "expected_category": "social_isolation", "expected_severity_min": "medium", }, ] # ============================================================================ # Fixtures # ============================================================================ @pytest.fixture def api_key(): """测试用 API key（不会被真实调用）""" return "test-minimax-api-key-12345" @pytest.fixture def screener(api_key): """创建筛查器实例""" return PsychoScreener(api_key=api_key) # ============================================================================ # 测试：真实调用 MiniMax API（需要有效的 API key） # ============================================================================ @pytest.mark.integration class TestPsychoScreenerIntegration: """集成测试：真实调用 MiniMax API（需设置 MINIMAX_API_KEY）""" @pytest.fixture(autouse=True) def check_api_key(self): api_key = os.environ.get("MINIMAX_API_KEY", "") if not api_key: pytest.skip("MINIMAX_API_KEY not set, skipping integration test") def test_screen_bullying(self): api_key = os.environ.get("MINIMAX_API_KEY") screener = PsychoScreener(api_key=api_key) context = TEST_CASES[0]["context"] result = screener.screen(context) print(f"\n[integration] bullying result: {result.model_dump_json(indent=2)}") assert result.detected is True assert result.category == "bullying" def test_screen_normal(self): api_key = os.environ.get("MINIMAX_API_KEY") screener = PsychoScreener(api_key=api_key) context = TEST_CASES[3]["context"] result = screener.screen(context) print(f"\n[integration] normal result: {result.model_dump_json(indent=2)}") assert result.detected is False # ============================================================================ # 测试：Mock 测试（不真实调用 API） # ============================================================================ @pytest.mark.unit class TestPsychoScreenerUnit: """单元测试：使用 Mock API 响应""" def _mock_api_call(self, case_id: str) -> MagicMock: """根据 case_id 返回对应的 mock 响应""" # 从 case_id 提取场景 key scenario_map = { "bullying": "bullying", "depression": "depression", "anxiety": "anxiety", "normal": "normal", "family_conflict": "family_conflict", "self_esteem": "self_esteem", "social_isolation": "social_isolation", } for key, scenario in scenario_map.items(): if key in case_id: mock_resp = MagicMock() mock_resp.raise_for_status = MagicMock() mock_resp.json.return_value = { "choices": [{"message": {"content": MOCK_RESPONSES[scenario]}}] } return mock_resp # 默认返回 normal mock_resp = MagicMock() mock_resp.raise_for_status = MagicMock() mock_resp.json.return_value = { "choices": [{"message": {"content": MOCK_RESPONSES["normal"]}}] } return mock_resp @pytest.mark.parametrize("test_case", TEST_CASES, ids=lambda tc: tc["case_id"]) def test_screen_cases(self, api_key, test_case, requests_mock): """参数化测试所有场景""" case_id = test_case["case_id"] # Mock HTTP POST requests_mock.post( "https://api.minimaxi.com/v1/text/chatcompletion_v2", json=self._mock_api_call(case_id).json(), ) screener = PsychoScreener(api_key=api_key) result = screener.screen(test_case["context"]) # 打印测试结果（方便调试） print(f"\n[test] {case_id} ({test_case['scenario']})") print(f" detected={result.detected}, category={result.category}, severity={result.severity}") print(f" summary={result.summary}") # 验证检测结果 assert result.detected == test_case["expected_detected"], \ f"[{case_id}] 检测结果不符：expected {test_case['expected_detected']}, got {result.detected}" if test_case["expected_detected"]: assert result.category == test_case["expected_category"], \ f"[{case_id}] 类别不符：expected {test_case['expected_category']}, got {result.category}" # 验证严重程度（使用顺序比较） severity_order = ["none", "low", "medium", "high"] min_idx = severity_order.index(test_case["expected_severity_min"]) result_idx = severity_order.index(result.severity) assert result_idx >= min_idx, \ f"[{case_id}] 严重程度不足：expected >= {test_case['expected_severity_min']}, got {result.severity}" def test_build_response_prefix_detected(self, api_key): """测试：检测到问题时生成前缀""" screener = PsychoScreener(api_key=api_key) result = ScreeningResult( detected=True, category="bullying", severity="medium", summary="孩子描述被同学欺负", suggestion="建议与家长沟通", raw_response="", ) prefix = screener.build_response_prefix(result) print(f"\nprefix: {prefix}") assert "已发现特定心理问题" in prefix assert "bullying" in prefix assert "medium" in prefix def test_build_response_prefix_not_detected(self, api_key): """测试：未检测到问题时返回空前缀""" screener = PsychoScreener(api_key=api_key) result = ScreeningResult( detected=False, category="none", severity="none", summary="未检测到心理问题", suggestion="", raw_response="", ) prefix = screener.build_response_prefix(result) assert prefix == "" def test_json_parse_error_handling(self, api_key, requests_mock): """测试：模型返回非 JSON 时优雅处理""" requests_mock.post( "https://api.minimaxi.com/v1/text/chatcompletion_v2", text="This is not JSON response", ) screener = PsychoScreener(api_key=api_key) result = screener.screen("some context") # 应该优雅降级，不抛出异常 assert result.detected is False assert "API调用失败" in result.summary # ============================================================================ # 测试：数据结构验证 # ============================================================================ @pytest.mark.unit class TestScreeningResultModel: """测试 ScreeningResult 数据模型""" def test_screening_result_valid(self): """测试有效数据""" result = ScreeningResult( detected=True, category="bullying", severity="high", summary="test", suggestion="test", raw_response="test", ) assert result.detected is True assert result.category == "bullying" def test_screening_result_defaults(self): """测试默认值""" result = ScreeningResult(detected=False) assert result.detected is False assert result.category == "none" assert result.severity == "none" assert result.summary == "" # ============================================================================ # 测试：集成小智AI流程 # ============================================================================ @pytest.mark.integration class TestXiaozhiIntegration: """ 测试：与小智AI的集成流程 假设 xiaozhi-esp32-server 返回的对话上下文经过本筛查器， 验证端到端的处理流程 """ def test_full_flow_with_bullying(self): """ 模拟完整流程： 1. 孩子说了一段可能被霸凌的话 2. 筛查器识别出问题 3. 原有回复被加上前缀 """ # 模拟 xiaozhi 返回的原始回复 original_response = "不要难过的小朋友，每个人都值得被友好对待哦。" # 这里需要真实 API key api_key = os.environ.get("MINIMAX_API_KEY", "") if not api_key: pytest.skip("MINIMAX_API_KEY not set") screener = PsychoScreener(api_key=api_key) context = ( "孩子：今天小朋友又欺负我了，把我推倒在地上，还骂我。\n" "孩子：我觉得好委屈，为什么他们要这样对我？" ) # 步骤1：筛查 screening_result = screener.screen(context) print(f"\n[flow] screening result: {screening_result.model_dump_json(indent=2)}") # 步骤2：构建前缀 prefix = screener.build_response_prefix(screening_result) # 步骤3：覆盖原有回复 if prefix: final_response = f"{prefix}\n\n{original_response}" else: final_response = original_response print(f"\n[flow] final response:\n{final_response}") # 验证前缀存在 if screening_result.detected: assert "已发现特定心理问题" in final_response else: assert "已发现特定心理问题" not in final_response import os # for integration tests