From f28e8b15388ad671b542982053aa24f128af5a42 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:58:56 +0800 Subject: [PATCH] feat(plugin): add source identity foundation (#6454) --- app/application/plugin/identity.py | 434 +++++++++++++++++ app/db/adapters/pluginidentity.py | 145 ++++++ app/db/models/__init__.py | 4 + app/db/models/pluginidentity.py | 45 ++ app/db/oper/pluginidentity.py | 58 +++ database/versions/d2e4f6a8b0c1_3_0_9.py | 76 +++ docs/rules/05-architecture.md | 2 +- docs/rules/10-data-and-persistent.md | 2 + .../architecture/dependency-baseline.json | 25 +- tests/test_plugin_identity.py | 435 ++++++++++++++++++ tests/test_plugin_identity_migration.py | 205 +++++++++ 11 files changed, 1427 insertions(+), 4 deletions(-) create mode 100644 app/application/plugin/identity.py create mode 100644 app/db/adapters/pluginidentity.py create mode 100644 app/db/models/pluginidentity.py create mode 100644 app/db/oper/pluginidentity.py create mode 100644 database/versions/d2e4f6a8b0c1_3_0_9.py create mode 100644 tests/test_plugin_identity.py create mode 100644 tests/test_plugin_identity_migration.py diff --git a/app/application/plugin/identity.py b/app/application/plugin/identity.py new file mode 100644 index 000000000..c1131a3eb --- /dev/null +++ b/app/application/plugin/identity.py @@ -0,0 +1,434 @@ +"""已安装物理插件的来源身份合同与存量迁移决策。""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, replace +from datetime import datetime +from enum import StrEnum +from typing import Protocol + + +_PLUGIN_ID_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9]{0,127}$") +_ONLINE_SOURCE_KEY_PATTERN = re.compile( + r"^github:[a-z0-9](?:[a-z0-9-]{0,38})/" + r"[a-z0-9._-]{1,100}$" +) +OFFICIAL_PLUGIN_SOURCE_KEY = "github:jxxghp/moviepilot-plugins" + + +class TrustedPluginSourceType(StrEnum): + """可被在线自动更新信任的来源类型。""" + + UNKNOWN = "unknown" + OFFICIAL = "official" + THIRD_PARTY = "third_party" + + +class PluginPayloadSourceType(StrEnum): + """最近一次已提交插件载荷的来源类型。""" + + UNKNOWN = "unknown" + OFFICIAL = "official" + THIRD_PARTY = "third_party" + LOCAL = "local" + + +class PluginBindingBasis(StrEnum): + """物理插件来源身份的建立依据。""" + + LEGACY_UNBOUND = "legacy_unbound" + LOCAL_ONLY = "local_only" + OFFICIAL_DEFAULT = "official_default" + TOFU = "tofu" + EXPLICIT_INSTALL = "explicit_install" + EXPLICIT_SOURCE_CHANGE = "explicit_source_change" + + +class PluginMarketAvailability(StrEnum): + """存量迁移观察市场候选时的可用状态。""" + + AVAILABLE = "available" + UNAVAILABLE = "unavailable" + + +class PluginIdentityConflictError(RuntimeError): + """来源身份的首次创建或 revision 条件更新已失去竞争。""" + + +def normalize_physical_plugin_id(plugin_id: str) -> str: + """校验物理插件 ID，并返回大小写无关的数据库身份键。""" + if plugin_id != plugin_id.strip() or not _PLUGIN_ID_PATTERN.fullmatch(plugin_id): + raise ValueError("插件 ID 必须以字母开头且只能包含 ASCII 字母或数字") + return plugin_id.lower() + + +def validate_online_source_key(source_key: str) -> str: + """校验由市场边界生成的稳定 GitHub 仓库身份键。""" + value = source_key.strip().lower() + if not _ONLINE_SOURCE_KEY_PATTERN.fullmatch(value): + raise ValueError("在线插件来源必须使用 github:/ 规范键") + return value + + +def _validate_online_source_classification( + source_key: str, + source_type: TrustedPluginSourceType | PluginPayloadSourceType, +) -> None: + """保证官方仓库键与官方来源类型始终双向一致。""" + is_official_key = source_key == OFFICIAL_PLUGIN_SOURCE_KEY + is_official_type = source_type.value == TrustedPluginSourceType.OFFICIAL.value + if is_official_key != is_official_type: + raise ValueError("官方来源类型只能对应 MoviePilot 官方插件仓库") + + +def _validate_optional_text( + value: str | None, + *, + field_name: str, + max_length: int, +) -> None: + """让应用层字符串合同与跨数据库列长度保持一致。""" + if value is not None and len(value) > max_length: + raise ValueError(f"{field_name} 长度不能超过 {max_length}") + + +def _validate_receipt(receipt: str | None) -> str | None: + """校验内容收据为带算法前缀的十六进制摘要。""" + if receipt is None: + return None + value = receipt.strip().lower() + if not re.fullmatch(r"sha256:[0-9a-f]{64}", value): + raise ValueError("插件载荷收据必须为 sha256:<64 hex>") + return value + + +@dataclass(frozen=True, slots=True) +class PluginIdentity: + """一份物理插件的可信更新来源与当前载荷审计事实。""" + + plugin_id: str + normalized_plugin_id: str + trusted_source_type: TrustedPluginSourceType + trusted_source_key: str | None + binding_basis: PluginBindingBasis + payload_source_type: PluginPayloadSourceType + payload_source_key: str | None + declared_version: str | None + package_generation: str | None + system_version: str | None + supports_v3: bool | None + supports_v3t: bool | None + payload_receipt: str | None + revision: int + created_at: datetime + updated_at: datetime + bound_at: datetime | None + payload_applied_at: datetime | None + + def __post_init__(self) -> None: + """拒绝不能作为后续来源门禁事实的矛盾状态。""" + normalized_id = normalize_physical_plugin_id(self.plugin_id) + if self.normalized_plugin_id != normalized_id: + raise ValueError("normalized_plugin_id 必须是 plugin_id 的规范化物理身份") + if self.revision < 1: + raise ValueError("来源身份 revision 必须从 1 开始") + if self.created_at.tzinfo is None or self.updated_at.tzinfo is None: + raise ValueError("来源身份审计时间必须包含时区") + if self.updated_at < self.created_at: + raise ValueError("来源身份更新时间不能早于创建时间") + for audit_time in (self.bound_at, self.payload_applied_at): + if audit_time is not None and audit_time.tzinfo is None: + raise ValueError("来源身份审计时间必须包含时区") + if audit_time is not None and not self.created_at <= audit_time <= self.updated_at: + raise ValueError("来源身份审计时间必须位于创建与更新时间之间") + + trusted_key = ( + validate_online_source_key(self.trusted_source_key) + if self.trusted_source_key + else None + ) + if self.trusted_source_type is TrustedPluginSourceType.UNKNOWN: + if trusted_key is not None or self.bound_at is not None: + raise ValueError("未绑定身份不能携带可信来源或绑定时间") + if self.binding_basis not in { + PluginBindingBasis.LEGACY_UNBOUND, + PluginBindingBasis.LOCAL_ONLY, + }: + raise ValueError("未绑定身份只能使用存量迁移或本地插件依据") + else: + if trusted_key is None or self.bound_at is None: + raise ValueError("已绑定身份必须携带规范来源和绑定时间") + if self.binding_basis in { + PluginBindingBasis.LEGACY_UNBOUND, + PluginBindingBasis.LOCAL_ONLY, + }: + raise ValueError("已绑定身份不能使用未绑定来源依据") + _validate_online_source_classification( + trusted_key, + self.trusted_source_type, + ) + if ( + self.binding_basis is PluginBindingBasis.OFFICIAL_DEFAULT + and self.trusted_source_type is not TrustedPluginSourceType.OFFICIAL + ): + raise ValueError("official_default 只能绑定官方来源") + if ( + self.binding_basis is PluginBindingBasis.TOFU + and self.trusted_source_type is not TrustedPluginSourceType.THIRD_PARTY + ): + raise ValueError("TOFU 只能绑定唯一第三方在线来源") + + payload_key = ( + validate_online_source_key(self.payload_source_key) + if self.payload_source_key + else None + ) + if self.payload_source_type in { + PluginPayloadSourceType.OFFICIAL, + PluginPayloadSourceType.THIRD_PARTY, + }: + if payload_key is None: + raise ValueError("在线载荷必须携带规范来源") + _validate_online_source_classification( + payload_key, + self.payload_source_type, + ) + elif payload_key is not None: + raise ValueError("未知或本地载荷不能携带在线来源键") + if self.payload_source_type is PluginPayloadSourceType.UNKNOWN: + if any(( + self.declared_version, + self.package_generation, + self.system_version, + self.supports_v3 is not None, + self.supports_v3t is not None, + self.payload_receipt, + self.payload_applied_at, + )): + raise ValueError("未知载荷不能携带版本、兼容声明、收据或应用时间") + else: + if not self.declared_version or not self.package_generation: + raise ValueError("已知载荷必须携带声明版本和包代际") + if self.payload_applied_at is None or self.payload_receipt is None: + raise ValueError("已知载荷必须携带应用时间和内容收据") + if ( + self.payload_source_type is not PluginPayloadSourceType.LOCAL + and ( + payload_key != trusted_key + or self.payload_source_type.value + != self.trusted_source_type.value + ) + ): + raise ValueError("在线载荷来源必须与可信更新来源一致") + if ( + self.binding_basis is PluginBindingBasis.LOCAL_ONLY + and self.payload_source_type is not PluginPayloadSourceType.LOCAL + ): + raise ValueError("本地插件身份必须携带已提交的本地载荷事实") + + if self.package_generation not in {None, "v1", "v2", "v3"}: + raise ValueError("插件包代际必须为 v1、v2 或 v3") + _validate_optional_text( + self.declared_version, + field_name="插件声明版本", + max_length=64, + ) + _validate_optional_text( + self.system_version, + field_name="插件系统版本要求", + max_length=128, + ) + object.__setattr__(self, "trusted_source_key", trusted_key) + object.__setattr__(self, "payload_source_key", payload_key) + object.__setattr__(self, "payload_receipt", _validate_receipt(self.payload_receipt)) + + +@dataclass(frozen=True, slots=True) +class PluginSourceCandidate: + """存量迁移可观察到的一个在线候选来源。""" + + source_type: TrustedPluginSourceType + source_key: str + + def __post_init__(self) -> None: + """候选只接受可绑定的规范在线来源。""" + if self.source_type is TrustedPluginSourceType.UNKNOWN: + raise ValueError("未知来源不能作为在线候选") + object.__setattr__(self, "source_key", validate_online_source_key(self.source_key)) + _validate_online_source_classification(self.source_key, self.source_type) + + +def plan_legacy_plugin_identity( + *, + plugin_id: str, + market_availability: PluginMarketAvailability, + online_candidates: tuple[PluginSourceCandidate, ...], + is_virtual_instance: bool, + now: datetime, +) -> PluginIdentity | None: + """为存量物理插件建立更新绑定，不冒充当前载荷来源。""" + if is_virtual_instance: + return None + normalized_id = normalize_physical_plugin_id(plugin_id) + candidates = { + (candidate.source_type, candidate.source_key): candidate + for candidate in online_candidates + } + official = sorted( + ( + candidate + for candidate in candidates.values() + if candidate.source_type is TrustedPluginSourceType.OFFICIAL + ), + key=lambda candidate: candidate.source_key, + ) + third_party = sorted( + ( + candidate + for candidate in candidates.values() + if candidate.source_type is TrustedPluginSourceType.THIRD_PARTY + ), + key=lambda candidate: candidate.source_key, + ) + + trusted_type = TrustedPluginSourceType.UNKNOWN + trusted_key = None + basis = PluginBindingBasis.LEGACY_UNBOUND + bound_at = None + if market_availability is PluginMarketAvailability.AVAILABLE and official: + trusted_type = TrustedPluginSourceType.OFFICIAL + trusted_key = official[0].source_key + basis = PluginBindingBasis.OFFICIAL_DEFAULT + bound_at = now + elif ( + market_availability is PluginMarketAvailability.AVAILABLE + and len(third_party) == 1 + ): + trusted_type = TrustedPluginSourceType.THIRD_PARTY + trusted_key = third_party[0].source_key + basis = PluginBindingBasis.TOFU + bound_at = now + + return PluginIdentity( + plugin_id=plugin_id, + normalized_plugin_id=normalized_id, + trusted_source_type=trusted_type, + trusted_source_key=trusted_key, + binding_basis=basis, + payload_source_type=PluginPayloadSourceType.UNKNOWN, + payload_source_key=None, + declared_version=None, + package_generation=None, + system_version=None, + supports_v3=None, + supports_v3t=None, + payload_receipt=None, + revision=1, + created_at=now, + updated_at=now, + bound_at=bound_at, + payload_applied_at=None, + ) + + +class PluginIdentityRepository(Protocol): + """来源身份条件写命令使用的无提交仓储端口。""" + + def get(self, plugin_id: str) -> PluginIdentity | None: + """按规范化物理插件 ID 读取身份。""" + + def stage_create(self, identity: PluginIdentity) -> None: + """暂存首次身份；唯一键竞争时抛出冲突错误。""" + + def stage_replace( + self, + identity: PluginIdentity, + *, + expected_revision: int, + ) -> bool: + """按 revision 条件暂存替换，并返回是否赢得竞争。""" + + +class PluginIdentityUnitOfWork(Protocol): + """来源身份条件写使用的事务端口。""" + + def commit(self) -> None: + """提交当前条件写。""" + + def rollback(self) -> None: + """回滚当前条件写。""" + + +class WritePluginIdentityCommand: + """以数据库 revision 原子创建或替换一份插件来源身份。""" + + def __init__( + self, + repository: PluginIdentityRepository, + unit_of_work: PluginIdentityUnitOfWork, + ) -> None: + """保存仓储与事务所有者。""" + self._repository = repository + self._unit_of_work = unit_of_work + + def execute( + self, + identity: PluginIdentity, + *, + expected_revision: int | None, + ) -> PluginIdentity: + """首次写要求记录不存在，后续写要求 revision 精确匹配。""" + candidate = replace( + identity, + normalized_plugin_id=normalize_physical_plugin_id(identity.plugin_id), + revision=1 if expected_revision is None else expected_revision + 1, + ) + try: + if expected_revision is None: + if ( + candidate.binding_basis + is PluginBindingBasis.EXPLICIT_SOURCE_CHANGE + ): + raise PluginIdentityConflictError( + "首次插件身份不能伪装成已确认的来源变更" + ) + self._repository.stage_create(candidate) + else: + current = self._repository.get(candidate.normalized_plugin_id) + if current is None or current.revision != expected_revision: + raise PluginIdentityConflictError( + f"插件 {candidate.plugin_id} 的来源身份已被其他任务更新" + ) + immutable_binding = ( + "plugin_id", + "normalized_plugin_id", + "trusted_source_type", + "trusted_source_key", + "binding_basis", + "created_at", + "bound_at", + ) + if any( + getattr(candidate, field_name) != getattr(current, field_name) + for field_name in immutable_binding + ): + raise PluginIdentityConflictError( + "普通插件身份更新不能改变物理 ID 或可信来源绑定" + ) + if candidate.updated_at < current.updated_at: + raise PluginIdentityConflictError( + "插件身份更新时间不能早于已提交记录" + ) + if not self._repository.stage_replace( + candidate, + expected_revision=expected_revision, + ): + raise PluginIdentityConflictError( + f"插件 {candidate.plugin_id} 的来源身份已被其他任务更新" + ) + self._unit_of_work.commit() + return candidate + except Exception: + self._unit_of_work.rollback() + raise diff --git a/app/db/adapters/pluginidentity.py b/app/db/adapters/pluginidentity.py new file mode 100644 index 000000000..9e76bab6c --- /dev/null +++ b/app/db/adapters/pluginidentity.py @@ -0,0 +1,145 @@ +"""插件来源身份 Application Port 的 SQLAlchemy 实现。""" + +from collections.abc import Callable +from datetime import datetime + +from sqlalchemy.orm import Session +from sqlalchemy.exc import IntegrityError + +from app.application.plugin.identity import ( + PluginBindingBasis, + PluginPayloadSourceType, + PluginIdentity, + PluginIdentityConflictError, + TrustedPluginSourceType, + WritePluginIdentityCommand, + normalize_physical_plugin_id, +) +from app.db.models.pluginidentity import PluginIdentity as IdentityModel +from app.db.oper.pluginidentity import PluginIdentityOper +from app.db.uow import SqlAlchemyUnitOfWork + + +def _parse_datetime(value: str | None) -> datetime | None: + """把数据库 ISO 时间还原为带时区应用值。""" + return datetime.fromisoformat(value) if value else None + + +def _to_record(model: IdentityModel) -> PluginIdentity: + """把持久化模型映射为已校验的应用身份。""" + return PluginIdentity( + plugin_id=model.plugin_id, + normalized_plugin_id=model.normalized_plugin_id, + trusted_source_type=TrustedPluginSourceType(model.trusted_source_type), + trusted_source_key=model.trusted_source_key, + binding_basis=PluginBindingBasis(model.binding_basis), + payload_source_type=PluginPayloadSourceType(model.payload_source_type), + payload_source_key=model.payload_source_key, + declared_version=model.declared_version, + package_generation=model.package_generation, + system_version=model.system_version, + supports_v3=model.supports_v3, + supports_v3t=model.supports_v3t, + payload_receipt=model.payload_receipt, + revision=model.revision, + created_at=datetime.fromisoformat(model.created_at), + updated_at=datetime.fromisoformat(model.updated_at), + bound_at=_parse_datetime(model.bound_at), + payload_applied_at=_parse_datetime(model.payload_applied_at), + ) + + +def _to_model(identity: PluginIdentity) -> IdentityModel: + """把应用身份映射为不拥有事务的持久化模型。""" + return IdentityModel( + plugin_id=identity.plugin_id, + normalized_plugin_id=identity.normalized_plugin_id, + trusted_source_type=identity.trusted_source_type.value, + trusted_source_key=identity.trusted_source_key, + binding_basis=identity.binding_basis.value, + payload_source_type=identity.payload_source_type.value, + payload_source_key=identity.payload_source_key, + declared_version=identity.declared_version, + package_generation=identity.package_generation, + system_version=identity.system_version, + supports_v3=identity.supports_v3, + supports_v3t=identity.supports_v3t, + payload_receipt=identity.payload_receipt, + revision=identity.revision, + created_at=identity.created_at.isoformat(), + updated_at=identity.updated_at.isoformat(), + bound_at=identity.bound_at.isoformat() if identity.bound_at else None, + payload_applied_at=( + identity.payload_applied_at.isoformat() + if identity.payload_applied_at + else None + ), + ) + + +class _SqlAlchemyIdentityRepository: + """绑定一个调用方 Session 的来源身份仓储。""" + + def __init__(self, session: Session) -> None: + """保存由事务适配器拥有的 Session。""" + self._oper = PluginIdentityOper(session) + + def get(self, plugin_id: str) -> PluginIdentity | None: + """读取并映射指定来源身份。""" + model = self._oper.get_by_plugin_id(plugin_id) + return _to_record(model) if model else None + + def stage_create(self, identity: PluginIdentity) -> None: + """暂存首次身份。""" + try: + self._oper.stage_create(_to_model(identity)) + except IntegrityError as error: + raise PluginIdentityConflictError( + f"插件 {identity.plugin_id} 的来源身份已存在" + ) from error + + def stage_replace( + self, + identity: PluginIdentity, + *, + expected_revision: int, + ) -> bool: + """按 revision 条件暂存替换。""" + return self._oper.stage_replace( + _to_model(identity), + expected_revision=expected_revision, + ) + + +class TransactionalPluginIdentityStore: + """为每次来源身份读写创建独占同步数据库会话。""" + + def __init__(self, session_factory: Callable[[], Session]) -> None: + """保存由组合根提供的同步 Session 工厂。""" + self._session_factory = session_factory + + def get(self, plugin_id: str) -> PluginIdentity | None: + """在短会话内读取指定物理插件身份。""" + session = self._session_factory() + try: + return _SqlAlchemyIdentityRepository(session).get( + normalize_physical_plugin_id(plugin_id) + ) + finally: + session.close() + + def compare_and_set( + self, + identity: PluginIdentity, + *, + expected_revision: int | None, + ) -> PluginIdentity: + """在一个事务内执行首次创建或 revision 条件替换。""" + session = self._session_factory() + try: + return WritePluginIdentityCommand( + repository=_SqlAlchemyIdentityRepository(session), + unit_of_work=SqlAlchemyUnitOfWork(session), + ).execute(identity, expected_revision=expected_revision) + finally: + session.close() diff --git a/app/db/models/__init__.py b/app/db/models/__init__.py index 73ff4d5fd..c575b510f 100644 --- a/app/db/models/__init__.py +++ b/app/db/models/__init__.py @@ -18,6 +18,10 @@ _MODEL_EXPORTS = { "OutboxMessage": ("app.db.models.outbox", "OutboxMessage"), "PassKey": ("app.db.models.passkey", "PassKey"), "PluginData": ("app.db.models.plugindata", "PluginData"), + "PluginIdentity": ( + "app.db.models.pluginidentity", + "PluginIdentity", + ), "Site": ("app.db.models.site", "Site"), "SiteIcon": ("app.db.models.siteicon", "SiteIcon"), "SiteStatistic": ("app.db.models.sitestatistic", "SiteStatistic"), diff --git a/app/db/models/pluginidentity.py b/app/db/models/pluginidentity.py new file mode 100644 index 000000000..c175f6f4b --- /dev/null +++ b/app/db/models/pluginidentity.py @@ -0,0 +1,45 @@ +"""已安装物理插件来源身份模型。""" + +from typing import Optional + +from sqlalchemy import Boolean, CheckConstraint, Integer, String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base, get_id_column + + +class PluginIdentity(Base): + """持久化一份大小写无关、可条件更新的物理插件来源身份。""" + + id = get_id_column() + plugin_id: Mapped[str] = mapped_column(String(128), nullable=False) + normalized_plugin_id: Mapped[str] = mapped_column(String(128), nullable=False) + trusted_source_type: Mapped[str] = mapped_column(String(20), nullable=False) + trusted_source_key: Mapped[Optional[str]] = mapped_column(String(255)) + binding_basis: Mapped[str] = mapped_column(String(32), nullable=False) + payload_source_type: Mapped[str] = mapped_column(String(20), nullable=False) + payload_source_key: Mapped[Optional[str]] = mapped_column(String(255)) + declared_version: Mapped[Optional[str]] = mapped_column(String(64)) + package_generation: Mapped[Optional[str]] = mapped_column(String(8)) + system_version: Mapped[Optional[str]] = mapped_column(String(128)) + supports_v3: Mapped[Optional[bool]] = mapped_column(Boolean) + supports_v3t: Mapped[Optional[bool]] = mapped_column(Boolean) + payload_receipt: Mapped[Optional[str]] = mapped_column(String(71)) + revision: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + created_at: Mapped[str] = mapped_column(String(40), nullable=False) + updated_at: Mapped[str] = mapped_column(String(40), nullable=False) + bound_at: Mapped[Optional[str]] = mapped_column(String(40)) + payload_applied_at: Mapped[Optional[str]] = mapped_column(String(40)) + + __table_args__ = ( + UniqueConstraint( + "normalized_plugin_id", + name="uq_pluginidentity_normalized_plugin_id", + ), + CheckConstraint( + "normalized_plugin_id <> '' " + "AND normalized_plugin_id = lower(normalized_plugin_id)", + name="ck_pluginidentity_normalized_plugin_id", + ), + CheckConstraint("revision >= 1", name="ck_pluginidentity_revision"), + ) diff --git a/app/db/oper/pluginidentity.py b/app/db/oper/pluginidentity.py new file mode 100644 index 000000000..4e2cfac1b --- /dev/null +++ b/app/db/oper/pluginidentity.py @@ -0,0 +1,58 @@ +"""插件来源身份的数据访问原语。""" + +from sqlalchemy import select, update +from sqlalchemy.orm import Session + +from app.db.base import DbOper, execute_dml +from app.db.models.pluginidentity import PluginIdentity + + +class PluginIdentityOper(DbOper): + """在调用方 Session 中查询并条件暂存插件来源身份。""" + + def get_by_plugin_id(self, plugin_id: str) -> PluginIdentity | None: + """按规范化物理插件 ID 查询唯一身份。""" + return self._execute_sync_query( + lambda session: session.execute( + select(PluginIdentity).where( + PluginIdentity.normalized_plugin_id == plugin_id + ) + ).scalar_one_or_none() + ) + + def stage_create(self, identity: PluginIdentity) -> None: + """暂存首次身份并立即暴露数据库唯一键竞争。""" + def stage(session: Session) -> None: + """加入并 flush 当前调用方事务。""" + session.add(identity) + session.flush() + + self._execute_sync_write(stage) + + def stage_replace( + self, + identity: PluginIdentity, + *, + expected_revision: int, + ) -> bool: + """仅在当前 revision 匹配时替换整份审计事实。""" + values = { + column.name: getattr(identity, column.name) + for column in PluginIdentity.__table__.columns + if column.name != "id" + } + return bool( + self._execute_sync_write( + lambda session: execute_dml( + session, + update(PluginIdentity) + .where( + PluginIdentity.normalized_plugin_id + == identity.normalized_plugin_id, + PluginIdentity.revision == expected_revision, + ) + .values(**values), + execution_options={"synchronize_session": False}, + ) + ) + ) diff --git a/database/versions/d2e4f6a8b0c1_3_0_9.py b/database/versions/d2e4f6a8b0c1_3_0_9.py new file mode 100644 index 000000000..993dc9f63 --- /dev/null +++ b/database/versions/d2e4f6a8b0c1_3_0_9.py @@ -0,0 +1,76 @@ +"""3.0.9 add installed plugin source identities. + +Revision ID: d2e4f6a8b0c1 +Revises: c7d9a1e4f2b6 +Create Date: 2026-08-25 +""" + +from alembic import op +import sqlalchemy as sa + + +revision = "d2e4f6a8b0c1" +down_revision = "c7d9a1e4f2b6" +branch_labels = None +depends_on = None + + +def _id_column(dialect_name: str) -> sa.Column: + """保持 PostgreSQL Identity 与 SQLite 整数主键的当前模型语义一致。""" + if dialect_name == "postgresql": + return sa.Column( + "id", + sa.Integer(), + sa.Identity(start=1, cycle=True), + nullable=False, + ) + return sa.Column("id", sa.Integer(), nullable=False) + + +def upgrade() -> None: + """创建一插件一行、支持 revision 条件写的来源身份表。""" + if "pluginidentity" in sa.inspect(op.get_bind()).get_table_names(): + return + op.create_table( + "pluginidentity", + _id_column(op.get_bind().dialect.name), + sa.Column("plugin_id", sa.String(length=128), nullable=False), + sa.Column("normalized_plugin_id", sa.String(length=128), nullable=False), + sa.Column("trusted_source_type", sa.String(length=20), nullable=False), + sa.Column("trusted_source_key", sa.String(length=255), nullable=True), + sa.Column("binding_basis", sa.String(length=32), nullable=False), + sa.Column("payload_source_type", sa.String(length=20), nullable=False), + sa.Column("payload_source_key", sa.String(length=255), nullable=True), + sa.Column("declared_version", sa.String(length=64), nullable=True), + sa.Column("package_generation", sa.String(length=8), nullable=True), + sa.Column("system_version", sa.String(length=128), nullable=True), + sa.Column("supports_v3", sa.Boolean(), nullable=True), + sa.Column("supports_v3t", sa.Boolean(), nullable=True), + sa.Column("payload_receipt", sa.String(length=71), nullable=True), + sa.Column("revision", sa.Integer(), nullable=False), + sa.Column("created_at", sa.String(length=40), nullable=False), + sa.Column("updated_at", sa.String(length=40), nullable=False), + sa.Column("bound_at", sa.String(length=40), nullable=True), + sa.Column("payload_applied_at", sa.String(length=40), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "normalized_plugin_id", + name="uq_pluginidentity_normalized_plugin_id", + ), + sa.CheckConstraint( + "normalized_plugin_id <> '' " + "AND normalized_plugin_id = lower(normalized_plugin_id)", + name="ck_pluginidentity_normalized_plugin_id", + ), + sa.CheckConstraint( + "revision >= 1", + name="ck_pluginidentity_revision", + ), + ) + + +def downgrade() -> None: + """删除尚未启用生产写入的插件来源身份表。""" + if "pluginidentity" not in sa.inspect(op.get_bind()).get_table_names(): + return + op.drop_table("pluginidentity") diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index f181bfd0c..07bc117f9 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -68,7 +68,7 @@ to make the directory tree look symmetrical. | `app/application/chain/` | Injectable Chain runtime context and compatibility provider | | `app/application/agentdata.py` | Named Agent data ports; canonical Agent consumers use `get_agent_*_port()` and do not alias legacy proxies to Oper classes | | `app/application/outbox.py` | Durable intent and Outbox repository/dispatcher contracts for post-commit side effects | -| `app/application/plugin/` | Plugin market catalog, installation command, runtime port, folder operations and dynamic-route use cases; filenames remain single words (`catalog.py`, `install.py`, `runtime.py`, `folders.py`, `routes.py`) | +| `app/application/plugin/` | Plugin market catalog, installation command, installed-plugin identity contract, runtime port, folder operations and dynamic-route use cases; filenames remain single words (`catalog.py`, `identity.py`, `install.py`, `runtime.py`, `folders.py`, `routes.py`) | | `app/application/server/` | MoviePilot Server reporting and sharing use cases; local data readers and transport callbacks are injected by startup | | `app/application/site/` | Configured site catalog, authentication level and index-resource capability; the generated extension and its data bundle stay together here | | `app/application/messaging/` | Message rendering/routing, interactions and the Agent-to-message bridge: `ingress.py` owns the single channel-to-host loopback boundary; `interaction.py` shared interaction contracts and view helpers; `router.py` unified interaction priority and callback dispatch; `site.py`/`subscribe.py`/`skill.py` per-command sessions, input parsing and views; `media.py` media interaction state while the business workflow stays in `MediaInteractionChain`; `plugin.py` plugin input capture and plugin button callbacks; `agent.py` agent choice state, callback protocol and WebAgent bridge; `message.py` notification rendering, templates and queue. Not a public SDK recommended for direct plugin use | diff --git a/docs/rules/10-data-and-persistent.md b/docs/rules/10-data-and-persistent.md index 2b64c8db4..d10c3d54f 100644 --- a/docs/rules/10-data-and-persistent.md +++ b/docs/rules/10-data-and-persistent.md @@ -19,6 +19,7 @@ Models are SQLAlchemy declarative classes. Each model maps to one database table | `Site` / `SiteIcon` / `SiteStatistic` / `SiteUserData` | Torrent site records and statistics | | `Message` | Message log | | `PluginData` | Plugin-persisted data | +| `PluginIdentity` | Installed physical-plugin source binding and payload provenance | | `PassKey` | Passkey authentication records | | `Workflow` | Workflow definitions | @@ -61,6 +62,7 @@ directly in chain, module, or endpoint code. | `MediaServerOper` | `oper/mediaserver.py` | | `MessageOper` | `oper/message.py` | | `PluginDataOper` | `oper/plugindata.py` | +| `PluginIdentityOper` | `oper/pluginidentity.py` | | `SiteOper` | `oper/site.py` | | `SubscribeHistoryOper` | `oper/subscribehistory.py` | | `SubscribeOper` | `oper/subscribe.py` | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index f1a8dcb8f..15b8aa8b1 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6661, - "edge_sha256": "e276e75348004a89b7f10f3520ce5eb935130cf11a8b45dc371e79937f5d67e3", + "edge_count": 6676, + "edge_sha256": "4244dafa5ec5179e2cfbf005dca97f9dde678f9c45e780288bbbbaca36ff4de5", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -3624,6 +3624,15 @@ "app.db.adapters.outbox -> app.db.base", "app.db.adapters.outbox -> app.db.models", "app.db.adapters.outbox -> app.db.models.outbox", + "app.db.adapters.pluginidentity -> app.application", + "app.db.adapters.pluginidentity -> app.application.plugin", + "app.db.adapters.pluginidentity -> app.application.plugin.identity", + "app.db.adapters.pluginidentity -> app.db", + "app.db.adapters.pluginidentity -> app.db.models", + "app.db.adapters.pluginidentity -> app.db.models.pluginidentity", + "app.db.adapters.pluginidentity -> app.db.oper", + "app.db.adapters.pluginidentity -> app.db.oper.pluginidentity", + "app.db.adapters.pluginidentity -> app.db.uow", "app.db.adapters.site -> app.db", "app.db.adapters.site -> app.db.oper", "app.db.adapters.site -> app.db.oper.site", @@ -3712,6 +3721,8 @@ "app.db.models.passkey -> app.db.base", "app.db.models.plugindata -> app.db", "app.db.models.plugindata -> app.db.base", + "app.db.models.pluginidentity -> app.db", + "app.db.models.pluginidentity -> app.db.base", "app.db.models.site -> app.db", "app.db.models.site -> app.db.base", "app.db.models.siteicon -> app.db", @@ -3788,6 +3799,10 @@ "app.db.oper.plugindata -> app.db.base", "app.db.oper.plugindata -> app.db.models", "app.db.oper.plugindata -> app.db.models.plugindata", + "app.db.oper.pluginidentity -> app.db", + "app.db.oper.pluginidentity -> app.db.base", + "app.db.oper.pluginidentity -> app.db.models", + "app.db.oper.pluginidentity -> app.db.models.pluginidentity", "app.db.oper.site -> app.db", "app.db.oper.site -> app.db.base", "app.db.oper.site -> app.db.models", @@ -6678,7 +6693,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 818, + "module_count": 822, "modules": [ "app", "app.adapters", @@ -6973,6 +6988,7 @@ "app.application.plugin.config", "app.application.plugin.data", "app.application.plugin.folders", + "app.application.plugin.identity", "app.application.plugin.install", "app.application.plugin.lifecycle", "app.application.plugin.routes", @@ -7062,6 +7078,7 @@ "app.db.adapters.chain", "app.db.adapters.download", "app.db.adapters.outbox", + "app.db.adapters.pluginidentity", "app.db.adapters.site", "app.db.adapters.subscription", "app.db.adapters.transaction", @@ -7085,6 +7102,7 @@ "app.db.models.outbox", "app.db.models.passkey", "app.db.models.plugindata", + "app.db.models.pluginidentity", "app.db.models.site", "app.db.models.siteicon", "app.db.models.sitestatistic", @@ -7106,6 +7124,7 @@ "app.db.oper.message", "app.db.oper.passkey", "app.db.oper.plugindata", + "app.db.oper.pluginidentity", "app.db.oper.site", "app.db.oper.subscribe", "app.db.oper.subscribehistory", diff --git a/tests/test_plugin_identity.py b/tests/test_plugin_identity.py new file mode 100644 index 000000000..2d3ef94f7 --- /dev/null +++ b/tests/test_plugin_identity.py @@ -0,0 +1,435 @@ +"""插件身份事实、条件写和存量迁移决策测试。""" + +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace +from datetime import datetime, timedelta, timezone +import threading + +import pytest +import sqlalchemy as sa +from sqlalchemy.orm import sessionmaker + +from app.application.plugin.identity import ( + PluginBindingBasis, + PluginIdentity, + PluginIdentityConflictError, + PluginMarketAvailability, + PluginPayloadSourceType, + PluginSourceCandidate, + TrustedPluginSourceType, + plan_legacy_plugin_identity, +) +from app.db.adapters.pluginidentity import TransactionalPluginIdentityStore +from app.db.models import load_all_models +from app.db.models.pluginidentity import PluginIdentity as PluginIdentityModel + + +NOW = datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc) +OFFICIAL_SOURCE = "github:jxxghp/moviepilot-plugins" +THIRD_PARTY_SOURCE = "github:example/moviepilot-plugins" + + +def _identity( + plugin_id: str = "DemoPlugin", + *, + declared_version: str | None = None, +) -> PluginIdentity: + """构造一份已从官方仓成功安装的物理插件身份。""" + return PluginIdentity( + plugin_id=plugin_id, + normalized_plugin_id=plugin_id.lower(), + trusted_source_type=TrustedPluginSourceType.OFFICIAL, + trusted_source_key=OFFICIAL_SOURCE, + binding_basis=PluginBindingBasis.OFFICIAL_DEFAULT, + payload_source_type=PluginPayloadSourceType.OFFICIAL, + payload_source_key=OFFICIAL_SOURCE, + declared_version=declared_version or "1.0.0", + package_generation="v3", + system_version=None, + supports_v3=None, + supports_v3t=None, + payload_receipt="sha256:" + "0" * 64, + revision=1, + created_at=NOW, + updated_at=NOW, + bound_at=NOW, + payload_applied_at=NOW, + ) + + +@pytest.fixture +def identity_store(tmp_path): + """创建可跨线程竞争的独立 SQLite 身份表。""" + engine = sa.create_engine( + f"sqlite:///{tmp_path / 'plugin-identity.db'}", + connect_args={"check_same_thread": False}, + ) + PluginIdentityModel.__table__.create(engine) + factory = sessionmaker(bind=engine) + try: + yield TransactionalPluginIdentityStore(factory) + finally: + engine.dispose() + + +def test_plugin_identity_model_is_registered_by_composition_entry() -> None: + """组合根加载模型后必须包含插件身份表及物理 ID 唯一约束。""" + load_all_models() + + table = PluginIdentityModel.__table__ + unique_columns = { + tuple(constraint.columns.keys()) + for constraint in table.constraints + if isinstance(constraint, sa.UniqueConstraint) + } + assert table.name == "pluginidentity" + assert ("normalized_plugin_id",) in unique_columns + + +def test_plugin_identity_store_normalizes_reads_and_rejects_case_duplicate( + identity_store, +) -> None: + """同一物理目录的大小写别名只能对应数据库中的一行。""" + created = identity_store.compare_and_set( + _identity("DemoPlugin"), + expected_revision=None, + ) + + assert identity_store.get("DEMOPLUGIN") == created + with pytest.raises(PluginIdentityConflictError): + identity_store.compare_and_set( + _identity("demoplugin"), + expected_revision=None, + ) + + +def test_plugin_identity_store_rejects_stale_revision(identity_store) -> None: + """旧安装事务不能覆盖已经提交的新身份。""" + original = identity_store.compare_and_set( + _identity(), + expected_revision=None, + ) + updated = identity_store.compare_and_set( + replace( + original, + declared_version="2.0.0", + updated_at=NOW + timedelta(seconds=1), + ), + expected_revision=original.revision, + ) + + with pytest.raises(PluginIdentityConflictError): + identity_store.compare_and_set( + replace( + original, + declared_version="stale", + updated_at=NOW + timedelta(seconds=2), + ), + expected_revision=original.revision, + ) + + assert identity_store.get("DemoPlugin") == updated + assert updated.revision == 2 + + +def test_plugin_identity_store_rejects_implicit_source_change(identity_store) -> None: + """通用条件写不能代替管理员显式换源命令。""" + original = identity_store.compare_and_set( + _identity(), + expected_revision=None, + ) + changed = replace( + original, + trusted_source_type=TrustedPluginSourceType.THIRD_PARTY, + trusted_source_key=THIRD_PARTY_SOURCE, + binding_basis=PluginBindingBasis.EXPLICIT_INSTALL, + payload_source_type=PluginPayloadSourceType.THIRD_PARTY, + payload_source_key=THIRD_PARTY_SOURCE, + updated_at=NOW + timedelta(seconds=1), + ) + + with pytest.raises(PluginIdentityConflictError, match="不能改变"): + identity_store.compare_and_set( + changed, + expected_revision=original.revision, + ) + + assert identity_store.get("DemoPlugin") == original + + +def test_plugin_identity_store_allows_only_one_same_revision_writer( + identity_store, +) -> None: + """两个独立会话竞争同一 revision 时必须恰好一个提交成功。""" + original = identity_store.compare_and_set( + _identity(), + expected_revision=None, + ) + barrier = threading.Barrier(2) + + def update(version: str) -> str: + """在相同起点并发提交不同版本。""" + barrier.wait() + try: + identity_store.compare_and_set( + replace( + original, + declared_version=version, + updated_at=NOW + timedelta(seconds=1), + ), + expected_revision=original.revision, + ) + return "applied" + except PluginIdentityConflictError: + return "conflict" + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(update, ("2.0.0", "3.0.0"))) + + assert sorted(results) == ["applied", "conflict"] + assert identity_store.get("DemoPlugin").revision == 2 + + +def test_local_payload_preserves_trusted_online_binding() -> None: + """本地开发覆盖只改变载荷事实，不得抹掉可信在线更新仓库。""" + identity = replace( + _identity(), + payload_source_type=PluginPayloadSourceType.LOCAL, + payload_source_key=None, + declared_version="2.0.0-dev", + package_generation="v3", + system_version=">=3.0.0", + supports_v3=True, + supports_v3t=False, + payload_receipt="sha256:" + "a" * 64, + payload_applied_at=NOW + timedelta(seconds=1), + updated_at=NOW + timedelta(seconds=1), + ) + + assert identity.trusted_source_key == OFFICIAL_SOURCE + assert identity.payload_source_key is None + assert identity.payload_source_type is PluginPayloadSourceType.LOCAL + + +def test_first_local_sync_has_a_nonlegacy_identity_basis() -> None: + """首次本地同步不得被记录为未知存量插件迁移。""" + identity = replace( + _identity(), + trusted_source_type=TrustedPluginSourceType.UNKNOWN, + trusted_source_key=None, + binding_basis=PluginBindingBasis.LOCAL_ONLY, + payload_source_type=PluginPayloadSourceType.LOCAL, + payload_source_key=None, + bound_at=None, + ) + + assert identity.binding_basis is PluginBindingBasis.LOCAL_ONLY + assert identity.trusted_source_key is None + + +def test_online_binding_rejects_local_only_basis() -> None: + """已绑定在线仓库不能冒用首次本地同步的身份依据。""" + with pytest.raises(ValueError, match="未绑定来源依据"): + replace(_identity(), binding_basis=PluginBindingBasis.LOCAL_ONLY) + + +@pytest.mark.parametrize( + ("field_name", "value"), + ( + ("bound_at", NOW - timedelta(seconds=1)), + ("bound_at", NOW + timedelta(seconds=1)), + ("payload_applied_at", NOW - timedelta(seconds=1)), + ("payload_applied_at", NOW + timedelta(seconds=1)), + ), +) +def test_plugin_identity_rejects_audit_times_outside_record_lifetime( + field_name, + value, +) -> None: + """来源审计时间必须落在该 revision 的创建和更新时间范围内。""" + with pytest.raises(ValueError, match="审计时间"): + replace(_identity(), **{field_name: value}) + + +def test_unknown_payload_rejects_version_and_receipt_evidence() -> None: + """存量载荷来源未知时不得保留看似经过安装确认的版本证据。""" + with pytest.raises(ValueError, match="未知载荷不能携带"): + replace( + _identity(), + payload_source_type=PluginPayloadSourceType.UNKNOWN, + payload_source_key=None, + payload_applied_at=None, + ) + + +def test_online_payload_must_match_trusted_source() -> None: + """在线载荷来源与可信更新仓库不一致时身份必须失败关闭。""" + with pytest.raises(ValueError, match="必须与可信更新来源一致"): + replace( + _identity(), + payload_source_type=PluginPayloadSourceType.THIRD_PARTY, + payload_source_key=THIRD_PARTY_SOURCE, + ) + + +@pytest.mark.parametrize( + ("source_type", "source_key"), + ( + (TrustedPluginSourceType.OFFICIAL, THIRD_PARTY_SOURCE), + (TrustedPluginSourceType.THIRD_PARTY, OFFICIAL_SOURCE), + ), +) +def test_trusted_source_type_must_match_official_repository( + source_type, + source_key, +) -> None: + """官方仓库键与官方来源类型不能形成相互矛盾的信任事实。""" + with pytest.raises(ValueError, match="官方来源类型"): + replace( + _identity(), + trusted_source_type=source_type, + trusted_source_key=source_key, + ) + + +@pytest.mark.parametrize( + ("source_type", "source_key"), + ( + (TrustedPluginSourceType.OFFICIAL, THIRD_PARTY_SOURCE), + (TrustedPluginSourceType.THIRD_PARTY, OFFICIAL_SOURCE), + ), +) +def test_source_candidate_type_must_match_official_repository( + source_type, + source_key, +) -> None: + """迁移候选也必须服从与持久化身份相同的来源分类合同。""" + with pytest.raises(ValueError, match="官方来源类型"): + PluginSourceCandidate(source_type, source_key) + + +@pytest.mark.parametrize( + "plugin_id", + (" DemoPlugin", "DemoPlugin ", "A" * 129), +) +def test_plugin_identity_rejects_noncanonical_physical_id(plugin_id) -> None: + """物理 ID 不得靠静默裁剪或数据库方言差异改变身份。""" + with pytest.raises(ValueError, match="插件 ID"): + _identity(plugin_id) + + +@pytest.mark.parametrize( + ("field_name", "value", "message"), + ( + ("declared_version", "v" * 65, "插件声明版本"), + ("system_version", ">" * 129, "插件系统版本要求"), + ), +) +def test_plugin_identity_rejects_values_longer_than_database_columns( + field_name, + value, + message, +) -> None: + """SQLite 与 PostgreSQL 必须在进入持久化前共享相同长度边界。""" + with pytest.raises(ValueError, match=message): + replace(_identity(), **{field_name: value}) + + +def test_legacy_official_candidate_binds_updates_but_not_payload() -> None: + """官方默认只建立未来更新绑定，不能冒充存量载荷来源。""" + identity = plan_legacy_plugin_identity( + plugin_id="DemoPlugin", + market_availability=PluginMarketAvailability.AVAILABLE, + online_candidates=( + PluginSourceCandidate( + TrustedPluginSourceType.THIRD_PARTY, + THIRD_PARTY_SOURCE, + ), + PluginSourceCandidate( + TrustedPluginSourceType.OFFICIAL, + OFFICIAL_SOURCE, + ), + ), + is_virtual_instance=False, + now=NOW, + ) + + assert identity.trusted_source_type is TrustedPluginSourceType.OFFICIAL + assert identity.trusted_source_key == OFFICIAL_SOURCE + assert identity.binding_basis is PluginBindingBasis.OFFICIAL_DEFAULT + assert identity.payload_source_type is PluginPayloadSourceType.UNKNOWN + assert identity.declared_version is None + + +def test_legacy_single_third_party_candidate_uses_tofu() -> None: + """唯一第三方候选可建立一次性更新绑定，但载荷仍保持未知。""" + identity = plan_legacy_plugin_identity( + plugin_id="DemoPlugin", + market_availability=PluginMarketAvailability.AVAILABLE, + online_candidates=( + PluginSourceCandidate( + TrustedPluginSourceType.THIRD_PARTY, + THIRD_PARTY_SOURCE, + ), + ), + is_virtual_instance=False, + now=NOW, + ) + + assert identity.binding_basis is PluginBindingBasis.TOFU + assert identity.trusted_source_key == THIRD_PARTY_SOURCE + assert identity.payload_source_type is PluginPayloadSourceType.UNKNOWN + + +@pytest.mark.parametrize( + ("availability", "candidates"), + ( + (PluginMarketAvailability.UNAVAILABLE, ()), + (PluginMarketAvailability.AVAILABLE, ()), + ( + PluginMarketAvailability.AVAILABLE, + ( + PluginSourceCandidate( + TrustedPluginSourceType.THIRD_PARTY, + "github:first/plugins", + ), + PluginSourceCandidate( + TrustedPluginSourceType.THIRD_PARTY, + "github:second/plugins", + ), + ), + ), + ), +) +def test_legacy_unavailable_empty_or_ambiguous_market_stays_unbound( + availability, + candidates, +) -> None: + """离线、无候选和多候选必须保持可区分输入，但均不得猜测绑定。""" + identity = plan_legacy_plugin_identity( + plugin_id="DemoPlugin", + market_availability=availability, + online_candidates=candidates, + is_virtual_instance=False, + now=NOW, + ) + + assert identity.trusted_source_type is TrustedPluginSourceType.UNKNOWN + assert identity.trusted_source_key is None + assert identity.binding_basis is PluginBindingBasis.LEGACY_UNBOUND + + +def test_virtual_instance_does_not_create_independent_plugin_identity() -> None: + """有效 PluginInstances 派生实例只继承物理宿主身份。""" + assert plan_legacy_plugin_identity( + plugin_id="DemoPluginWork", + market_availability=PluginMarketAvailability.AVAILABLE, + online_candidates=( + PluginSourceCandidate( + TrustedPluginSourceType.OFFICIAL, + OFFICIAL_SOURCE, + ), + ), + is_virtual_instance=True, + now=NOW, + ) is None diff --git a/tests/test_plugin_identity_migration.py b/tests/test_plugin_identity_migration.py new file mode 100644 index 000000000..8309fd85f --- /dev/null +++ b/tests/test_plugin_identity_migration.py @@ -0,0 +1,205 @@ +"""插件身份表 Alembic 迁移测试。""" + +import importlib +import os +import uuid + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.dialects import postgresql +from sqlalchemy.schema import CreateTable + +try: + import psycopg2 as postgres_driver + from psycopg2 import sql + POSTGRESQL_DIALECT = "postgresql+psycopg2" +except ModuleNotFoundError: + import psycopg as postgres_driver + from psycopg import sql + POSTGRESQL_DIALECT = "postgresql+psycopg" + +from app.db.models.pluginidentity import PluginIdentity + + +MIGRATION = "database.versions.d2e4f6a8b0c1_3_0_9" + + +def _bind_migration(monkeypatch, connection): + """把迁移绑定到隔离数据库连接。""" + migration = importlib.import_module(MIGRATION) + context = MigrationContext.configure(connection) + monkeypatch.setattr(migration, "op", Operations(context)) + return migration + + +def test_plugin_identity_migration_upgrades_twice_and_downgrades( + monkeypatch, +) -> None: + """旧 SQLite schema 应可重复升级并完整删除 dormant 身份表。""" + engine = sa.create_engine("sqlite://") + with engine.begin() as connection: + migration = _bind_migration(monkeypatch, connection) + + migration.upgrade() + migration.upgrade() + + inspector = sa.inspect(connection) + assert "pluginidentity" in inspector.get_table_names() + columns = { + column["name"] for column in inspector.get_columns("pluginidentity") + } + assert columns == {column.name for column in PluginIdentity.__table__.columns} + unique_constraints = { + constraint["name"]: tuple(constraint["column_names"]) + for constraint in inspector.get_unique_constraints("pluginidentity") + } + assert unique_constraints["uq_pluginidentity_normalized_plugin_id"] == ( + "normalized_plugin_id", + ) + check_constraints = { + constraint["name"] + for constraint in inspector.get_check_constraints("pluginidentity") + } + assert check_constraints == { + "ck_pluginidentity_normalized_plugin_id", + "ck_pluginidentity_revision", + } + + migration.downgrade() + assert "pluginidentity" not in sa.inspect(connection).get_table_names() + + +def test_plugin_identity_migration_accepts_fresh_current_schema(monkeypatch) -> None: + """create_all 已建当前表时重复升级不得创建冲突对象。""" + engine = sa.create_engine("sqlite://") + with engine.begin() as connection: + PluginIdentity.__table__.create(connection) + migration = _bind_migration(monkeypatch, connection) + + migration.upgrade() + migration.upgrade() + + assert { + column["name"] + for column in sa.inspect(connection).get_columns("pluginidentity") + } == {column.name for column in PluginIdentity.__table__.columns} + + +def test_plugin_identity_migration_matches_postgresql_identity() -> None: + """独立 Alembic 路径应保留 PostgreSQL 循环 Identity 主键。""" + migration = importlib.import_module(MIGRATION) + metadata = sa.MetaData() + table = sa.Table( + "pluginidentity", + metadata, + migration._id_column("postgresql"), + sa.PrimaryKeyConstraint("id"), + ) + + identity = table.c.id.identity + assert identity is not None + assert identity.start == 1 + assert identity.cycle is True + ddl = str(CreateTable(table).compile(dialect=postgresql.dialect())) + assert "GENERATED BY DEFAULT AS IDENTITY" in ddl + assert "CYCLE" in ddl + + +def test_plugin_identity_migration_runs_on_postgresql(monkeypatch) -> None: + """已配置的隔离 PostgreSQL 应执行真实建表、约束和回滚。""" + prefix = "MOVIEPILOT_TEST_POSTGRESQL_" + host = os.getenv(f"{prefix}HOST") + database = os.getenv(f"{prefix}DATABASE") + username = os.getenv(f"{prefix}USERNAME") + if not host or not database or not username: + pytest.skip("未配置隔离 PostgreSQL migration 测试库") + + port = os.getenv(f"{prefix}PORT", "5432") + password = os.getenv(f"{prefix}PASSWORD", "") + schema = f"plugin_identity_{uuid.uuid4().hex}" + with postgres_driver.connect( + host=host, + port=port, + dbname=database, + user=username, + password=password, + ) as connection: + connection.autocommit = True + with connection.cursor() as cursor: + cursor.execute( + sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(schema)) + ) + + engine = None + try: + engine = sa.create_engine( + sa.URL.create( + POSTGRESQL_DIALECT, + username=username, + password=password, + host=host, + port=int(port), + database=database, + ), + connect_args={"options": f"-csearch_path={schema}"}, + ) + with engine.begin() as connection: + migration = _bind_migration(monkeypatch, connection) + migration.upgrade() + migration.upgrade() + + table = sa.Table( + "pluginidentity", + sa.MetaData(), + autoload_with=connection, + ) + inserted_id = connection.execute( + table.insert().values( + plugin_id="DemoPlugin", + normalized_plugin_id="demoplugin", + trusted_source_type="unknown", + binding_basis="legacy_unbound", + payload_source_type="unknown", + revision=1, + created_at="2026-08-25T12:00:00+00:00", + updated_at="2026-08-25T12:00:00+00:00", + ).returning(table.c.id) + ).scalar_one() + assert inserted_id == 1 + + with pytest.raises(sa.exc.IntegrityError): + with connection.begin_nested(): + connection.execute( + table.insert().values( + plugin_id="UppercaseKey", + normalized_plugin_id="UppercaseKey", + trusted_source_type="unknown", + binding_basis="legacy_unbound", + payload_source_type="unknown", + revision=1, + created_at="2026-08-25T12:00:00+00:00", + updated_at="2026-08-25T12:00:00+00:00", + ) + ) + + migration.downgrade() + assert "pluginidentity" not in sa.inspect(connection).get_table_names() + finally: + if engine is not None: + engine.dispose() + with postgres_driver.connect( + host=host, + port=port, + dbname=database, + user=username, + password=password, + ) as connection: + connection.autocommit = True + with connection.cursor() as cursor: + cursor.execute( + sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format( + sql.Identifier(schema) + ) + )