Jesse_ChenandClaude Opus 5.5
f3ed950982
bvr_002..004 (Sunaphaa/Anaphaa/Duradhara) and bvr_016..018 (Vesi/Vosi/ Ubhayachara) take upstream yinduzhanxing 2dd6062a's logic (Mars..Saturn only). kemadruma_yoga also ignores the nodes (product decision: classical definition; upstream still counts them). The six lunar/solar duplicates of upstream's RETIRED_RULE_IDS are retired; upstream's four other retirements are out of scope. TASK-upstream-sync4-20261002 T5. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N4f2nya58RoRu4yEmJgRGE
333 lines
14 KiB
Python
333 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""BUG-735: custom yoga 表达式的编译缓存。
|
||
|
||
三件事各自独立证明：
|
||
|
||
1. **等价**：对规则表里全部去重表达式，缓存版本与改前实现逐条同值。
|
||
证明方式是同进程差分（BUG-733 立下的做法）：参照实现在本文件里逐字复刻改前代码，
|
||
与生产实现在同一进程、同一批 exec_globals 上对跑，不写跨机 golden。
|
||
2. **只缓存 code object**：同一条表达式配不同的盘必须得到不同答案。
|
||
这条是红线——缓存求值结果会让所有人拿到同一张盘的 yoga 判定。
|
||
3. **编译次数**：第二次检测的源码编译次数为 0，不随请求数线性增长。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import builtins
|
||
import json
|
||
import sys
|
||
import types
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT / "scripts"))
|
||
RULES_PATH = ROOT / "references" / "yoga_rules.json"
|
||
|
||
import yoga_engine # noqa: E402
|
||
from yoga_engine import YogaContext, YogaEngine # noqa: E402
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 公开虚构示例盘。不是任何真实用户的出生资料。
|
||
# ---------------------------------------------------------------------------
|
||
SAMPLE_CHARTS = [
|
||
(
|
||
"Aries",
|
||
{
|
||
"Sun": {"house": 10, "sign": "Capricorn", "degree": 16.4},
|
||
"Moon": {"house": 4, "sign": "Cancer", "degree": 2.1},
|
||
"Mars": {"house": 12, "sign": "Pisces", "degree": 23.8},
|
||
"Mercury": {"house": 9, "sign": "Sagittarius", "degree": 27.3},
|
||
"Jupiter": {"house": 4, "sign": "Cancer", "degree": 8.9},
|
||
"Venus": {"house": 9, "sign": "Sagittarius", "degree": 13.6},
|
||
"Saturn": {"house": 10, "sign": "Capricorn", "degree": 20.2},
|
||
"Rahu": {"house": 8, "sign": "Scorpio", "degree": 1.5},
|
||
"Ketu": {"house": 2, "sign": "Taurus", "degree": 1.5},
|
||
},
|
||
),
|
||
(
|
||
"Leo",
|
||
{
|
||
"Sun": {"house": 5, "sign": "Gemini", "degree": 10.0},
|
||
"Moon": {"house": 5, "sign": "Gemini", "degree": 15.0},
|
||
"Mars": {"house": 8, "sign": "Pisces", "degree": 10.0},
|
||
"Mercury": {"house": 10, "sign": "Capricorn", "degree": 10.0},
|
||
"Jupiter": {"house": 11, "sign": "Sagittarius", "degree": 5.0},
|
||
"Venus": {"house": 2, "sign": "Libra", "degree": 10.0},
|
||
"Saturn": {"house": 7, "sign": "Aquarius", "degree": 10.0},
|
||
"Rahu": {"house": 6, "sign": "Capricorn", "degree": 10.0},
|
||
"Ketu": {"house": 12, "sign": "Cancer", "degree": 10.0},
|
||
},
|
||
),
|
||
(
|
||
"Libra",
|
||
{
|
||
"Sun": {"house": 1, "sign": "Libra", "degree": 22.0},
|
||
"Moon": {"house": 7, "sign": "Aries", "degree": 4.0},
|
||
"Mars": {"house": 4, "sign": "Capricorn", "degree": 28.0},
|
||
"Mercury": {"house": 2, "sign": "Scorpio", "degree": 3.0},
|
||
"Jupiter": {"house": 10, "sign": "Cancer", "degree": 17.0},
|
||
"Venus": {"house": 12, "sign": "Virgo", "degree": 9.0},
|
||
"Saturn": {"house": 1, "sign": "Libra", "degree": 6.0},
|
||
"Rahu": {"house": 3, "sign": "Sagittarius", "degree": 19.0},
|
||
"Ketu": {"house": 9, "sign": "Gemini", "degree": 19.0},
|
||
},
|
||
),
|
||
]
|
||
|
||
|
||
def _all_rule_expressions() -> list[str]:
|
||
"""规则表里全部 `expr`（去重后按文本排序，保证用例顺序稳定）。"""
|
||
data = json.loads(RULES_PATH.read_text(encoding="utf-8"))
|
||
found: list[str] = []
|
||
|
||
def walk(node: object) -> None:
|
||
if isinstance(node, dict):
|
||
value = node.get("expr")
|
||
if isinstance(value, str) and value.strip():
|
||
found.append(value)
|
||
for child in node.values():
|
||
walk(child)
|
||
elif isinstance(node, list):
|
||
for child in node:
|
||
walk(child)
|
||
|
||
walk(data)
|
||
return sorted(set(found))
|
||
|
||
|
||
RULE_EXPRESSIONS = _all_rule_expressions()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 参照实现：BUG-735 改前的 `_eval_custom` 尾部，逐字复刻。
|
||
# ---------------------------------------------------------------------------
|
||
def _reference_capture_tail_expr(block):
|
||
if not block:
|
||
return block
|
||
last = block[-1]
|
||
if isinstance(last, ast.Expr):
|
||
block[-1] = ast.Assign(targets=[ast.Name(id="__result__", ctx=ast.Store())], value=last.value)
|
||
elif isinstance(last, ast.If):
|
||
last.body = _reference_capture_tail_expr(last.body)
|
||
last.orelse = _reference_capture_tail_expr(last.orelse)
|
||
elif isinstance(last, (ast.For, ast.While)):
|
||
last.body = _reference_capture_tail_expr(last.body)
|
||
last.orelse = _reference_capture_tail_expr(last.orelse)
|
||
elif isinstance(last, ast.Try):
|
||
last.body = _reference_capture_tail_expr(last.body)
|
||
last.orelse = _reference_capture_tail_expr(last.orelse)
|
||
last.finalbody = _reference_capture_tail_expr(last.finalbody)
|
||
for handler in last.handlers:
|
||
handler.body = _reference_capture_tail_expr(handler.body)
|
||
return block
|
||
|
||
|
||
def _reference_run(expr, exec_globals):
|
||
"""改前行为：每次重新编译；先 eval，SyntaxError 退到改写过的多语句 exec。"""
|
||
result = None
|
||
try:
|
||
result = eval(expr, exec_globals, exec_globals)
|
||
except SyntaxError:
|
||
try:
|
||
tree = ast.parse(expr.strip(), mode="exec")
|
||
tree.body = _reference_capture_tail_expr(tree.body)
|
||
ast.fix_missing_locations(tree)
|
||
exec(compile(tree, "<yoga_custom>", "exec"), exec_globals, exec_globals)
|
||
result = exec_globals.get("__result__")
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
return result
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def engine() -> YogaEngine:
|
||
return YogaEngine(str(RULES_PATH))
|
||
|
||
|
||
def _globals_for(engine: YogaEngine, chart_index: int) -> dict:
|
||
ascendant, planets = SAMPLE_CHARTS[chart_index]
|
||
ctx = YogaContext(planets, ascendant)
|
||
return engine._build_custom_exec_globals(ctx, {}, {})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1. 等价：同进程差分
|
||
# ---------------------------------------------------------------------------
|
||
def test_cached_compile_matches_pre_change_behaviour_for_every_rule_expression(engine) -> None:
|
||
"""全部 192 条去重表达式，缓存版与参照版在三张示例盘上逐条同值。"""
|
||
# 原值: >= 190
|
||
# 新值: >= 184
|
||
# 原因: TASK-upstream-sync4-20261002 T5：bvr_002～004、bvr_016～018 六条从 custom 表达式改为
|
||
# 声明式 has_planet_in_house_from（上游 2dd6062a），表达式条数相应减少；等价证明仍覆盖其余全部表达式
|
||
assert len(RULE_EXPRESSIONS) >= 184, "规则表表达式数量异常，等价证明的覆盖面不足"
|
||
|
||
mismatches = []
|
||
for chart_index in range(len(SAMPLE_CHARTS)):
|
||
for expr in RULE_EXPRESSIONS:
|
||
# 两侧各用一份全新命名空间：exec 会往里写 __result__，不能共用。
|
||
live = yoga_engine._run_custom_expr(expr, _globals_for(engine, chart_index))
|
||
reference = _reference_run(expr, _globals_for(engine, chart_index))
|
||
if live != reference or bool(live) != bool(reference):
|
||
mismatches.append((chart_index, expr[:80], reference, live))
|
||
|
||
assert not mismatches, f"{len(mismatches)} 条表达式与改前行为不同: {mismatches[:3]}"
|
||
|
||
|
||
def test_reference_differential_can_actually_fail(engine) -> None:
|
||
"""反向验证：差分不是恒真——喂一个被改坏的表达式两侧就该分叉。"""
|
||
exec_globals_a = _globals_for(engine, 0)
|
||
exec_globals_b = _globals_for(engine, 0)
|
||
assert yoga_engine._run_custom_expr("house_of('Sun')", exec_globals_a) == _reference_run(
|
||
"house_of('Sun')", exec_globals_b
|
||
)
|
||
assert yoga_engine._run_custom_expr("house_of('Sun')", _globals_for(engine, 0)) != _reference_run(
|
||
"house_of('Moon')", _globals_for(engine, 0)
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 2. 红线：只缓存 code object，绝不缓存求值结果
|
||
# ---------------------------------------------------------------------------
|
||
def test_cache_stores_code_objects_not_results() -> None:
|
||
expr = "house_of('Sun')"
|
||
yoga_engine._compiled_eval_code.cache_clear()
|
||
compiled = yoga_engine._compiled_eval_code(expr)
|
||
assert isinstance(compiled, types.CodeType), "eval 分支缓存里必须是 code object"
|
||
|
||
multi = "x = house_of('Sun')\nx"
|
||
yoga_engine._compiled_exec_code.cache_clear()
|
||
assert yoga_engine._compiled_eval_code(multi) is None, "多语句表达式在 eval 模式下必须仍然是 None"
|
||
assert isinstance(yoga_engine._compiled_exec_code(multi), types.CodeType)
|
||
|
||
|
||
def test_same_expression_on_different_charts_never_reuses_a_result(engine) -> None:
|
||
"""红线：同一条表达式配不同的盘必须给出不同答案（不得跨盘串结果）。"""
|
||
expr = "house_of('Sun')"
|
||
first = yoga_engine._run_custom_expr(expr, _globals_for(engine, 0))
|
||
second = yoga_engine._run_custom_expr(expr, _globals_for(engine, 1))
|
||
third = yoga_engine._run_custom_expr(expr, _globals_for(engine, 0))
|
||
|
||
assert first == 10 and second == 5, "示例盘的 Sun 宫位应当不同，否则这条断言没有鉴别力"
|
||
assert first != second, "求值结果被缓存了：同一表达式在不同盘上返回了同一个答案"
|
||
assert first == third, "同一张盘重复求值应当稳定"
|
||
|
||
multi = "h = house_of('Sun')\nh + 100"
|
||
assert yoga_engine._run_custom_expr(multi, _globals_for(engine, 0)) == 110
|
||
assert yoga_engine._run_custom_expr(multi, _globals_for(engine, 1)) == 105
|
||
|
||
|
||
def test_run_custom_expr_is_not_memoized() -> None:
|
||
"""求值入口本身不得带缓存装饰器。"""
|
||
assert not hasattr(yoga_engine._run_custom_expr, "cache_info")
|
||
assert hasattr(yoga_engine._compiled_eval_code, "cache_info")
|
||
assert hasattr(yoga_engine._compiled_exec_code, "cache_info")
|
||
|
||
|
||
def test_compile_cache_is_bounded() -> None:
|
||
"""编译缓存必须有界（红线 §5.3）。"""
|
||
for cached in (yoga_engine._compiled_eval_code, yoga_engine._compiled_exec_code):
|
||
maxsize = cached.cache_info().maxsize
|
||
assert maxsize is not None, "编译缓存不得无界"
|
||
assert maxsize >= 256, "maxsize 必须容纳规则表去重后的表达式数量"
|
||
assert yoga_engine._CUSTOM_EXPR_CACHE_MAXSIZE >= len(RULE_EXPRESSIONS)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 3. 编译次数
|
||
# ---------------------------------------------------------------------------
|
||
class _CompileCounter:
|
||
"""统计一次检测里发生了多少次「源码 → 字节码」编译。
|
||
|
||
`eval(str)` 的编译发生在 CPython 内部、不经过 builtins.compile，
|
||
所以两者都要数：`eval(str)` 调用数 + `builtins.compile` 调用数。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self.eval_str = 0
|
||
self.eval_code = 0
|
||
self.compile = 0
|
||
self._orig_eval = builtins.eval
|
||
self._orig_compile = builtins.compile
|
||
|
||
def __enter__(self) -> "_CompileCounter":
|
||
def counting_eval(source, *args, **kwargs):
|
||
if isinstance(source, str):
|
||
self.eval_str += 1
|
||
else:
|
||
self.eval_code += 1
|
||
return self._orig_eval(source, *args, **kwargs)
|
||
|
||
def counting_compile(source, *args, **kwargs):
|
||
self.compile += 1
|
||
return self._orig_compile(source, *args, **kwargs)
|
||
|
||
builtins.eval = counting_eval
|
||
builtins.compile = counting_compile
|
||
return self
|
||
|
||
def __exit__(self, *exc) -> None:
|
||
builtins.eval = self._orig_eval
|
||
builtins.compile = self._orig_compile
|
||
|
||
@property
|
||
def source_compiles(self) -> int:
|
||
return self.eval_str + self.compile
|
||
|
||
|
||
def test_second_detection_compiles_nothing(engine) -> None:
|
||
ascendant, planets = SAMPLE_CHARTS[0]
|
||
yoga_engine._compiled_eval_code.cache_clear()
|
||
yoga_engine._compiled_exec_code.cache_clear()
|
||
|
||
with _CompileCounter() as first:
|
||
first_result = engine.detect(planets, ascendant)
|
||
with _CompileCounter() as second:
|
||
second_result = engine.detect(planets, ascendant)
|
||
with _CompileCounter() as third:
|
||
third_result = engine.detect(planets, ascendant)
|
||
|
||
assert first.source_compiles > 0, "冷缓存的第一次检测必须真的编译"
|
||
assert second.source_compiles == 0, f"第二次检测仍编译了 {second.source_compiles} 次"
|
||
assert third.source_compiles == 0, "编译次数不得随请求数增长"
|
||
assert second.eval_code > 0, "第二次检测应当在复用 code object"
|
||
assert second.eval_str == 0, "不得再有字符串 eval"
|
||
assert first_result == second_result == third_result
|
||
|
||
|
||
def test_detection_output_is_unchanged_across_charts(engine) -> None:
|
||
"""缓存不得改变任何一张盘的判定输出。"""
|
||
for ascendant, planets in SAMPLE_CHARTS:
|
||
yoga_engine._compiled_eval_code.cache_clear()
|
||
yoga_engine._compiled_exec_code.cache_clear()
|
||
cold = engine.detect(planets, ascendant)
|
||
warm = engine.detect(planets, ascendant)
|
||
assert cold == warm
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 4. 来源合同：expr 只来自仓库内的静态规则表
|
||
# ---------------------------------------------------------------------------
|
||
def test_rule_expressions_come_only_from_the_static_rules_table() -> None:
|
||
"""红线 §5.3：若有用户输入能进入 `expr`，立刻停手。"""
|
||
assert RULES_PATH.is_file()
|
||
assert RULES_PATH.resolve().is_relative_to(ROOT.resolve())
|
||
for expr in RULE_EXPRESSIONS:
|
||
assert isinstance(expr, str)
|
||
|
||
source = (ROOT / "scripts" / "yoga_engine.py").read_text(encoding="utf-8")
|
||
# 表达式的唯一读取点。
|
||
assert source.count('cond.get("expr"') == 1
|
||
|
||
# 生产入口不得把 rules_path 交给调用方：默认路径是仓库内的规则表。
|
||
for entry in ("scripts/jyotish_engine.py", "scripts/jyotish_api_server.py"):
|
||
text = (ROOT / entry).read_text(encoding="utf-8")
|
||
for line in text.splitlines():
|
||
if "detect_yogas(" in line and "def " not in line:
|
||
assert "rules_path" not in line, f"{entry} 给 yoga 引擎传了自定义规则路径: {line.strip()}"
|