tao.chenandtao.chen
5d74285f34
The drop-sync refactor is now contractually enforced:
* create_storage({"mode": "sync"|"async"|<any>}) raises StorageConfigError
-- the regression we most want to catch is someone "restoring" the
sync shape by re-introducing mode= handling.
* create_storage() preserves caller's dict, returns an AsyncStorageBackend
subclass, and wraps constructor TypeError into a useful StorageConfigError.
* build_storage_config emits no mode key on either local or s3 branch.
* register_backend() refuses to silently overwrite a different class on
the same name (collision guard), and accepts same-class re-registration.
* registry surfaces the built-in local + s3 classes after import; the
conflict-test cleanup pattern avoids leaking global state across tests.
Adds [tool.pytest.ini_options] (asyncio_mode=auto, testpaths=tests) so
uv run --package common pytest common/tests works from the workspace root.
17 new tests, all green. Backend suite still 136 passed.
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
"""后端注册表测试：覆盖 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) |