merge(HEL-192): 集成问天解势修复到 .11 正式线(含 HEL-183 顶栏修复)

This commit is contained in:
总工
2026-08-27 14:58:11 +00:00
9 changed files with 376 additions and 13 deletions
+150 -1
View File
@@ -1,13 +1,162 @@
from __future__ import annotations
import json
import tempfile
import unittest
from http import HTTPStatus
from pathlib import Path
from unittest.mock import patch
from backend.features.heaven.engine import build_five_phase_field, hexagram_from_lines
from backend.features.heaven.knowledge import prepare_heaven_context
from backend.features.heaven.http import HeavenHttpMixin
from backend.features.heaven.knowledge import (
HeavenKnowledgeError,
clear_heaven_knowledge_cache,
prepare_heaven_context,
resolve_heaven_knowledge_path,
_knowledge_catalog,
)
from backend.features.heaven.six_yao import build_six_yao_chart
class HeavenKnowledgeTests(unittest.TestCase):
def tearDown(self) -> None:
clear_heaven_knowledge_cache()
def test_catalog_loads_from_trusted_repo_file(self):
clear_heaven_knowledge_cache()
path = resolve_heaven_knowledge_path()
catalog = _knowledge_catalog()
self.assertTrue(path.is_file())
self.assertEqual(path.name, "heaven_knowledge.json")
self.assertTrue(str(catalog.get("version") or "").startswith("2026."))
self.assertIn("zhouyi", catalog["sources"])
self.assertIn("neijing", catalog["sources"])
self.assertEqual(len(catalog["fortune"]["qi"]), 6)
self.assertEqual(len(catalog["fortune"]["personal_relations"]), 10)
def test_missing_knowledge_file_raises_chinese_structured_error(self):
clear_heaven_knowledge_cache()
missing = Path(tempfile.mkdtemp()) / "missing-heaven_knowledge.json"
with patch(
"backend.features.heaven.knowledge.KNOWLEDGE_FILE", missing
), patch(
"backend.features.heaven.knowledge.KNOWLEDGE_SEED_FILE",
missing.with_name("missing-seed.json"),
):
with self.assertRaises(HeavenKnowledgeError) as raised:
_knowledge_catalog()
self.assertEqual(raised.exception.error_code, "heaven_knowledge_missing")
self.assertIn("缺失", str(raised.exception))
def test_corrupt_knowledge_json_raises_chinese_structured_error(self):
clear_heaven_knowledge_cache()
with tempfile.TemporaryDirectory() as temp_dir:
broken = Path(temp_dir) / "heaven_knowledge.json"
broken.write_text("{not-json", encoding="utf-8")
with patch(
"backend.features.heaven.knowledge.KNOWLEDGE_FILE", broken
), patch(
"backend.features.heaven.knowledge.KNOWLEDGE_SEED_FILE",
Path(temp_dir) / "unused-seed.json",
):
with self.assertRaises(HeavenKnowledgeError) as raised:
_knowledge_catalog()
self.assertEqual(raised.exception.error_code, "heaven_knowledge_invalid")
self.assertIn("损坏", str(raised.exception))
def test_interpret_http_returns_structured_chinese_error_for_missing_file(self):
class FakeHandler(HeavenHttpMixin):
def __init__(self) -> None:
self.payload = None
self.status = None
self.application_service = type(
"Svc",
(),
{
"heaven_interpret": staticmethod(
lambda _body: (_ for _ in ()).throw(
HeavenKnowledgeError(
"问天知识文件缺失:未找到 heaven_knowledge.json。",
code="heaven_knowledge_missing",
)
)
)
},
)()
def read_json_body(self):
return {"mode": "trend", "trade_date": "2026-08-04"}
def send_json(self, payload, status=HTTPStatus.OK, headers=None):
self.payload = payload
self.status = status
handler = FakeHandler()
handler.heaven_interpret()
self.assertEqual(handler.status, HTTPStatus.BAD_REQUEST)
self.assertIn("缺失", handler.payload["error"])
self.assertEqual(handler.payload["code"], "heaven_knowledge_missing")
def test_interpret_http_returns_structured_chinese_error_for_corrupt_json(self):
class FakeHandler(HeavenHttpMixin):
def __init__(self) -> None:
self.payload = None
self.status = None
self.application_service = type(
"Svc",
(),
{
"heaven_interpret": staticmethod(
lambda _body: (_ for _ in ()).throw(
HeavenKnowledgeError(
"问天知识文件 JSON 损坏(heaven_knowledge.json),无法解析:第 1 行附近。",
code="heaven_knowledge_invalid",
)
)
)
},
)()
def read_json_body(self):
return {"mode": "trend", "trade_date": "2026-08-04"}
def send_json(self, payload, status=HTTPStatus.OK, headers=None):
self.payload = payload
self.status = status
handler = FakeHandler()
handler.heaven_interpret()
self.assertEqual(handler.status, HTTPStatus.BAD_REQUEST)
self.assertIn("损坏", handler.payload["error"])
self.assertEqual(handler.payload["code"], "heaven_knowledge_invalid")
def test_seed_fallback_when_data_file_missing(self):
clear_heaven_knowledge_cache()
with tempfile.TemporaryDirectory() as temp_dir:
seed = Path(temp_dir) / "seed.json"
seed.write_text(
json.dumps(
{
"version": "test-seed",
"sources": {"zhouyi": {"title": "周易"}},
"trend": {"method": "m", "rules": {"stable": "s", "single": "a", "multiple": "b"}},
"fortune": {},
"heart": {},
},
ensure_ascii=False,
),
encoding="utf-8",
)
missing_data = Path(temp_dir) / "data-heaven_knowledge.json"
with patch(
"backend.features.heaven.knowledge.KNOWLEDGE_FILE", missing_data
), patch(
"backend.features.heaven.knowledge.KNOWLEDGE_SEED_FILE", seed
):
catalog = _knowledge_catalog()
self.assertEqual(catalog["version"], "test-seed")
def test_fortune_context_excludes_weighted_summary_and_adds_bounded_industry_symbols(self):
field = build_five_phase_field("2026-08-04")
prepared = prepare_heaven_context(