xiaobaifupan/app/tests/test_heaven_readings.py

from __future__ import annotations

import tempfile
import threading
import unittest
from pathlib import Path
from unittest.mock import patch

from database import ReviewDatabase
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"},
            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_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_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"],
            "完整的解势结果",
        )


if __name__ == "__main__":
    unittest.main()