"""后端注册表测试:覆盖 register / get / 内置 backend 注册。""" from common.storage.base import AsyncStorageBackend from common.storage.exceptions import StorageConfigError from common.storage.registry import ( get_backend_class, register_backend, registered_backends, ) def test_local_and_s3_are_registered_at_import_time(): """import common.storage 应该触发 local / s3 的注册。""" backend_classes = registered_backends() assert "local" in backend_classes assert "s3" in backend_classes for cls in backend_classes.values(): assert issubclass(cls, AsyncStorageBackend) def test_get_backend_class_returns_async_subclass(): cls = get_backend_class("local") assert issubclass(cls, AsyncStorageBackend) def test_get_backend_class_unknown_raises(): import pytest with pytest.raises(StorageConfigError) as exc_info: get_backend_class("does-not-exist") assert "未知的存储后端" in str(exc_info.value) assert "does-not-exist" in str(exc_info.value) def test_register_backend_idempotent_for_same_class(): """同一个类对象重复注册是 no-op,不抛错(``is`` 比对,避免重复 import 时误冲突)。""" from common.storage.registry import _REGISTRY class _SameAgain(AsyncStorageBackend): pass _REGISTRY["same-again-test"] = _SameAgain # 第二次装饰同一个类对象:当前实现里装饰器返回 cls 并把 _REGISTRY[name] 重新写一遍。 # 直接重新调用 register_backend("same-again-test")(_SameAgain) 不抛错即可。 fn = register_backend("same-again-test") result = fn(_SameAgain) assert result is _SameAgain assert _REGISTRY["same-again-test"] is _SameAgain _REGISTRY.pop("same-again-test", None) def test_register_backend_conflict_raises(): import pytest @register_backend("conflict-test") class A(AsyncStorageBackend): pass with pytest.raises(StorageConfigError) as exc_info: @register_backend("conflict-test") class B(AsyncStorageBackend): pass assert "已被注册" in str(exc_info.value) # cleanup:避免污染全局 registry 影响其他测试 from common.storage.registry import _REGISTRY _REGISTRY.pop("conflict-test", None)