Jyotisha/tests/test_yoga_expr_compile_cache.py

#!/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 条去重表达式，缓存版与参照版在三张示例盘上逐条同值。"""
    assert len(RULE_EXPRESSIONS) >= 190, "规则表表达式数量异常，等价证明的覆盖面不足"

    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()}"