games-development-ai/tier2/gen-worker/tests/test_genconfig_nacos.py

"""test_genconfig_nacos.py —— genconfig 接 Nacos 热源(配置控制面阶段二·步骤4)单测。

全程 mock / fake,不触真 Nacos。锁死设计 §4 步骤4「验证」要求的四条异常/回落路径,外加两档隔离、
优先级、门阈值热读、脏值兜底与工厂 best-effort:

  回落链四条(先锁死才许真接线):
    ① 未 start / 未 attach → genconfig 走原链(env→YAML→内置默认),行为 = 现状;
       热源已 attach 但缺该键 → 回落 env(热源无值不截断)。
    ② Nacos 不可达 → NacosHotConfig 磁盘快照/默认兜底(get_config 抛)→ store 空 → genconfig 回落本地。
    ③ namespace 配错 → 订阅的空间/dataId 收不到值(get_config 返 None)→ store 空 → 回落本地
       (佐证前置任务:写读两侧 namespaceId 必须同一真实值,否则配置静默推不到)。
    ④ YAML 回落 → 热源无该键、无 env → 取本地 YAML 与内置默认。

  另:优先级 Nacos > env(命中即压过 env);cheap 与 tier2 两档各读各 dataId 的热源、不串档;
  软预算(budget)与门阈值(gates)都吃上热读;脏热值保旧回落不穿透;工厂旗默认关 / best-effort /
  按档 dataId / NO_PROXY 在建 client 前 / 回退开关 detach。

跑法:
  PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest \
      tier2/gen-worker/tests/test_genconfig_nacos.py -v
"""

from __future__ import annotations

import json
import os
import sys
from unittest import mock

import pytest

# 包内/直跑兼容:把 gen-worker/ 加进 sys.path(PYTHONPATH 已指 gen-worker 时这步是冗余保险)。
_HERE = os.path.dirname(os.path.abspath(__file__))
_GW = os.path.dirname(_HERE)  # gen-worker/
if _GW not in sys.path:
    sys.path.insert(0, _GW)

from worker import genconfig, genconfig_nacos  # noqa: E402 —— 被测模块
from worker.nacos_hotconfig import NacosHotConfig  # noqa: E402


# 所有测试可能碰到的 env,fixture 统一快照 + 还原,防跨测/跨会话泄漏。
_ENV_KEYS = (
    "TIER2_GEN__BUDGET__RMB_HARD_LIMIT",
    "TIER2_GEN__GATES__QPASS_GO_MIN",
    "TIER2_GENCONFIG",
    "TIER2_GENCONFIG_NACOS",
    "TIER2_GENCONFIG_NACOS_DATAID",
    "TIER2_GENCONFIG_NACOS_GROUP",
    "NACOS_SERVER_ADDR",
    "NACOS_NAMESPACE",
    "NACOS_USERNAME",
    "NACOS_PASSWORD",
    "NO_PROXY",
    "no_proxy",
)


@pytest.fixture()
def clean_genconfig():
    """每测前后:detach 热源 + reload YAML 缓存 + 清相关 env,保证从纯本地基线起跑、测后不泄漏。"""
    genconfig.detach_hot_source()
    genconfig.reload()
    snap = {k: os.environ.get(k) for k in _ENV_KEYS}
    for k in _ENV_KEYS:
        os.environ.pop(k, None)
    try:
        yield
    finally:
        genconfig.detach_hot_source()
        genconfig.reload()
        for k, v in snap.items():
            if v is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = v


class _FakeHotSource:
    """最小热源:满足 genconfig 期望的 get(key, default)->value 契约(NacosHotConfig 读接口的子集)。

    genconfig 用 src.get("area.key", _MISSING) 查:store 无该键时把 genconfig 传入的 sentinel 原样返回,
    genconfig 据此判「热源没这个键」继续回落。故这里 default 必须原样透传、不能替换成别的默认。
    """

    def __init__(self, store: dict, raise_on_get: bool = False):
        self._store = dict(store)
        self._raise = raise_on_get  # True 模拟热源取值异常(get() 抛)

    def get(self, key, default=None):
        if self._raise:
            raise RuntimeError("hot source boom")
        return self._store.get(key, default)


class _FakeNacosClient:
    """fake v1 nacos client:喂给真 NacosHotConfig,验路径②③(不可达 / 读不到)。契约对齐已核验的 v1。"""

    def __init__(self, content, unreachable: bool = False):
        self._content = content
        self.unreachable = unreachable  # True → get_config 抛(模拟连磁盘快照都没有的不可达)

    def get_config(self, data_id, group, timeout=None, no_snapshot=None):
        if self.unreachable:
            raise RuntimeError("nacos unreachable")
        return self._content  # None 模拟「订阅的空间/dataId 收不到值」(namespace 配错表现)

    def add_config_watcher(self, data_id, group, cb, content=None):
        pass  # 订阅 no-op:这些回落测试不涉及推送热更


# ════════════════════════════════════════════════════════════════════════════
# 回落链四条(设计 §4 步骤4 验证要求)
# ════════════════════════════════════════════════════════════════════════════

def test_path1_not_attached_uses_local_chain(clean_genconfig):
    """① 未 attach → genconfig 完全走原链(env→YAML→内置默认),行为 = 现状字节不变。"""
    assert genconfig._hot_source is None  # 默认未接热源
    baseline = genconfig.get("budget", "rmb_hard_limit", 50.0)  # 纯本地基线(真 YAML 或内置默认)
    # 设 env 一次性覆盖仍照常生效(未 attach 时 env 就是最高优先级)。
    os.environ["TIER2_GEN__BUDGET__RMB_HARD_LIMIT"] = "88.0"
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 88.0
    assert baseline != 88.0  # 基线不是 env 值,证明确实走了本地链


def test_path1b_attached_but_key_absent_falls_back_to_env(clean_genconfig):
    """① 热源已 attach 但缺该键 → 回落 env(热源无值不把取值截断成 None)。"""
    genconfig.attach_hot_source(_FakeHotSource({"gates.qpass_go_min": 0.9}))  # 只有别的键
    os.environ["TIER2_GEN__BUDGET__RMB_HARD_LIMIT"] = "77.0"
    # 热源没有 budget.rmb_hard_limit → 回落 env 77.0(不是热源那条 gates 键、更不是 None)。
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 77.0


def test_path2_nacos_unreachable_falls_back_to_local(clean_genconfig):
    """② Nacos 不可达:真 NacosHotConfig + 不可达 client → store 空 → genconfig 回落本地。"""
    genconfig.detach_hot_source()
    baseline = genconfig.get("budget", "rmb_hard_limit", 50.0)  # 先取纯本地基线
    fake = _FakeNacosClient(None, unreachable=True)
    hot = NacosHotConfig(client=fake, data_id="gen-hot-params-tier2", group="DEFAULT_GROUP", defaults={})
    hot.start()  # 不可达:get_config 抛,best-effort 留空 store,不抛
    genconfig.attach_hot_source(hot)
    # 热源 store 空(不可达兜底)→ genconfig 回落本地基线,不被 Nacos 连累。
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline


def test_path3_namespace_misconfig_reads_nothing_falls_back(clean_genconfig):
    """③ namespace 配错:订阅空间/dataId 收不到值(get_config 返 None)→ store 空 → 回落本地。

    佐证前置任务:写侧 publish 进真实 namespaceId、读侧订阅另一个(如误写 public)→ 永远读不到,
    表现就是本条——store 停在空、genconfig 回落本地。故写读两侧 namespaceId 必须一次性对齐。
    """
    genconfig.detach_hot_source()
    baseline = genconfig.get("budget", "rmb_hard_limit", 50.0)
    fake = _FakeNacosClient(None)  # 非不可达,但订阅的空间读不到任何内容
    hot = NacosHotConfig(client=fake, data_id="gen-hot-params-tier2", group="DEFAULT_GROUP", defaults={})
    hot.start()
    genconfig.attach_hot_source(hot)
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline


def test_path4_yaml_fallback_when_hot_and_env_absent(clean_genconfig):
    """④ YAML 回落:热源无该键、无 env → 取本地 YAML 与内置默认(结果 = 纯本地基线)。"""
    genconfig.detach_hot_source()
    baseline = genconfig.get("budget", "rmb_hard_limit", 50.0)
    genconfig.attach_hot_source(_FakeHotSource({}))  # 空热源(无任何键)
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline  # 回落本地,与未 attach 一致


# ════════════════════════════════════════════════════════════════════════════
# 优先级 / 两档隔离 / 门阈值 / 脏值
# ════════════════════════════════════════════════════════════════════════════

def test_hot_source_overrides_env_and_yaml(clean_genconfig):
    """优先级 Nacos > env > YAML:热源命中即压过 env(配置中心才是权威,遗留 env 不得静默盖住激活值)。"""
    os.environ["TIER2_GEN__BUDGET__RMB_HARD_LIMIT"] = "77.0"  # 机器上遗留实验 env
    genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": 99.0}))
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 99.0  # 热源赢


def test_two_tiers_read_own_dataid(clean_genconfig):
    """两档各读各 dataId:tier2 进程热源只有 tier2 键、cheap 进程热源只有 cheap 键,互不串档。"""
    # —— tier2 进程:先取 cheap 键的纯本地基线(fixture 保证起始未 attach),再 attach tier2 dataId 的热源 ——
    cheap_local = genconfig.get("budget", "cheap_rmb_hard_limit", 10.0)  # cheap 键纯本地基线
    genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": 55.0}))
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 55.0            # tier2 键命中热源
    assert genconfig.get("budget", "cheap_rmb_hard_limit", 10.0) == cheap_local  # cheap 键热源没有→回落本地

    # —— cheap 进程:换成 attach cheap dataId 的热源(store 里是 budget.cheap_rmb_hard_limit)——
    genconfig.detach_hot_source()
    tier2_baseline = genconfig.get("budget", "rmb_hard_limit", 50.0)  # tier2 键的纯本地基线
    genconfig.attach_hot_source(_FakeHotSource({"budget.cheap_rmb_hard_limit": 12.0}))
    assert genconfig.get("budget", "cheap_rmb_hard_limit", 10.0) == 12.0     # cheap 键命中热源
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == tier2_baseline  # tier2 键热源没有→回落本地


def test_gate_thresholds_also_hot_read(clean_genconfig):
    """门阈值(gates 区)与软预算(budget 区)一样吃上热读——路 B 覆盖预算 + 阈值两类。"""
    genconfig.attach_hot_source(_FakeHotSource({
        "budget.rmb_hard_limit": 66.0,
        "gates.qpass_go_min": 0.66,
    }))
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 66.0
    assert genconfig.get("gates", "qpass_go_min", 0.40) == 0.66


def test_hot_value_coerced_by_type(clean_genconfig):
    """热源值按内置默认类型转型:JSON 里的字符串 "45" 落到 float 旋钮上得 45.0(与 env/YAML 同款转型)。"""
    genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": "45"}))
    got = genconfig.get("budget", "rmb_hard_limit", 50.0)
    assert got == 45.0 and isinstance(got, float)


def test_dirty_hot_value_falls_back_no_throw(clean_genconfig):
    """脏热值(转不动的字符串)→ 保旧回落本地、不抛、不穿透成 None。"""
    genconfig.detach_hot_source()
    baseline = genconfig.get("budget", "rmb_hard_limit", 50.0)
    genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": "not-a-number"}))
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline  # 脏值被挡,回落本地


def test_hot_source_get_exception_falls_back(clean_genconfig):
    """热源 get 自身抛异常 → genconfig 视为「没取到」回落本地,绝不连累主链。"""
    genconfig.detach_hot_source()
    baseline = genconfig.get("budget", "rmb_hard_limit", 50.0)
    genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": 99.0}, raise_on_get=True))
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == baseline


# ════════════════════════════════════════════════════════════════════════════
# 工厂(build_and_attach_hot_source)
# ════════════════════════════════════════════════════════════════════════════

def test_factory_disabled_by_default(clean_genconfig):
    """工厂旗默认关:不设 TIER2_GENCONFIG_NACOS → build 返回 None、不 attach、genconfig 走原链。"""
    assert genconfig_nacos.is_enabled() is False
    hot = genconfig_nacos.build_and_attach_hot_source(tier="tier2")
    assert hot is None
    assert genconfig._hot_source is None


def test_factory_attaches_and_genconfig_reads_hot(clean_genconfig):
    """启用旗打开 + fake client 有值 → 工厂建 client → NacosHotConfig 初读 → attach → genconfig 命中热值。"""
    os.environ["TIER2_GENCONFIG_NACOS"] = "1"
    fake_client = _FakeNacosClient(json.dumps({"budget.rmb_hard_limit": 42.0}))
    with mock.patch("nacos.NacosClient", return_value=fake_client):
        hot = genconfig_nacos.build_and_attach_hot_source(tier="tier2")
    assert hot is not None
    assert genconfig._hot_source is hot
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 42.0  # 端到端:经工厂 attach 后读到热值


def test_factory_best_effort_on_client_error(clean_genconfig):
    """工厂 best-effort:client 构造抛 → build 返回 None、不抛、genconfig 未 attach、回落纯本地。"""
    os.environ["TIER2_GENCONFIG_NACOS"] = "1"
    with mock.patch("nacos.NacosClient", side_effect=RuntimeError("nacos unreachable")):
        hot = genconfig_nacos.build_and_attach_hot_source(tier="tier2")  # 不应抛
    assert hot is None
    assert genconfig._hot_source is None


def test_factory_dataid_per_tier(clean_genconfig):
    """按档隔离:tier="cheap" 订阅 gen-hot-params-cheap、tier="tier2" 订阅 gen-hot-params-tier2。"""
    os.environ["TIER2_GENCONFIG_NACOS"] = "1"
    seen = {}

    def _spy_hotconfig(*, client, data_id, group, defaults):
        seen["data_id"] = data_id
        seen["defaults"] = defaults
        return mock.MagicMock()

    # cheap 档
    with mock.patch("nacos.NacosClient", return_value=mock.MagicMock()), \
            mock.patch("worker.nacos_hotconfig.NacosHotConfig", side_effect=_spy_hotconfig):
        genconfig_nacos.build_and_attach_hot_source(tier="cheap")
    assert seen["data_id"] == "gen-hot-params-cheap"
    assert seen["defaults"] == {}  # 坑④:热源 defaults 必须空,不引入第二默认值

    genconfig.detach_hot_source()
    # tier2 档
    with mock.patch("nacos.NacosClient", return_value=mock.MagicMock()), \
            mock.patch("worker.nacos_hotconfig.NacosHotConfig", side_effect=_spy_hotconfig):
        genconfig_nacos.build_and_attach_hot_source(tier="tier2")
    assert seen["data_id"] == "gen-hot-params-tier2"


def test_factory_dataid_env_override(clean_genconfig):
    """env TIER2_GENCONFIG_NACOS_DATAID 显式覆盖档默认(演示脚本用它指向演示专用 dataId)。"""
    os.environ["TIER2_GENCONFIG_NACOS"] = "1"
    os.environ["TIER2_GENCONFIG_NACOS_DATAID"] = "gen-hot-params-demo"
    seen = {}

    def _spy_hotconfig(*, client, data_id, group, defaults):
        seen["data_id"] = data_id
        return mock.MagicMock()

    with mock.patch("nacos.NacosClient", return_value=mock.MagicMock()), \
            mock.patch("worker.nacos_hotconfig.NacosHotConfig", side_effect=_spy_hotconfig):
        genconfig_nacos.build_and_attach_hot_source(tier="tier2")  # 档默认本应 tier2,被 env 覆盖
    assert seen["data_id"] == "gen-hot-params-demo"


def test_factory_proxy_bypass_before_client_build(clean_genconfig):
    """内网铁律:NO_PROXY 旁路必须在【建 client 前】完成 —— 构造 NacosClient 时 NO_PROXY 已含 Nacos host。"""
    os.environ["TIER2_GENCONFIG_NACOS"] = "1"
    os.environ["NACOS_SERVER_ADDR"] = "100.64.0.8:8848"
    seen = {}

    def _spy_ctor(*_a, **_kw):
        seen["no_proxy_at_build"] = os.environ.get("NO_PROXY", "")
        return mock.MagicMock()

    with mock.patch("nacos.NacosClient", side_effect=_spy_ctor), \
            mock.patch("worker.nacos_hotconfig.NacosHotConfig", return_value=mock.MagicMock()):
        genconfig_nacos.build_and_attach_hot_source(tier="tier2")
    assert "100.64.0.8" in seen.get("no_proxy_at_build", "")


def test_factory_namespace_defaults_empty_string(clean_genconfig):
    """真实 namespaceId 默认空串(勿写 public;前置任务口径)——构造 client 时 namespace 关键字 == ""。"""
    os.environ["TIER2_GENCONFIG_NACOS"] = "1"
    ctor = mock.MagicMock(return_value=mock.MagicMock())
    with mock.patch("nacos.NacosClient", ctor), \
            mock.patch("worker.nacos_hotconfig.NacosHotConfig", return_value=mock.MagicMock()):
        genconfig_nacos.build_and_attach_hot_source(tier="tier2")
    _args, kwargs = ctor.call_args
    assert kwargs.get("namespace") == ""  # 默认公共空间 = 空串
    assert kwargs.get("username") == "nacos"


def test_rollback_switch_detach(clean_genconfig):
    """回退开关:detach 后 genconfig 回落纯本地链(设计 §3.6/§6 一处开关切回本地 YAML)。"""
    genconfig.attach_hot_source(_FakeHotSource({"budget.rmb_hard_limit": 99.0}))
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) == 99.0
    genconfig_nacos.detach()  # 回退
    assert genconfig._hot_source is None
    assert genconfig.get("budget", "rmb_hard_limit", 50.0) != 99.0  # 回落本地,不再是热值