feat(runtime): 为旧门面补上分阶段退役,与命中观测形成闭环 (#6430)

This commit is contained in:
Aqr-K
2026-08-24 10:10:56 +08:00
committed by GitHub
parent af97f2c27a
commit f55f5edeca
7 changed files with 731 additions and 3 deletions
+2
View File
@@ -553,6 +553,8 @@ class ConfigModel(BaseModel):
USAGE_STATISTIC_SHARE: bool = True USAGE_STATISTIC_SHARE: bool = True
# 是否开启插件热加载 # 是否开启插件热加载
PLUGIN_AUTO_RELOAD: bool = False PLUGIN_AUTO_RELOAD: bool = False
# 临时放行的废弃标识,多个用,分隔;仅对已进入停用阶段的接口有效,用于观察真实依赖方
DEPRECATION_ENABLED: Optional[str] = None
# 本地插件仓库目录,多个地址使用,分隔 # 本地插件仓库目录,多个地址使用,分隔
PLUGIN_LOCAL_REPO_PATHS: Optional[str] = None PLUGIN_LOCAL_REPO_PATHS: Optional[str] = None
+7
View File
@@ -0,0 +1,7 @@
"""渐进式废弃与清理边界;文案登记见 notices,运行期行为见 policy,具体入口必须从子模块显式导入。
一条能力退场分四步:先只登记不打扰(``SILENT``),靠 ``compat.facade.hit`` 一类指标
观察真实用量;确认可以收口后转为标记预警(``WARN``);预警足够久之后默认停用并留开关
逼出剩余依赖方(``DISABLED``);最后物理删除(``REMOVED``)。推进阶段只需改
``notices.NOTICES`` 中该条登记的 ``stage``,调用点无需改动。
"""
+144
View File
@@ -0,0 +1,144 @@
"""废弃登记与文案。
全仓所有「即将废弃」的对外文案集中在本模块的 ``NOTICES`` 里,调用点只引用稳定的
``key``,不各自拼写提示语。旧 Facade 的登记标识与 ``compat.facade.hit`` 指标的
``facade``/``operation`` 标签保持一致:整个 Facade 用 ``Facade``,单个方法用
``Facade.method``。
"""
from dataclasses import dataclass
from enum import IntEnum
from typing import Dict, Optional
class DeprecationStage(IntEnum):
"""废弃生命周期阶段,数值越大距离物理删除越近。"""
# 仅登记:功能照常且不打扰用户,只靠指标观察真实用量
SILENT = 0
# 标记预警:功能照常,首次触达时输出一次告警
WARN = 1
# 默认停用:触达即报错,需把标识写进 DEPRECATION_ENABLED 才临时恢复
DISABLED = 2
# 彻底移除:实现已从代码中删除,触达即报错且无法恢复
REMOVED = 3
@dataclass(frozen=True)
class DeprecationNotice:
"""
单条废弃登记
:param key: 稳定标识,用作开关配置项与告警去重的键
:param subject: 被废弃的符号或能力
:param stage: 当前所处阶段
:param since: 开始废弃的版本
:param replacement: 替代方案
:param reason: 废弃原因
:param remove_in: 计划物理删除的版本,未定时为 None
"""
key: str
subject: str
stage: DeprecationStage
since: str
replacement: str
reason: str
remove_in: Optional[str] = None
def message(self, context: Optional[str] = None) -> str:
"""
构造面向调用方的提示语
:param context: 触发来源,例如具体方法名或插件标识
:return: 单行提示语
"""
parts = [f"{self.subject}{self.since} 起进入废弃流程"]
parts.append(f"计划在 {self.remove_in} 移除" if self.remove_in else "移除版本待定")
if context:
parts.append(f"触发来源:{context}")
parts.append(f"原因:{self.reason}")
parts.append(f"请改用:{self.replacement}")
if self.stage is DeprecationStage.DISABLED:
parts.append(f"当前已默认停用,如需临时恢复请将 {self.key} 加入 DEPRECATION_ENABLED")
elif self.stage is DeprecationStage.REMOVED:
parts.append("当前已彻底移除,无法恢复")
return "".join(parts)
NOTICES: Dict[str, DeprecationNotice] = {
notice.key: notice
for notice in (
DeprecationNotice(
key="PluginManager._modify_plugin_files",
subject="PluginManager._modify_plugin_files()",
stage=DeprecationStage.SILENT,
since="v3.0.0",
replacement="get_plugin_system().package._modify_plugin_files()",
reason="分身文件改写已由插件包适配器实现,此处只为旧内部调用保留转发",
),
DeprecationNotice(
key="PluginManager._modify_python_file",
subject="PluginManager._modify_python_file()",
stage=DeprecationStage.SILENT,
since="v3.0.0",
replacement="get_plugin_system().package._modify_python_file()",
reason="Python 文件改写已由插件包适配器实现,此处只为旧内部调用保留转发",
),
DeprecationNotice(
key="PluginManager._modify_federation_files",
subject="PluginManager._modify_federation_files()",
stage=DeprecationStage.SILENT,
since="v3.0.0",
replacement="get_plugin_system().package._modify_federation_files()",
reason="联邦文件改写已由插件包适配器实现,此处只为旧内部调用保留转发",
),
DeprecationNotice(
key="PluginManager._rename_federation_assets",
subject="PluginManager._rename_federation_assets()",
stage=DeprecationStage.SILENT,
since="v3.0.0",
replacement="get_plugin_system().package._rename_federation_assets()",
reason="联邦资源重命名已由插件包适配器实现,此处只为旧内部调用保留转发",
),
DeprecationNotice(
key="PluginHelper.find_missing_dependencies",
subject="PluginHelper.find_missing_dependencies()",
stage=DeprecationStage.SILENT,
since="v3.0.0",
replacement="PluginDependencyInstaller.find_missing()",
reason="依赖处理已拆分到独立依赖适配器,此处只为旧市场入口保留转发",
),
DeprecationNotice(
key="PluginHelper.install_dependencies",
subject="PluginHelper.install_dependencies()",
stage=DeprecationStage.SILENT,
since="v3.0.0",
replacement="PluginDependencyInstaller.install()",
reason="依赖处理已拆分到独立依赖适配器,此处只为旧市场入口保留转发",
),
DeprecationNotice(
key="PluginHelper.async_find_missing_dependencies",
subject="PluginHelper.async_find_missing_dependencies()",
stage=DeprecationStage.SILENT,
since="v3.0.0",
replacement="PluginDependencyInstaller.async_find_missing()",
reason="依赖处理已拆分到独立依赖适配器,此处只为旧异步市场入口保留转发",
),
DeprecationNotice(
key="PluginHelper.async_install_dependencies",
subject="PluginHelper.async_install_dependencies()",
stage=DeprecationStage.SILENT,
since="v3.0.0",
replacement="PluginDependencyInstaller.async_install()",
reason="依赖处理已拆分到独立依赖适配器,此处只为旧异步市场入口保留转发",
),
DeprecationNotice(
key="SystemUtils.is_bluray_dir",
subject="SystemUtils.is_bluray_dir()",
stage=DeprecationStage.WARN,
since="v3.0.0",
replacement="StorageChain().is_bluray_folder()",
reason="只按本地路径判断蓝光目录,无法覆盖非本地存储",
),
)
}
+207
View File
@@ -0,0 +1,207 @@
"""废弃阶段的运行期行为。
调用点只需回答两个问题:这条废弃路径现在还该不该生效(``is_active``),以及走到这里
时要不要留痕或拦截(``enforce``)。阶段的推进只改 ``notices.NOTICES`` 里的 ``stage``
调用点不动。
"""
import functools
import inspect
import threading
from typing import Any, Callable, FrozenSet, Optional, Set, Tuple
from app.runtime.deprecation import notices
from app.runtime.deprecation.notices import DeprecationNotice, DeprecationStage
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting
# 已告警过的 (标识, 触发来源),保证每个来源只留一次痕迹
_warned: Set[Tuple[str, Optional[str]]] = set()
_warned_lock = threading.Lock()
class DeprecatedFeatureError(RuntimeError):
"""触达了已停用或已移除的废弃能力。"""
def find_notice(key: str) -> Optional[DeprecationNotice]:
"""
查找废弃登记
:param key: 废弃标识
:return: 对应登记,未登记时为 None
"""
return notices.NOTICES.get(key)
def get_notice(key: str) -> DeprecationNotice:
"""
取出废弃登记
:param key: 废弃标识
:return: 对应登记
:raises KeyError: 标识未登记
"""
notice = find_notice(key)
if notice is None:
raise KeyError(f"未登记的废弃标识:{key}")
return notice
def all_notices() -> Tuple[DeprecationNotice, ...]:
"""
列出全部废弃登记
:return: 按标识升序排列的登记
"""
return tuple(notices.NOTICES[key] for key in sorted(notices.NOTICES))
def _enabled_keys() -> FrozenSet[str]:
"""
读取被显式恢复的废弃标识集合
:return: 标识集合
"""
configured = get_runtime_setting("DEPRECATION_ENABLED") or ""
return frozenset(item.strip() for item in str(configured).split(",") if item.strip())
def _notice_active(notice: DeprecationNotice) -> bool:
"""
判断一条登记当前是否仍应生效
:param notice: 废弃登记
:return: 生效为 True
"""
if notice.stage <= DeprecationStage.WARN:
return True
if notice.stage is DeprecationStage.DISABLED:
return notice.key in _enabled_keys()
return False
def is_active(key: str) -> bool:
"""
判断废弃路径当前是否仍应生效
登记与预警阶段照常生效;停用阶段默认不生效,仅当标识出现在 DEPRECATION_ENABLED 中
才恢复;移除阶段无论如何都不生效。
:param key: 废弃标识
:return: 生效为 True
:raises KeyError: 标识未登记
"""
return _notice_active(get_notice(key))
def warn(key: str, *, context: Optional[str] = None) -> None:
"""
就废弃路径留下一次告警
仅登记阶段不打扰用户;其余阶段下同一 (标识, 触发来源) 在单个进程内只告警一次,
避免热路径刷屏。
:param key: 废弃标识
:param context: 触发来源,例如具体方法名或插件标识
:raises KeyError: 标识未登记
"""
notice = get_notice(key)
if notice.stage is DeprecationStage.SILENT:
return
dedup_key = (key, context)
with _warned_lock:
if dedup_key in _warned:
return
_warned.add(dedup_key)
logger.warning(notice.message(context))
def guard(key: str, *, context: Optional[str] = None) -> None:
"""
拦截已停用或已移除的废弃能力
:param key: 废弃标识
:param context: 触发来源
:raises KeyError: 标识未登记
:raises DeprecatedFeatureError: 该能力当前不应再生效
"""
notice = get_notice(key)
if not _notice_active(notice):
raise DeprecatedFeatureError(notice.message(context))
def enforce(key: str, *, context: Optional[str] = None) -> None:
"""
对一次废弃路径的触达执行当前阶段的处置
未登记的标识视为尚未纳入废弃流程,直接放行;已登记的先拦截再留痕。
:param key: 废弃标识
:param context: 触发来源
:raises DeprecatedFeatureError: 该能力当前不应再生效
"""
if find_notice(key) is None:
return
guard(key, context=context)
warn(key, context=context)
def enforce_facade(facade: str, operation: str) -> None:
"""
对一次旧 Facade 命中执行当前阶段的处置
先按 ``facade.operation`` 精确匹配,再退回整个 Facade 的登记,两者都没有登记则放行,
因此未纳入废弃流程的 Facade 不受任何影响。
:param facade: Facade 标识,与 compat.facade.hit 指标的 facade 标签一致
:param operation: 被调用的方法名
:raises DeprecatedFeatureError: 该 Facade 或方法当前不应再生效
"""
registry = notices.NOTICES
if not registry:
return
context = f"{facade}.{operation}"
for key in (context, facade):
if key in registry:
enforce(key, context=context)
return
def deprecated(key: str) -> Callable:
"""
把废弃语义施加到一个可调用对象上
停用与移除阶段直接抛错,其余阶段留痕后照常执行。
:param key: 废弃标识
:return: 装饰器
"""
def decorator(func: Callable) -> Callable:
if inspect.iscoroutinefunction(func):
@functools.wraps(func)
async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
context = getattr(func, "__qualname__", None)
guard(key, context=context)
warn(key, context=context)
return await func(*args, **kwargs)
return async_wrapper
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
context = getattr(func, "__qualname__", None)
guard(key, context=context)
warn(key, context=context)
return func(*args, **kwargs)
return wrapper
return decorator
def reset_warned() -> None:
"""清空告警去重记录。"""
with _warned_lock:
_warned.clear()
+16
View File
@@ -117,6 +117,20 @@ def observe_compat_facade(facade: str) -> Callable[[_FacadeClass], _FacadeClass]
return decorate return decorate
def _enforce_deprecation(facade: str, operation: str) -> None:
"""对一次旧 Facade 命中执行其当前的废弃阶段处置。
观测回答「谁还在用」,废弃阶段回答「什么时候不再让用」,两者共用同一组标签。
未登记废弃通告的 Facade 在此完全不受影响。
:param facade: Facade 标识,与 compat.facade.hit 指标的 facade 标签一致
:param operation: 被调用的方法名
"""
from app.runtime.deprecation.policy import enforce_facade
enforce_facade(facade, operation)
def _wrap_compat_method( def _wrap_compat_method(
method: Callable[..., Any], method: Callable[..., Any],
facade: str, facade: str,
@@ -135,6 +149,7 @@ def _wrap_compat_method(
visibility=visibility, visibility=visibility,
abi_source="legacy_facade", abi_source="legacy_facade",
) )
_enforce_deprecation(facade, operation)
return await method(*args, **kwargs) return await method(*args, **kwargs)
return cast(Callable[..., Any], async_wrapper) return cast(Callable[..., Any], async_wrapper)
@@ -148,6 +163,7 @@ def _wrap_compat_method(
visibility=visibility, visibility=visibility,
abi_source="legacy_facade", abi_source="legacy_facade",
) )
_enforce_deprecation(facade, operation)
return method(*args, **kwargs) return method(*args, **kwargs)
return cast(Callable[..., Any], wrapper) return cast(Callable[..., Any], wrapper)
+14 -3
View File
@@ -13,8 +13,8 @@
"runtime_to_db": [], "runtime_to_db": [],
"workflow_to_db": [] "workflow_to_db": []
}, },
"edge_count": 6546, "edge_count": 6554,
"edge_sha256": "9799edea47a44d52ae4d5d4bf3e38614b1d1a255f850891b3cc1d8fb85b40735", "edge_sha256": "1c72744be67d95f98eac1c7b1a72bdbf7b6d1d4f337556b8e22293c3a9525c67",
"edges": [ "edges": [
"app -> app.runtime", "app -> app.runtime",
"app -> app.runtime.compat", "app -> app.runtime.compat",
@@ -5588,6 +5588,11 @@
"app.runtime.config -> app.schemas.types", "app.runtime.config -> app.schemas.types",
"app.runtime.debounce -> app.runtime", "app.runtime.debounce -> app.runtime",
"app.runtime.debounce -> app.runtime.log", "app.runtime.debounce -> app.runtime.log",
"app.runtime.deprecation.policy -> app.runtime",
"app.runtime.deprecation.policy -> app.runtime.deprecation",
"app.runtime.deprecation.policy -> app.runtime.deprecation.notices",
"app.runtime.deprecation.policy -> app.runtime.log",
"app.runtime.deprecation.policy -> app.runtime.settings",
"app.runtime.event.binding -> app.runtime", "app.runtime.event.binding -> app.runtime",
"app.runtime.event.binding -> app.runtime.event", "app.runtime.event.binding -> app.runtime.event",
"app.runtime.event.binding -> app.runtime.event.registry", "app.runtime.event.binding -> app.runtime.event.registry",
@@ -5772,6 +5777,9 @@
"app.runtime.extensions.service_config -> app.schemas", "app.runtime.extensions.service_config -> app.schemas",
"app.runtime.extensions.service_config -> app.schemas.system", "app.runtime.extensions.service_config -> app.schemas.system",
"app.runtime.extensions.service_config -> app.schemas.types", "app.runtime.extensions.service_config -> app.schemas.types",
"app.runtime.observability -> app.runtime",
"app.runtime.observability -> app.runtime.deprecation",
"app.runtime.observability -> app.runtime.deprecation.policy",
"app.runtime.progress -> app.runtime", "app.runtime.progress -> app.runtime",
"app.runtime.progress -> app.runtime.cache", "app.runtime.progress -> app.runtime.cache",
"app.runtime.progress -> app.runtime.localization", "app.runtime.progress -> app.runtime.localization",
@@ -6563,7 +6571,7 @@
"app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow",
"app.workflow.actions.transfer_file -> app.workflow.actions" "app.workflow.actions.transfer_file -> app.workflow.actions"
], ],
"module_count": 806, "module_count": 809,
"modules": [ "modules": [
"app", "app",
"app.adapters", "app.adapters",
@@ -7218,6 +7226,9 @@
"app.runtime.config", "app.runtime.config",
"app.runtime.correlation", "app.runtime.correlation",
"app.runtime.debounce", "app.runtime.debounce",
"app.runtime.deprecation",
"app.runtime.deprecation.notices",
"app.runtime.deprecation.policy",
"app.runtime.event", "app.runtime.event",
"app.runtime.event.binding", "app.runtime.event.binding",
"app.runtime.event.contracts", "app.runtime.event.contracts",
+341
View File
@@ -0,0 +1,341 @@
"""渐进式废弃登记的阶段行为,以及与旧 Facade 命中观测的联动测试。"""
import asyncio
import inspect
from dataclasses import dataclass, field
from typing import Dict, List, Mapping, Tuple
import pytest
from app.runtime.config import settings
from app.runtime.deprecation import notices as notices_module
from app.runtime.deprecation import policy
from app.runtime.deprecation.notices import NOTICES, DeprecationNotice, DeprecationStage
from app.runtime.deprecation.policy import (
DeprecatedFeatureError,
all_notices,
deprecated,
enforce_facade,
guard,
is_active,
warn,
)
from app.runtime.observability import (
MetricSpec,
configure_observation,
observe_compat_facade,
)
@dataclass
class _RecordingLogger:
"""只记录 warning 文本的日志替身。"""
messages: List[str] = field(default_factory=list)
def warning(self, message: str) -> None:
"""记录一条告警文本。"""
self.messages.append(message)
@dataclass
class _RecordingObservationPort:
"""保存已经通过标签合同校验的指标写入。"""
records: List[Tuple[str, Dict[str, str]]] = field(default_factory=list)
def record(self, spec: MetricSpec, value: float, labels: Mapping[str, str]) -> None:
"""追加一条不可变测试快照。"""
self.records.append((spec.name, dict(labels)))
@pytest.fixture(autouse=True)
def _reset_warned():
"""避免告警去重记录在用例间泄漏。"""
policy.reset_warned()
yield
policy.reset_warned()
@pytest.fixture
def warnings_log(monkeypatch) -> _RecordingLogger:
"""把废弃告警接到可断言的日志替身上。"""
recorder = _RecordingLogger()
monkeypatch.setattr(policy, "logger", recorder)
return recorder
@pytest.fixture
def observation_port():
"""安装可断言的观测端口,用例结束恢复 no-op。"""
port = _RecordingObservationPort()
configure_observation(port)
yield port
configure_observation(None)
@pytest.fixture
def registry(monkeypatch):
"""
用可控登记表替换全局登记表,避免用例依赖真实登记内容。
:return: ``register(stage, key) -> DeprecationNotice``
"""
table: Dict[str, DeprecationNotice] = {}
monkeypatch.setattr(notices_module, "NOTICES", table)
def register(stage: DeprecationStage, key: str = "demo.legacy") -> DeprecationNotice:
"""
登记一条指定阶段的废弃通告
:param stage: 所处阶段
:param key: 废弃标识
:return: 登记
"""
notice = DeprecationNotice(
key=key,
subject="示例旧入口",
stage=stage,
since="v3.0.0",
replacement="示例新入口",
reason="仅用于测试",
)
table[key] = notice
return notice
return register
def test_silent_stage_stays_active_and_quiet(registry, warnings_log) -> None:
"""仅登记阶段不改变行为,也不产生任何告警。"""
registry(DeprecationStage.SILENT)
assert is_active("demo.legacy") is True
guard("demo.legacy")
warn("demo.legacy")
assert warnings_log.messages == []
def test_warn_stage_logs_once_per_context(registry, warnings_log) -> None:
"""预警阶段功能照常,同一来源只留一次痕迹,不同来源各留一次。"""
registry(DeprecationStage.WARN)
assert is_active("demo.legacy") is True
warn("demo.legacy", context="PluginA")
warn("demo.legacy", context="PluginA")
warn("demo.legacy", context="PluginB")
assert len(warnings_log.messages) == 2
assert "示例旧入口 自 v3.0.0 起进入废弃流程" in warnings_log.messages[0]
assert "移除版本待定" in warnings_log.messages[0]
assert "触发来源:PluginA" in warnings_log.messages[0]
assert "请改用:示例新入口" in warnings_log.messages[0]
assert "触发来源:PluginB" in warnings_log.messages[1]
def test_disabled_stage_blocks_until_explicitly_enabled(registry, monkeypatch) -> None:
"""停用阶段默认抛错,仅在标识写进 DEPRECATION_ENABLED 后恢复。"""
registry(DeprecationStage.DISABLED)
monkeypatch.setattr(settings, "DEPRECATION_ENABLED", None)
assert is_active("demo.legacy") is False
with pytest.raises(DeprecatedFeatureError, match="已默认停用"):
guard("demo.legacy")
monkeypatch.setattr(settings, "DEPRECATION_ENABLED", "other.key, demo.legacy")
assert is_active("demo.legacy") is True
guard("demo.legacy")
def test_removed_stage_cannot_be_restored(registry, monkeypatch) -> None:
"""移除阶段无视开关,一律抛错并说明无法恢复。"""
registry(DeprecationStage.REMOVED)
monkeypatch.setattr(settings, "DEPRECATION_ENABLED", "demo.legacy")
assert is_active("demo.legacy") is False
with pytest.raises(DeprecatedFeatureError, match="已彻底移除"):
guard("demo.legacy")
def test_unregistered_key_is_rejected_by_lookups(registry) -> None:
"""未登记标识在查询入口上直接报错,避免登记表与调用点写法漂移。"""
registry(DeprecationStage.WARN)
with pytest.raises(KeyError, match="未登记的废弃标识"):
is_active("demo.missing")
with pytest.raises(KeyError, match="未登记的废弃标识"):
warn("demo.missing")
with pytest.raises(KeyError, match="未登记的废弃标识"):
guard("demo.missing")
def test_unregistered_key_passes_through_enforcement(registry, warnings_log) -> None:
"""未纳入废弃流程的触达点原样放行,既不报错也不留痕。"""
registry(DeprecationStage.DISABLED)
policy.enforce("demo.missing")
enforce_facade("OtherFacade", "any_method")
assert warnings_log.messages == []
def test_deprecated_decorator_keeps_call_and_metadata(registry, warnings_log) -> None:
"""预警阶段的装饰器留痕后照常执行,并保留原函数元数据。"""
registry(DeprecationStage.WARN)
@deprecated("demo.legacy")
def legacy_call(value: int) -> int:
"""返回入参本身。"""
return value
assert legacy_call(3) == 3
assert legacy_call(4) == 4
assert legacy_call.__name__ == "legacy_call"
assert len(warnings_log.messages) == 1
def test_deprecated_decorator_blocks_disabled_stage(registry, monkeypatch) -> None:
"""停用阶段的装饰器直接拦截,不再执行原实现。"""
registry(DeprecationStage.DISABLED)
monkeypatch.setattr(settings, "DEPRECATION_ENABLED", None)
calls: List[int] = []
@deprecated("demo.legacy")
def legacy_call() -> None:
"""记录一次实际执行。"""
calls.append(1)
with pytest.raises(DeprecatedFeatureError):
legacy_call()
assert calls == []
def test_deprecated_decorator_keeps_async_call_shape(registry, warnings_log) -> None:
"""异步函数经装饰后仍可被调度器识别为协程函数。"""
registry(DeprecationStage.WARN)
@deprecated("demo.legacy")
async def legacy_call(value: int) -> int:
"""返回入参本身。"""
return value
assert inspect.iscoroutinefunction(legacy_call) is True
assert asyncio.run(legacy_call(3)) == 3
assert len(warnings_log.messages) == 1
def test_compat_facade_hit_is_recorded_before_stage_is_applied(
registry, monkeypatch, observation_port
) -> None:
"""停用阶段仍先记账再拦截,命中数不会因为收口而丢失。"""
registry(DeprecationStage.DISABLED, key="DemoFacade.legacy_call")
monkeypatch.setattr(settings, "DEPRECATION_ENABLED", None)
@observe_compat_facade("DemoFacade")
class DemoFacade:
"""被观测的旧 Facade 替身。"""
@staticmethod
def legacy_call() -> str:
"""返回固定结果。"""
return "called"
with pytest.raises(DeprecatedFeatureError):
DemoFacade.legacy_call()
assert observation_port.records == [
(
"compat.facade.hit",
{
"facade": "DemoFacade",
"operation": "legacy_call",
"visibility": "public",
"abi_source": "legacy_facade",
},
)
]
def test_facade_notice_covers_every_operation(registry, warnings_log) -> None:
"""整个 Facade 的登记覆盖其所有方法,且每个方法各留一次痕迹。"""
registry(DeprecationStage.WARN, key="DemoFacade")
@observe_compat_facade("DemoFacade")
class DemoFacade:
"""被观测的旧 Facade 替身。"""
@staticmethod
def first() -> None:
"""空实现。"""
@staticmethod
def second() -> None:
"""空实现。"""
DemoFacade.first()
DemoFacade.first()
DemoFacade.second()
assert len(warnings_log.messages) == 2
assert "触发来源:DemoFacade.first" in warnings_log.messages[0]
assert "触发来源:DemoFacade.second" in warnings_log.messages[1]
def test_method_notice_takes_precedence_over_facade_notice(registry, warnings_log) -> None:
"""同时登记 Facade 与其单个方法时,按方法级登记处置。"""
registry(DeprecationStage.WARN, key="DemoFacade")
registry(DeprecationStage.REMOVED, key="DemoFacade.legacy_call")
@observe_compat_facade("DemoFacade")
class DemoFacade:
"""被观测的旧 Facade 替身。"""
@staticmethod
def legacy_call() -> None:
"""空实现。"""
@staticmethod
def other_call() -> None:
"""空实现。"""
with pytest.raises(DeprecatedFeatureError, match="已彻底移除"):
DemoFacade.legacy_call()
DemoFacade.other_call()
assert "触发来源:DemoFacade.other_call" in warnings_log.messages[0]
def test_async_facade_method_applies_stage(registry, monkeypatch, observation_port) -> None:
"""异步旧 Facade 方法同样先记账再按阶段拦截。"""
registry(DeprecationStage.DISABLED, key="DemoFacade.async_call")
monkeypatch.setattr(settings, "DEPRECATION_ENABLED", None)
@observe_compat_facade("DemoFacade")
class DemoFacade:
"""被观测的旧 Facade 替身。"""
@staticmethod
async def async_call() -> str:
"""返回固定结果。"""
return "called"
with pytest.raises(DeprecatedFeatureError):
asyncio.run(DemoFacade.async_call())
assert observation_port.records[0][1]["operation"] == "async_call"
def test_registered_notices_are_self_consistent() -> None:
"""真实登记表的键与登记本身一致,且每条都给出替代方案与原因。"""
assert NOTICES
for key, notice in NOTICES.items():
assert key == notice.key
assert notice.since
assert notice.replacement
assert notice.reason
assert isinstance(notice.stage, DeprecationStage)
assert all_notices() == tuple(NOTICES[key] for key in sorted(NOTICES))