mirror of
https://github.com/MemeMeow-Studio/MemeMeow.git
synced 2026-10-19 06:21:45 +08:00
infstellar
5ecda39843
- Implement tests for the ImageProcessingWorker to validate task deduplication and policy interactions. - Create tests for operation policy to ensure proper grant handling and error responses. - Add tests for reverse image provider response structure to validate empty and malformed results. - Develop comprehensive tests for scope resolution and service validation to ensure correct behavior in various scenarios.
741 lines
30 KiB
Python
741 lines
30 KiB
Python
"""应用 scope 解析边界和显式 FastAPI 装配测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import api as api_module
|
|
from api import bind_request_scope, create_app
|
|
from backend.database import ScopeContext
|
|
from backend.scope import (
|
|
LocalScopeResolver,
|
|
ScopeResolutionError,
|
|
ScopeServiceFactory,
|
|
ScopeServices,
|
|
ScopeServicesError,
|
|
resolve_scope,
|
|
validate_scope_services,
|
|
)
|
|
from backend.tasks import TaskRecord
|
|
|
|
|
|
def _request(resolver: object | None = None):
|
|
"""构造不依赖 ASGI 生命周期的最小 resolver 请求。"""
|
|
state = SimpleNamespace()
|
|
if resolver is not None:
|
|
state.scope_resolver = resolver
|
|
return SimpleNamespace(app=SimpleNamespace(state=state))
|
|
|
|
|
|
class _ClosedScopeResolver:
|
|
"""模拟适配宿主注入的非 local 可信 scope resolver。"""
|
|
|
|
def resolve(self, request):
|
|
"""返回宿主认证后绑定的 scope，不读取客户端请求字段。"""
|
|
del request
|
|
return "scope-a"
|
|
|
|
|
|
def _install_lifespan_doubles(monkeypatch, tmp_path: Path) -> SimpleNamespace:
|
|
"""为生命周期测试注入无数据库副作用的资源、Worker 和 scope facade 替身。"""
|
|
|
|
class FakeEngine:
|
|
"""记录共享 Engine 是否在生命周期结束时释放。"""
|
|
|
|
def __init__(self) -> None:
|
|
self.disposed = False
|
|
|
|
def dispose(self) -> None:
|
|
"""记录 dispose 调用。"""
|
|
self.disposed = True
|
|
|
|
class FakeSettings:
|
|
"""提供默认 factory 装配所需的最小只读配置。"""
|
|
|
|
image_root = tmp_path / "images"
|
|
data_root = tmp_path / "data"
|
|
opencode_runtime_root = tmp_path / "runtime"
|
|
database_url = "postgresql+psycopg://host/test"
|
|
expected_database_revision = "test-revision"
|
|
opencode_concurrency = 1
|
|
agent_backpressure = 32
|
|
settings_version = "test-settings"
|
|
worker_lease_seconds = 120
|
|
worker_max_attempts = 3
|
|
|
|
def ensure_directories(self) -> None:
|
|
"""创建测试运行目录，不触碰数据库 scope。"""
|
|
self.data_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
class FakeResources:
|
|
"""记录 local scope 开关并拒绝任何未预期的 preflight。"""
|
|
|
|
def __init__(self, engine, **kwargs) -> None:
|
|
self.engine = engine
|
|
self.require_local_scope = kwargs["require_local_scope"]
|
|
self.preflight_calls: list[object] = []
|
|
|
|
def flat_preflight(self, scope) -> dict[str, object]:
|
|
"""记录 local 预检调用，供 OSS local 路径断言。"""
|
|
self.preflight_calls.append(scope)
|
|
return {}
|
|
|
|
class FakeRunner:
|
|
"""隔离生命周期测试中的外部 Agent runner。"""
|
|
|
|
def shutdown(self) -> None:
|
|
"""满足应用关闭协议。"""
|
|
|
|
class FakeWorkerManager:
|
|
"""记录进程级 manager 的 handler、启动和关闭动作。"""
|
|
|
|
last_instance = None
|
|
|
|
def __init__(self, resources, **kwargs) -> None:
|
|
del resources
|
|
type(self).last_instance = self
|
|
self.executor = kwargs["executor"]
|
|
self.owner = "test-worker"
|
|
self.registered: dict[str, object] = {}
|
|
self.started = False
|
|
self.stopped = False
|
|
self.task_resolver = None
|
|
self.scope_resolver = None
|
|
|
|
@property
|
|
def worker_count(self) -> int:
|
|
"""返回测试中唯一的进程级 manager 数量。"""
|
|
return 0 if self.stopped else 1
|
|
|
|
def set_service_resolvers(self, task_resolver, scope_resolver) -> None:
|
|
"""记录 factory 安装的持久任务和 scope facade 解析器。"""
|
|
self.task_resolver = task_resolver
|
|
self.scope_resolver = scope_resolver
|
|
|
|
def register(self, task_type, handler) -> None:
|
|
"""记录全局 handler 注册，不创建任何 scope facade。"""
|
|
self.registered[task_type] = handler
|
|
|
|
def start(self) -> dict[str, list[str]]:
|
|
"""记录 manager 启动并返回标准诊断结构。"""
|
|
self.started = True
|
|
return {"started": [self.owner], "invalid_tasks": []}
|
|
|
|
def shutdown(self) -> None:
|
|
"""记录 manager 关闭。"""
|
|
self.stopped = True
|
|
|
|
class DummyMetadata:
|
|
"""scope-bound 元数据 facade 替身。"""
|
|
|
|
def __init__(self, _resources, *, scope_id):
|
|
self.scope = scope_id if isinstance(scope_id, ScopeContext) else ScopeContext(scope_id)
|
|
service_scope_calls.append(self.scope.scope_id)
|
|
self.blob_store = SimpleNamespace()
|
|
self.recovered = False
|
|
|
|
def recover_storage(self, *, limit: int) -> dict[str, int]:
|
|
"""记录 local 启动时的存储恢复调用。"""
|
|
del limit
|
|
self.recovered = True
|
|
return {}
|
|
|
|
class DummySearch:
|
|
"""scope-bound 搜索 facade 替身。"""
|
|
|
|
def __init__(self, _settings, _resources, _metadata, *, scope_id):
|
|
self.scope = scope_id if isinstance(scope_id, ScopeContext) else ScopeContext(scope_id)
|
|
service_scope_calls.append(self.scope.scope_id)
|
|
|
|
class DummyTasks:
|
|
"""只记录共享 manager，不创建 scope 私有线程池。"""
|
|
|
|
def __init__(self, _resources, *, scope_id, worker_manager, **_kwargs):
|
|
self.scope = scope_id if isinstance(scope_id, ScopeContext) else ScopeContext(scope_id)
|
|
service_scope_calls.append(self.scope.scope_id)
|
|
self.worker_manager = worker_manager
|
|
self.registered: dict[str, object] = {}
|
|
self.finalizer = None
|
|
|
|
def register(self, task_type, handler) -> None:
|
|
"""记录 facade 转发到进程级 registry 的 handler。"""
|
|
self.registered[task_type] = handler
|
|
self.worker_manager.register(task_type, handler)
|
|
|
|
def set_batch_finalizer(self, callback) -> None:
|
|
"""记录 scope 专属批次收束回调。"""
|
|
self.finalizer = callback
|
|
|
|
class DummyReverse:
|
|
"""scope-bound 反向图片 facade 替身。"""
|
|
|
|
def __init__(self, _settings, _resources, *, scope_id, provider=None):
|
|
del provider
|
|
self.scope = scope_id if isinstance(scope_id, ScopeContext) else ScopeContext(scope_id)
|
|
service_scope_calls.append(self.scope.scope_id)
|
|
|
|
class DummyVisual:
|
|
"""scope-bound 视觉 facade 替身。"""
|
|
|
|
def __init__(self, _settings, _resources, *, scope_id):
|
|
self.scope = scope_id if isinstance(scope_id, ScopeContext) else ScopeContext(scope_id)
|
|
service_scope_calls.append(self.scope.scope_id)
|
|
|
|
engine = FakeEngine()
|
|
settings = FakeSettings()
|
|
resources: list[FakeResources] = []
|
|
check_calls: list[dict[str, object]] = []
|
|
service_scope_calls: list[str] = []
|
|
|
|
def make_resources(engine_arg, **kwargs):
|
|
"""保存生命周期实际传入的 local-scope 配置。"""
|
|
value = FakeResources(engine_arg, **kwargs)
|
|
resources.append(value)
|
|
return value
|
|
|
|
def check_database(_engine, **kwargs):
|
|
"""记录数据库启动门禁参数，不连接真实 PostgreSQL。"""
|
|
check_calls.append(kwargs)
|
|
return {"revision": "test", "pgvector": True}
|
|
|
|
monkeypatch.setattr(api_module.Settings, "from_env", classmethod(lambda cls: settings))
|
|
monkeypatch.setattr(api_module, "create_engine_for_settings", lambda _settings: engine)
|
|
monkeypatch.setattr(api_module, "check_database", check_database)
|
|
monkeypatch.setattr(api_module, "DatabaseResources", make_resources)
|
|
monkeypatch.setattr(api_module, "OpenCodeRunner", lambda _settings: FakeRunner())
|
|
monkeypatch.setattr(api_module, "OpenCodeActivityReader", lambda _root: object())
|
|
monkeypatch.setattr(api_module, "VisualInferenceClient", lambda _settings: object())
|
|
monkeypatch.setattr(api_module, "PostgresTaskWorkerManager", FakeWorkerManager)
|
|
monkeypatch.setattr(api_module, "PostgresMetadataService", DummyMetadata)
|
|
monkeypatch.setattr(api_module, "PostgresSearchService", DummySearch)
|
|
monkeypatch.setattr(api_module, "PostgresTaskService", DummyTasks)
|
|
monkeypatch.setattr(api_module, "ReverseImageService", DummyReverse)
|
|
monkeypatch.setattr(api_module, "VisualSearchService", DummyVisual)
|
|
return SimpleNamespace(engine=engine, settings=settings, resources=resources, check_calls=check_calls, service_scope_calls=service_scope_calls, worker_class=FakeWorkerManager)
|
|
|
|
|
|
def test_local_scope_resolver_ignores_client_fields_and_is_immutable() -> None:
|
|
"""local 适配器不读取伪造字段，ScopeContext 不能被原地修改。"""
|
|
request = _request(LocalScopeResolver("local"))
|
|
request.query_params = {"scope_id": "scope-b", "user_id": "attacker"}
|
|
assert resolve_scope(request) == ScopeContext("local")
|
|
with pytest.raises((AttributeError, TypeError)):
|
|
resolve_scope(request).scope_id = "scope-b" # type: ignore[misc]
|
|
|
|
|
|
def test_missing_or_invalid_resolver_fails_closed() -> None:
|
|
"""缺失 resolver、异常 resolver 和空结果都不能回退 local。"""
|
|
with pytest.raises(ScopeResolutionError, match="未配置"):
|
|
resolve_scope(_request())
|
|
|
|
class BrokenResolver:
|
|
"""模拟宿主认证上下文异常。"""
|
|
|
|
def resolve(self, request):
|
|
"""返回无效结果以验证稳定失败。"""
|
|
del request
|
|
return None
|
|
|
|
with pytest.raises(ScopeResolutionError):
|
|
resolve_scope(_request(BrokenResolver()))
|
|
|
|
|
|
def test_create_app_requires_explicit_resolver() -> None:
|
|
"""应用工厂漏配 resolver 时直接 fail-closed。"""
|
|
with pytest.raises(ScopeResolutionError):
|
|
create_app(scope_resolver=None)
|
|
application = create_app(scope_resolver=LocalScopeResolver("local"))
|
|
assert application.state.scope_resolver.scope.scope_id == "local"
|
|
|
|
|
|
def test_scope_resolver_supports_explicit_scope_method_and_rejects_delimiter_controls() -> None:
|
|
"""宿主只实现 resolve_scope 时仍可注入，DEL 等控制字符不能进入 scope。"""
|
|
|
|
class ScopeMethodResolver:
|
|
"""模拟适配宿主的显式 resolver 方法。"""
|
|
|
|
resolve = "not-callable"
|
|
|
|
def resolve_scope(self, request):
|
|
"""返回可信 scope，证明备用方法不会被同名属性遮蔽。"""
|
|
del request
|
|
return "scope-a"
|
|
|
|
assert resolve_scope(_request(ScopeMethodResolver())) == ScopeContext("scope-a")
|
|
with pytest.raises(ValueError, match="scope_invalid"):
|
|
ScopeContext("scope-\x7f-a")
|
|
|
|
|
|
def test_create_app_keeps_host_factory_injection() -> None:
|
|
"""应用装配应保留宿主 factory，不能在构造阶段替换成隐式 local 服务。"""
|
|
factory = object()
|
|
provider = object()
|
|
application = create_app(scope_resolver=LocalScopeResolver("local"), service_factory=factory, agent_input_provider=provider)
|
|
assert application.state.service_factory is factory
|
|
assert application.state.agent_input_provider is provider
|
|
|
|
|
|
def test_host_lifespan_never_requires_local_scope_or_preflight(monkeypatch, tmp_path: Path) -> None:
|
|
"""宿主 factory 启动不调用 local 装配，也允许数据库没有 local namespace。"""
|
|
class FakeEngine:
|
|
"""记录生命周期是否正确释放共享 Engine。"""
|
|
|
|
def __init__(self) -> None:
|
|
self.disposed = False
|
|
|
|
def dispose(self) -> None:
|
|
"""记录 dispose 调用。"""
|
|
self.disposed = True
|
|
|
|
class FakeSettings:
|
|
"""提供 host lifespan 所需的最小只读配置。"""
|
|
|
|
image_root = tmp_path / "host-images"
|
|
data_root = tmp_path / "host-data"
|
|
opencode_runtime_root = tmp_path / "host-runtime"
|
|
database_url = "postgresql+psycopg://host/test"
|
|
expected_database_revision = "test-revision"
|
|
|
|
def ensure_directories(self) -> None:
|
|
"""host 仍需要创建运行时目录，但不创建 local scope。"""
|
|
self.data_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
class FakeResources:
|
|
"""拒绝任何未授权的 local preflight 调用。"""
|
|
|
|
def __init__(self, engine, **kwargs) -> None:
|
|
self.require_local_scope = kwargs["require_local_scope"]
|
|
self.preflight_calls: list[object] = []
|
|
|
|
def flat_preflight(self, scope) -> dict[str, object]:
|
|
"""记录调用，测试应断言该方法从未被调用。"""
|
|
self.preflight_calls.append(scope)
|
|
raise AssertionError("host lifespan 不应执行 local preflight")
|
|
|
|
class FakeRunner:
|
|
"""隔离测试中的外部 Agent runner。"""
|
|
|
|
def shutdown(self) -> None:
|
|
"""满足 lifespan 关闭协议。"""
|
|
|
|
class HostFactory:
|
|
"""只提供 scope-aware factory 协议，不提供 local scope。"""
|
|
|
|
def __init__(self) -> None:
|
|
self.for_scope_calls: list[object] = []
|
|
self.started = False
|
|
self.stopped = False
|
|
|
|
def for_scope(self, scope):
|
|
"""任何 local 调用都说明 host 启动仍有隐式 fallback。"""
|
|
self.for_scope_calls.append(scope)
|
|
if getattr(scope, "scope_id", scope) == "local":
|
|
raise AssertionError("host factory 不应装配 local")
|
|
raise AssertionError("测试不应在 lifespan 内创建业务 scope")
|
|
|
|
def for_task(self, task_id):
|
|
"""提供内部 callback 所需的协议占位。"""
|
|
del task_id
|
|
return None
|
|
|
|
def start_all(self):
|
|
"""host 自己负责全局 Worker 启动。"""
|
|
self.started = True
|
|
return {"started": ["host-worker"], "invalid_tasks": []}
|
|
|
|
def shutdown(self) -> None:
|
|
"""记录 host factory 关闭。"""
|
|
self.stopped = True
|
|
|
|
engine = FakeEngine()
|
|
settings = FakeSettings()
|
|
resources: list[FakeResources] = []
|
|
factory = HostFactory()
|
|
check_calls: list[dict[str, object]] = []
|
|
|
|
def make_resources(engine_arg, **kwargs):
|
|
"""保存实际传入的 host local-scope 开关。"""
|
|
value = FakeResources(engine_arg, **kwargs)
|
|
resources.append(value)
|
|
return value
|
|
|
|
monkeypatch.setattr(api_module.Settings, "from_env", classmethod(lambda cls: settings))
|
|
monkeypatch.setattr(api_module, "create_engine_for_settings", lambda _settings: engine)
|
|
monkeypatch.setattr(api_module, "check_database", lambda _engine, **kwargs: (check_calls.append(kwargs) or {"revision": "test", "pgvector": True}))
|
|
monkeypatch.setattr(api_module, "DatabaseResources", make_resources)
|
|
monkeypatch.setattr(api_module, "OpenCodeRunner", lambda _settings: FakeRunner())
|
|
monkeypatch.setattr(api_module, "OpenCodeActivityReader", lambda _root: object())
|
|
monkeypatch.setattr(api_module, "VisualInferenceClient", lambda _settings: object())
|
|
|
|
application = create_app(scope_resolver=_ClosedScopeResolver(), service_factory=factory)
|
|
asyncio.run(_exercise_lifespan(application, factory, resources, engine))
|
|
assert check_calls[0]["require_local_installation"] is False
|
|
|
|
|
|
async def _exercise_lifespan(application, factory, resources, engine) -> None:
|
|
"""运行 host lifespan 并断言 local 资源从未被访问。"""
|
|
async with api_module.lifespan(application):
|
|
assert resources[0].require_local_scope is False
|
|
assert resources[0].preflight_calls == []
|
|
assert factory.for_scope_calls == []
|
|
assert factory.started is True
|
|
assert not hasattr(application.state, "metadata")
|
|
assert not hasattr(application.state, "tasks")
|
|
assert factory.stopped is True
|
|
assert engine.disposed is True
|
|
|
|
|
|
def test_nonlocal_resolver_with_default_factory_never_preloads_local(monkeypatch, tmp_path: Path) -> None:
|
|
"""非 local resolver 使用核心默认 factory 时不创建、预检或缓存 local facade。"""
|
|
doubles = _install_lifespan_doubles(monkeypatch, tmp_path)
|
|
application = create_app(scope_resolver=_ClosedScopeResolver())
|
|
|
|
async def exercise() -> None:
|
|
"""运行默认 factory 生命周期并检查启动前后的 scope 边界。"""
|
|
async with api_module.lifespan(application):
|
|
manager = doubles.worker_class.last_instance
|
|
assert manager is not None
|
|
assert manager.started is True
|
|
assert set(manager.registered) == {"cache_generation", "metadata_repair", "visual_embedding_generation", "meme_context_generation"}
|
|
assert doubles.resources[0].require_local_scope is False
|
|
assert doubles.resources[0].preflight_calls == []
|
|
assert doubles.check_calls[0]["require_local_installation"] is False
|
|
assert not hasattr(application.state, "resolver")
|
|
assert not hasattr(application.state, "metadata")
|
|
assert not hasattr(application.state, "tasks")
|
|
assert doubles.service_scope_calls == []
|
|
assert isinstance(application.state.service_factory, ScopeServiceFactory)
|
|
assert application.state.service_factory._services == {}
|
|
|
|
asyncio.run(exercise())
|
|
manager = doubles.worker_class.last_instance
|
|
assert manager is not None and manager.stopped is True
|
|
assert doubles.engine.disposed is True
|
|
|
|
|
|
def test_explicit_oss_local_resolver_keeps_local_preflight_and_facade(monkeypatch, tmp_path: Path) -> None:
|
|
"""只有显式 OSS LocalScopeResolver("local") 才保留 local 预检和默认 facade。"""
|
|
assert isinstance(api_module.app.state.scope_resolver, LocalScopeResolver)
|
|
assert api_module.app.state.scope_resolver.scope.scope_id == "local"
|
|
doubles = _install_lifespan_doubles(monkeypatch, tmp_path)
|
|
application = create_app(scope_resolver=LocalScopeResolver("local"))
|
|
|
|
async def exercise() -> None:
|
|
"""运行 OSS local 生命周期并检查 local 资源确实被显式装配。"""
|
|
async with api_module.lifespan(application):
|
|
manager = doubles.worker_class.last_instance
|
|
assert manager is not None
|
|
assert manager.started is True
|
|
assert doubles.resources[0].require_local_scope is True
|
|
assert doubles.resources[0].preflight_calls == [ScopeContext("local")]
|
|
assert doubles.check_calls[0]["require_local_installation"] is True
|
|
assert hasattr(application.state, "resolver")
|
|
assert hasattr(application.state, "metadata")
|
|
assert hasattr(application.state, "tasks")
|
|
assert doubles.service_scope_calls == ["local"] * 5
|
|
assert set(application.state.service_factory._services) == {"local"}
|
|
assert application.state.service_factory._services["local"].metadata.recovered is True
|
|
|
|
asyncio.run(exercise())
|
|
manager = doubles.worker_class.last_instance
|
|
assert manager is not None and manager.stopped is True
|
|
assert doubles.engine.disposed is True
|
|
|
|
|
|
def test_many_scope_facades_share_one_process_worker_manager(monkeypatch) -> None:
|
|
"""大量 scope 只创建轻量 facade，不复制 Worker、线程池或 owner。"""
|
|
import backend.pg_services as pg_services
|
|
import backend.reverse_image as reverse_image
|
|
import backend.visual as visual
|
|
|
|
class DummyMetadata:
|
|
"""scope facade 测试替身。"""
|
|
|
|
def __init__(self, _resources, *, scope_id):
|
|
self.scope = scope_id
|
|
self.blob_store = SimpleNamespace()
|
|
|
|
class DummySearch:
|
|
"""搜索 facade 测试替身。"""
|
|
|
|
def __init__(self, _settings, _resources, _metadata, *, scope_id):
|
|
self.scope = scope_id
|
|
|
|
class DummyTasks:
|
|
"""只记录共享 manager，不创建自己的 executor。"""
|
|
|
|
def __init__(self, _resources, *, scope_id, worker_manager, **_kwargs):
|
|
self.scope = ScopeContext(scope_id)
|
|
self.worker_manager = worker_manager
|
|
|
|
def register(self, _task_type, _handler):
|
|
"""兼容 factory 的 handler 注册钩子。"""
|
|
|
|
def set_batch_finalizer(self, _callback):
|
|
"""兼容批次 finalizer 注册钩子。"""
|
|
|
|
class DummyReverse:
|
|
"""反向图片 facade 测试替身。"""
|
|
|
|
def __init__(self, _settings, _resources, *, scope_id, provider=None):
|
|
self.scope = scope_id
|
|
|
|
class DummyVisual:
|
|
"""视觉 facade 测试替身。"""
|
|
|
|
def __init__(self, _settings, _resources, *, scope_id):
|
|
self.scope = scope_id
|
|
|
|
monkeypatch.setattr(pg_services, "PostgresMetadataService", DummyMetadata)
|
|
monkeypatch.setattr(pg_services, "PostgresSearchService", DummySearch)
|
|
monkeypatch.setattr(pg_services, "PostgresTaskService", DummyTasks)
|
|
monkeypatch.setattr(reverse_image, "ReverseImageService", DummyReverse)
|
|
monkeypatch.setattr(visual, "VisualSearchService", DummyVisual)
|
|
|
|
class EmptySession:
|
|
"""让 manager 关闭路径无需真实数据库即可验证 executor 生命周期。"""
|
|
|
|
def __enter__(self):
|
|
"""返回空 session。"""
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
"""结束空 session。"""
|
|
|
|
def scalars(self, _statement):
|
|
"""关闭测试没有 running 任务。"""
|
|
return []
|
|
|
|
def commit(self) -> None:
|
|
"""兼容 manager 的事务提交。"""
|
|
|
|
settings = SimpleNamespace(opencode_concurrency=2, agent_backpressure=32, settings_version="test", worker_lease_seconds=120, worker_max_attempts=3)
|
|
manager_resources = SimpleNamespace(factory=lambda: EmptySession())
|
|
factory = ScopeServiceFactory(manager_resources, settings)
|
|
try:
|
|
services = [factory.for_scope(f"scope-{index}") for index in range(100)]
|
|
assert factory._services == {}
|
|
assert len({id(service.tasks.worker_manager) for service in services}) == 1
|
|
assert factory._worker_manager.worker_count == 1
|
|
assert factory._worker_manager.owner == services[0].tasks.worker_manager.owner
|
|
finally:
|
|
factory.shutdown()
|
|
|
|
|
|
def test_internal_callback_does_not_require_user_request_scope() -> None:
|
|
"""内部 callback 由 task 控制面恢复 scope，不应被公共 resolver 提前拦截。"""
|
|
state = SimpleNamespace(service_factory=SimpleNamespace(for_task=lambda _task_id: None))
|
|
request = SimpleNamespace(url=SimpleNamespace(path="/internal/visual-search/match"), app=SimpleNamespace(state=state))
|
|
called: list[bool] = []
|
|
|
|
async def call_next(_request):
|
|
"""记录 callback 已进入路由层。"""
|
|
called.append(True)
|
|
return "ok"
|
|
|
|
assert asyncio.run(bind_request_scope(request, call_next)) == "ok"
|
|
assert called == [True]
|
|
|
|
|
|
def test_public_scope_middleware_fails_closed_without_resolver() -> None:
|
|
"""公共请求缺少 resolver 时在业务路由前返回稳定 503。"""
|
|
state = SimpleNamespace(service_factory=SimpleNamespace(for_scope=lambda _scope: None))
|
|
request = SimpleNamespace(url=SimpleNamespace(path="/images"), app=SimpleNamespace(state=state))
|
|
|
|
async def call_next(_request):
|
|
"""该回调不应在 scope 解析失败时被调用。"""
|
|
raise AssertionError("业务路由不应执行")
|
|
|
|
response = asyncio.run(bind_request_scope(request, call_next))
|
|
assert response.status_code == 503
|
|
|
|
|
|
def test_task_scope_is_internal_and_never_serialized_as_public_field() -> None:
|
|
"""任务快照可以内部保存 scope，但公开结构不能把授权事实回显给客户端。"""
|
|
record = TaskRecord(task_id="task-1", task_type="cache_generation", scope_id="scope-a")
|
|
assert record.scope_id == "scope-a"
|
|
assert "scope_id" not in record.as_dict()
|
|
|
|
|
|
def _scope_services(scope_id: str, *, child_scope_id: str | None = None) -> ScopeServices:
|
|
"""构造只含 scope 声明的最小服务视图，供一致性测试使用。"""
|
|
child_scope = ScopeContext(child_scope_id or scope_id)
|
|
child = SimpleNamespace(scope=child_scope, blob_store=SimpleNamespace(scope=child_scope))
|
|
return ScopeServices(
|
|
ScopeContext(scope_id),
|
|
child,
|
|
child,
|
|
child,
|
|
child,
|
|
child,
|
|
)
|
|
|
|
|
|
def test_scope_service_validation_rejects_outer_and_child_scope_mismatch() -> None:
|
|
"""factory 返回错误外层或子服务 scope 时必须在业务访问前失败。"""
|
|
with pytest.raises(ScopeServicesError, match="外层"):
|
|
validate_scope_services("scope-a", _scope_services("scope-b"))
|
|
with pytest.raises(ScopeServicesError, match="子服务"):
|
|
validate_scope_services("scope-a", _scope_services("scope-a", child_scope_id="scope-b"))
|
|
|
|
|
|
def test_worker_factory_mismatch_settles_only_the_complete_claim() -> None:
|
|
"""Worker 装配返回错误 scope 时只用完整 claim fencing 收束自身任务。"""
|
|
from backend.pg_services import PostgresTaskWorkerManager
|
|
|
|
manager = PostgresTaskWorkerManager.__new__(PostgresTaskWorkerManager)
|
|
claim = SimpleNamespace(
|
|
id="task-a",
|
|
scope_id="scope-a",
|
|
claim_generation=7,
|
|
lease_owner="worker-a",
|
|
lease_expires_at=object(),
|
|
)
|
|
settled: list[object] = []
|
|
finished: list[tuple[str, bool]] = []
|
|
manager._scope_service_resolver = lambda _scope: _scope_services("scope-b")
|
|
manager._claim_for_task = lambda _task_id: claim
|
|
manager._fail_unresolvable = lambda value: settled.append(value)
|
|
manager._task_finished = lambda task_id, *, claimed: finished.append((task_id, claimed))
|
|
manager._run("task-a")
|
|
assert settled == [claim]
|
|
assert finished == [("task-a", True)]
|
|
|
|
|
|
def test_worker_reclaim_after_delayed_factory_failure_does_not_settle_new_claim() -> None:
|
|
"""装配异常延迟到重新认领后，旧 claim 的收束不能覆盖新 generation。"""
|
|
from backend.pg_services import PostgresTaskWorkerManager
|
|
|
|
manager = PostgresTaskWorkerManager.__new__(PostgresTaskWorkerManager)
|
|
first = SimpleNamespace(id="task-a", scope_id="scope-a", claim_generation=7)
|
|
second = SimpleNamespace(id="task-a", scope_id="scope-a", claim_generation=8)
|
|
claims = iter((first, second))
|
|
settled: list[object] = []
|
|
finished: list[tuple[str, bool]] = []
|
|
executed: list[tuple[str, object]] = []
|
|
|
|
class TaskFacade:
|
|
"""记录第二次有效 claim 的执行，避免触碰数据库。"""
|
|
|
|
scope = ScopeContext("scope-a")
|
|
|
|
def _run(self, task_id, *, preclaimed):
|
|
"""记录有效 claim，证明旧 claim 未被复用。"""
|
|
executed.append((task_id, preclaimed))
|
|
|
|
valid = _scope_services("scope-a")
|
|
valid = ScopeServices(valid.scope, valid.metadata, valid.search, TaskFacade(), valid.reverse_image, valid.visual_search)
|
|
manager._claim_for_task = lambda _task_id: next(claims)
|
|
resolver_calls = 0
|
|
|
|
def resolve_services(_scope):
|
|
"""第一次模拟延迟装配异常，第二次返回重新认领后的有效视图。"""
|
|
nonlocal resolver_calls
|
|
resolver_calls += 1
|
|
if resolver_calls == 1:
|
|
raise RuntimeError("late assembly")
|
|
return valid
|
|
|
|
manager._scope_service_resolver = resolve_services
|
|
manager._fail_unresolvable = lambda value: settled.append(value)
|
|
manager._task_finished = lambda task_id, *, claimed: finished.append((task_id, claimed))
|
|
manager._run("task-a")
|
|
manager._run("task-a")
|
|
assert settled == [first]
|
|
assert executed == [("task-a", second)]
|
|
# 第二次有效 claim 的收束由 scope task facade 自己负责，manager 不能再次
|
|
# 把它当作装配失败 claim 终止。
|
|
assert finished == [("task-a", True)]
|
|
|
|
|
|
def test_worker_successful_claim_settlement_releases_only_matching_lane_slot() -> None:
|
|
"""有效 claim 的装配失败会释放自身槽位，fencing 不匹配时保留他人槽位。"""
|
|
from backend.pg_services import PostgresTaskWorkerManager
|
|
|
|
claim = SimpleNamespace(id="task-a", scope_id="scope-a", claim_generation=7)
|
|
task = SimpleNamespace(
|
|
id="task-a",
|
|
scope_id="scope-a",
|
|
status="running",
|
|
completed_at=None,
|
|
lease_owner="worker-a",
|
|
lease_expires_at=object(),
|
|
message=None,
|
|
error=None,
|
|
)
|
|
slot = SimpleNamespace(
|
|
task_scope_id="scope-a",
|
|
task_id="task-a",
|
|
lease_owner="worker-a",
|
|
claim_generation=7,
|
|
lease_expires_at=object(),
|
|
)
|
|
|
|
class Session:
|
|
"""按查询顺序返回当前 claim 行和其 lane 槽位。"""
|
|
|
|
def __init__(self):
|
|
self.calls = 0
|
|
self.committed = False
|
|
|
|
def scalar(self, _statement):
|
|
"""提供锁定查询的最小替身。"""
|
|
self.calls += 1
|
|
return task if self.calls == 1 else slot
|
|
|
|
def commit(self):
|
|
"""记录事务收束。"""
|
|
self.committed = True
|
|
|
|
class Resources:
|
|
"""提供 manager 所需的短事务上下文。"""
|
|
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
def factory(self):
|
|
"""返回同一个 fake session。"""
|
|
class Context:
|
|
def __enter__(inner_self):
|
|
return self.session
|
|
|
|
def __exit__(inner_self, *_args):
|
|
return None
|
|
|
|
return Context()
|
|
|
|
manager = PostgresTaskWorkerManager.__new__(PostgresTaskWorkerManager)
|
|
manager.owner = "worker-a"
|
|
session = Session()
|
|
manager.resources = Resources(session)
|
|
manager._fail_unresolvable(claim)
|
|
assert task.status == "failed"
|
|
assert task.lease_owner is None
|
|
assert task.lease_expires_at is None
|
|
assert slot.task_id is None
|
|
assert slot.lease_owner is None
|
|
assert session.committed is True
|
|
|
|
foreign_slot = SimpleNamespace(
|
|
task_scope_id="scope-a",
|
|
task_id="task-a",
|
|
lease_owner="worker-b",
|
|
claim_generation=8,
|
|
lease_expires_at=object(),
|
|
)
|
|
|
|
class SlotSession:
|
|
"""只返回另一个 Worker 的槽位，验证 fencing 拒绝不会清空它。"""
|
|
|
|
def scalar(self, _statement):
|
|
"""返回被保护的槽位。"""
|
|
return foreign_slot
|
|
|
|
assert PostgresTaskWorkerManager._release_slot(SlotSession(), "scope-a", "task-a", owner="worker-a", claim_generation=7) is False
|
|
assert foreign_slot.task_id == "task-a"
|
|
assert foreign_slot.lease_owner == "worker-b"
|