309 lines
12 KiB
Python
309 lines
12 KiB
Python
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()
|