from __future__ import annotations import tempfile import threading import unittest from pathlib import Path from unittest.mock import patch from database import ReviewDatabase from backend.features.heaven.agent import HEAVEN_PROMPT_VERSIONS from server import DashboardService class HeavenReadingTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.database = ReviewDatabase(Path(self.temp.name) / "review.db") self.owner = self.database.create_user("heaven_owner", "salt", "hash") self.other = self.database.create_user("heaven_other", "salt", "hash") def tearDown(self) -> None: self.temp.cleanup() def save(self, user_id: int, dedupe_key: str = "fortune:20260723") -> dict: return self.database.save_heaven_reading( user_id, "fortune", "20260723", "2026-07-23 观气", "丙午年 · 乙未月 · 己丑日", "当日解运结果", { "calendar_date": "20260723", "interpretation_version": HEAVEN_PROMPT_VERSIONS["fortune"], }, dedupe_key, ) def test_history_is_private_and_delete_requires_ownership(self): reading = self.save(self.owner["id"]) self.assertEqual( self.database.list_heaven_readings(self.owner["id"], "fortune")[0]["answer"], "当日解运结果", ) self.assertEqual( self.database.list_heaven_readings(self.other["id"], "fortune"), [] ) self.assertFalse( self.database.delete_heaven_reading(self.other["id"], reading["id"]) ) self.assertTrue( self.database.delete_heaven_reading(self.owner["id"], reading["id"]) ) def test_fortune_dedupe_keeps_the_first_successful_result(self): first = self.save(self.owner["id"]) second = self.database.save_heaven_reading( self.owner["id"], "fortune", "20260723", "replacement", "replacement", "不应覆盖", {}, "fortune:20260723", ) self.assertEqual(second["id"], first["id"]) self.assertEqual(second["answer"], "当日解运结果") def test_fortune_interpretation_reuses_saved_result_before_llm(self): existing = self.save(self.owner["id"]) service = DashboardService.__new__(DashboardService) service.database = self.database service._request_context = threading.local() service._request_context.user_id = self.owner["id"] with patch.object(service, "heaven_setup") as setup, patch.object( service, "_call_heaven_agent" ) as call_agent: result = service.heaven_interpret( {"mode": "fortune", "trade_date": "2026-07-23"} ) setup.assert_not_called() call_agent.assert_not_called() self.assertTrue(result["reused"]) self.assertEqual(result["reading"]["id"], existing["id"]) self.assertEqual(result["answer"], "当日解运结果") def test_setup_only_exposes_current_fortune_prompt_version(self): current = self.save(self.owner["id"]) service = DashboardService.__new__(DashboardService) service.database = self.database service._request_context = threading.local() service._request_context.user_id = self.owner["id"] self.assertEqual( service._reusable_daily_fortune_reading("20260723")["id"], current["id"], ) self.database.save_heaven_reading( self.owner["id"], "fortune", "20260724", "2026-07-24 观气", "旧版本", "旧提示词生成的结果", {"interpretation_version": "heaven-fortune-v2"}, "fortune:20260724", ) self.assertIsNone(service._reusable_daily_fortune_reading("20260724")) def test_legacy_truncated_fortune_is_regenerated_once(self): self.database.save_heaven_reading( self.owner["id"], "fortune", "20260723", "2026-07-23 观气", "丙午年 · 乙未月 · 己丑日", "旧版解运结果被截断……", {"calendar_date": "20260723"}, "fortune:20260723", ) service = DashboardService.__new__(DashboardService) service.database = self.database service._request_context = threading.local() service._request_context.user_id = self.owner["id"] setup = { "calendar_date": "20260723", "field": { "sector_catalog": [], "balance": [{"element": "火"}], "pillars": {"year": "丙午", "month": "乙未", "day": "己丑"}, }, } with patch.object(service, "heaven_setup", return_value=setup), patch.object( service, "account_personal_field", return_value={} ), patch.object( service, "_call_heaven_agent", return_value=({"answer": "完整的当日解运结果", "model": "test", "latency_ms": 1}, "primary"), ): result = service.heaven_interpret( {"mode": "fortune", "trade_date": "2026-07-23"} ) self.assertFalse(result["reused"]) self.assertEqual(result["reading"]["answer"], "完整的当日解运结果") self.assertEqual( len(self.database.list_heaven_readings(self.owner["id"], "fortune")), 1 ) def test_stale_prompt_version_is_replaced_after_successful_regeneration(self): existing = self.database.save_heaven_reading( self.owner["id"], "fortune", "20260723", "2026-07-23 观气", "丙午年 · 乙未月 · 己丑日", "旧提示词生成的完整结果", {"interpretation_version": "heaven-fortune-v2"}, "fortune:20260723", ) service = DashboardService.__new__(DashboardService) service.database = self.database service._request_context = threading.local() service._request_context.user_id = self.owner["id"] setup = { "calendar_date": "20260723", "field": { "sector_catalog": [], "balance": [{"element": "火"}], "pillars": {"year": "丙午", "month": "乙未", "day": "己丑"}, }, } with patch.object(service, "heaven_setup", return_value=setup), patch.object( service, "account_personal_field", return_value={} ), patch.object( service, "_call_heaven_agent", return_value=({"answer": "新版提示词生成的完整结果", "model": "test", "latency_ms": 1}, "primary"), ): result = service.heaven_interpret( {"mode": "fortune", "trade_date": "2026-07-23"} ) self.assertFalse(result["reused"]) self.assertEqual(result["reading"]["id"], existing["id"] + 1) self.assertEqual(result["answer"], "新版提示词生成的完整结果") self.assertEqual( self.database.heaven_reading_interpretation_version( self.owner["id"], result["reading"]["id"] ), HEAVEN_PROMPT_VERSIONS["fortune"], ) self.assertEqual( len(self.database.list_heaven_readings(self.owner["id"], "fortune")), 1 ) def test_trend_interpretation_saves_successful_agent_result(self): service = DashboardService.__new__(DashboardService) service.database = self.database service._request_context = threading.local() service._request_context.user_id = self.owner["id"] setup = { "trade_date": "20260723", "chart": { "available": True, "sector": "银行", "stock": {"code": "000001", "name": "平安银行"}, "hexagram": { "name": "中孚", "lines": [], "transformed": {"name": "小畜"}, }, "movement": {}, }, } with patch.object(service, "heaven_setup", return_value=setup), patch.object( service, "_call_heaven_agent", return_value=( {"answer": "完整的解势结果", "model": "test", "latency_ms": 1}, "primary", ), ): result = service.heaven_interpret( { "mode": "trend", "trade_date": "2026-07-23", "stock_code": "000001", } ) self.assertFalse(result["reused"]) self.assertEqual(result["answer"], "完整的解势结果") self.assertEqual(result["reading"]["subject"], "000001 平安银行") self.assertEqual( self.database.list_heaven_readings(self.owner["id"], "trend")[0]["answer"], "完整的解势结果", ) def test_heart_interpretation_sends_question_and_deterministic_najia(self): service = DashboardService.__new__(DashboardService) service.database = self.database service._request_context = threading.local() service._request_context.user_id = self.owner["id"] with patch.object( service, "_call_heaven_agent", return_value=( {"answer": "完整的观心解读，包含本卦、世应、动变与针对所问的条件化判断。", "model": "test", "latency_ms": 1}, "primary", ), ) as call_agent: result = service.heaven_interpret( { "mode": "heart", "trade_date": "2026-08-04", "lines": [6, 7, 8, 9, 7, 8], "question": "这笔交易最需要防范什么？", "question_preset": "trade", "cast_at": "2026-08-04T10:30:00+08:00", } ) agent_context = call_agent.call_args.args[1] calculation = agent_context["calculation"] self.assertEqual(calculation["question"], "这笔交易最需要防范什么？") self.assertEqual(calculation["question_preset"], "trade") self.assertEqual(calculation["six_yao"]["method"], "京房纳甲·八宫世应") self.assertIn("这笔交易最需要防范什么", result["reading"]["subject_detail"]) def test_fallback_notice_describes_the_backup_service(self): service = DashboardService.__new__(DashboardService) service.database = self.database service._request_context = threading.local() service._request_context.user_id = self.owner["id"] with patch.object( service, "_call_heaven_agent", return_value=( {"answer": "完整的观心解读，说明所问、卦象依据、动变与可验证条件。", "model": "backup", "latency_ms": 1}, "fallback", ), ): result = service.heaven_interpret( { "mode": "heart", "trade_date": "2026-08-05", "lines": [6, 7, 8, 9, 7, 8], "question": "这笔交易最需要防范什么？", "question_preset": "trade", "cast_at": "2026-08-05T10:30:00+08:00", } ) self.assertEqual(result["notice"], "当前智能服务繁忙，已自动切换备用服务。") if __name__ == "__main__": unittest.main()