diff --git a/packages/syft-datasets/src/syft_datasets/dataset_manager.py b/packages/syft-datasets/src/syft_datasets/dataset_manager.py index 28c9383da9c..686a62c9517 100644 --- a/packages/syft-datasets/src/syft_datasets/dataset_manager.py +++ b/packages/syft-datasets/src/syft_datasets/dataset_manager.py @@ -1,7 +1,9 @@ from pathlib import Path + from typing_extensions import Self import yaml +from syft_migration import ProtocolSchema from .types import PathLike, to_path from syft_notebook_ui.types import TableList @@ -20,18 +22,34 @@ class SyftDatasetManager: - def __init__(self, syftbox_folder_path: PathLike, email: str): + def __init__( + self, + syftbox_folder_path: PathLike, + email: str, + peer_schemas: dict[str, ProtocolSchema] | None = None, + ): self.syftbox_config = SyftBoxConfig( syftbox_folder=to_path(syftbox_folder_path), email=email ) - # peer_schemas (peer email -> dataset ProtocolSchema) will be filled in by - # syft-client later; until then every peer resolves to the widest- - # compatible protocol, so datasets are written in that layout. - self.storage = DatasetStorage(config=self.syftbox_config) + # peer_schemas (peer email -> dataset ProtocolSchema): syft-client + # passes PeerManager's live map here (updated in place as peer version + # files load). Peers without an entry resolve to the widest-compatible + # protocol, so datasets stay readable by unknown-version peers. + self.storage = DatasetStorage( + config=self.syftbox_config, peer_schemas=peer_schemas + ) @classmethod - def from_config(cls, config: SyftBoxConfig) -> Self: - return cls(syftbox_folder_path=config.syftbox_folder, email=config.email) + def from_config( + cls, + config: SyftBoxConfig, + peer_schemas: dict[str, ProtocolSchema] | None = None, + ) -> Self: + return cls( + syftbox_folder_path=config.syftbox_folder, + email=config.email, + peer_schemas=peer_schemas, + ) def create( self, diff --git a/packages/syft-datasets/src/syft_datasets/dataset_storage.py b/packages/syft-datasets/src/syft_datasets/dataset_storage.py index a6c0bcbfb65..fe06581ad3d 100644 --- a/packages/syft-datasets/src/syft_datasets/dataset_storage.py +++ b/packages/syft-datasets/src/syft_datasets/dataset_storage.py @@ -98,10 +98,15 @@ def __init__( self.config = config self.registry = registry self.service = MigrationService(registry=registry) - # peer email -> dataset ProtocolSchema; filled in by syft-client later. - # Peers without an entry cannot be assumed to read the current layout, so + # peer email -> dataset ProtocolSchema; syft-client passes PeerManager's + # live map here (updated in place as peer version files load). Peers + # without an entry cannot be assumed to read the current layout, so # they resolve to the widest-compatible (oldest) protocol. - self.peer_schemas: dict[str, ProtocolSchema] = peer_schemas or {} + # `is not None`, not `or`: the live dict starts empty and `or {}` would + # drop the shared reference, freezing negotiation at construction time. + self.peer_schemas: dict[str, ProtocolSchema] = ( + peer_schemas if peer_schemas is not None else {} + ) self.codecs = [cls(config) for cls in CODECS] @property diff --git a/packages/syft-job/src/syft_job/client.py b/packages/syft-job/src/syft_job/client.py index 05f958a7584..5af9830b5b2 100644 --- a/packages/syft-job/src/syft_job/client.py +++ b/packages/syft-job/src/syft_job/client.py @@ -81,8 +81,12 @@ def __init__( self._validate_user_email() @classmethod - def from_config(cls, config: SyftJobConfig) -> "JobClient": - return cls(config, config.current_user_email) + def from_config( + cls, + config: SyftJobConfig, + peer_schemas: Optional[dict[str, ProtocolSchema]] = None, + ) -> "JobClient": + return cls(config, config.current_user_email, peer_schemas=peer_schemas) def _validate_user_email(self) -> None: """Validate that the user_email directory exists in SyftBox root.""" diff --git a/packages/syft-job/src/syft_job/job_storage.py b/packages/syft-job/src/syft_job/job_storage.py index b0a85c4a4df..5507841e628 100644 --- a/packages/syft-job/src/syft_job/job_storage.py +++ b/packages/syft-job/src/syft_job/job_storage.py @@ -37,9 +37,15 @@ def __init__( self.config = config self.registry = registry self.service = MigrationService(registry=registry) - # peer email -> job ProtocolSchema; filled in by syft-client later. - # Peers without an entry are assumed to run the current protocol. - self.peer_schemas: dict[str, ProtocolSchema] = peer_schemas or {} + # peer email -> job ProtocolSchema; syft-client passes PeerManager's + # live map here (updated in place as peer version files load). Peers + # without an entry are assumed to run the current protocol. + # `is not None`, not `or`: syft-client passes a live (initially empty) + # dict it mutates as peer version files load; `or {}` would drop the + # shared reference and freeze negotiation at construction-time state. + self.peer_schemas: dict[str, ProtocolSchema] = ( + peer_schemas if peer_schemas is not None else {} + ) self.codecs = [cls(config) for cls in CODECS] @property @@ -106,8 +112,8 @@ def new_submission_ref(self, do_email: str, job_name: str) -> JobRef: datasite_email=do_email, ds_email=self.config.current_user_email, job_name=job_name, - # Until syft-client fills peer_schemas, unknown peers are assumed - # to run the current protocol. + # Peers without a known schema are assumed to run the current + # protocol. protocol_version=self.negotiated_protocol_version_for_peer( do_email, raise_on_unknown=False ), diff --git a/pyproject.toml b/pyproject.toml index 40c9fb98c24..e41ae92c98b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,6 +42,7 @@ dependencies = [ "syft-dataset==0.1.20", # TODO: move to optional-dependencies after adding lazy imports "syft-permissions==0.1.14", "syft-perms==0.1.14", + "syft-migration", "syft-crypto-python>=0.1.2b2", ] diff --git a/scripts/export_release_artifact.py b/scripts/export_release_artifact.py new file mode 100644 index 00000000000..f1c686fe0c7 --- /dev/null +++ b/scripts/export_release_artifact.py @@ -0,0 +1,46 @@ +"""Export the release artifacts for the current syft-client version. + +Run on EVERY release (uv run python scripts/export_release_artifact.py): +always writes the package release info; additionally writes the protocol +artifact when this release introduces a new protocol version. +""" + +import sys + +from syft_client.migrations.history import PACKAGE_ARTIFACTS_DIR, PROTOCOLS_DIR +from syft_client.migrations.registry import ( + SYFT_CLIENT_PROTOCOL_VERSION, + client_registry, +) +from syft_client.version import SYFT_CLIENT_VERSION + + +def main() -> None: + # Import the package so every versioned object is registered. + import syft_client # noqa: F401 + + if client_registry.protocol_changed_without_bump(): + sys.exit( + "The syft-client protocol changed compared to the released " + f"protocol-{SYFT_CLIENT_PROTOCOL_VERSION}.json; bump " + "SYFT_CLIENT_PROTOCOL_VERSION in syft_client/migrations/registry.py " + "before releasing." + ) + + info_path = PACKAGE_ARTIFACTS_DIR / f"syft-client-{SYFT_CLIENT_VERSION}.json" + if info_path.exists(): + sys.exit( + f"{info_path} already exists — release artifacts are frozen once " + "written. Bump SYFT_CLIENT_VERSION before exporting." + ) + client_registry.compute_released_package_protocol_info().save(info_path) + print(f"Wrote {info_path}") + + protocol_path = PROTOCOLS_DIR / f"protocol-{SYFT_CLIENT_PROTOCOL_VERSION}.json" + if not protocol_path.exists(): + client_registry.compute_released_protocol().save(protocol_path) + print(f"Wrote {protocol_path} (new protocol version)") + + +if __name__ == "__main__": + main() diff --git a/scripts/generate_release_fixture.py b/scripts/generate_release_fixture.py new file mode 100644 index 00000000000..7de720904a4 --- /dev/null +++ b/scripts/generate_release_fixture.py @@ -0,0 +1,117 @@ +"""Generate a p2p backward-compatibility fixture for the current syft-client release. + +Run on EVERY release, after bumping the version: + + uv run python scripts/generate_release_fixture.py + +Writes the serialized artifacts exactly as this release produces them, into + + tests/migrations/p2p/fixtures/syft_client--protocol

/ + SYFT_version.json # the published version file + msgv2_<...>.tar.gz # a proposed-changes message (DS -> DO) + syfteventsmessagev3_<...>.tar.gz # an events message (DO -> watchers) + +Unlike syft-job there is no local SyftBox tree to snapshot (storage is the +Google Drive transport), so fixtures are directories of captured blobs; future +releases loop over them (test_older_protocol_compatibility.py) to prove they +can still read and round-trip older serialized data. + +Protocol 0 / release 0.1.117 predates this script; its fixture +(syft_client-0.1.117-protocol0) is hand-authored, like protocol-0.json. +""" + +import sys +from pathlib import Path + +from syft_client.migrations.registry import SYFT_CLIENT_PROTOCOL_VERSION +from syft_client.sync.events.file_change_event import ( + FileChangeEvent, + FileChangeEventsMessage, +) +from syft_client.sync.messages.proposed_filechange import ( + ProposedFileChange, + ProposedFileChangesMessage, +) +from syft_client.sync.version.version_info import VersionInfo +from syft_client.version import SYFT_CLIENT_VERSION + +DO_EMAIL = "do@test.org" +DS_EMAIL = "ds@test.org" + +FIXTURES_DIR = Path(__file__).resolve().parents[1] / "tests" / "migrations" / "p2p" / "fixtures" + + +def build_version_info() -> VersionInfo: + # Not VersionInfo.current(): the detected install source is an absolute + # local path on dev machines, which must not leak into a committed fixture. + return VersionInfo.current().model_copy( + update={"syft_client_install_source": "pip"} + ) + + +def build_proposed_message() -> ProposedFileChangesMessage: + return ProposedFileChangesMessage( + sender_email=DS_EMAIL, + proposed_file_changes=[ + ProposedFileChange( + path_in_datasite="data/notes.txt", + content="hello from the release fixture", + datasite_email=DO_EMAIL, + ), + ProposedFileChange( + path_in_datasite="data/blob.bin", + content=b"\x00\x01\x02fixture-binary", + datasite_email=DO_EMAIL, + ), + ProposedFileChange( + path_in_datasite="data/removed.txt", + content=None, + old_hash="0" * 64, + is_deleted=True, + datasite_email=DO_EMAIL, + ), + ], + ) + + +def build_events_message( + proposed: ProposedFileChangesMessage, +) -> FileChangeEventsMessage: + events = [ + FileChangeEvent.from_proposed_filechange(change) + for change in proposed.proposed_file_changes + ] + return FileChangeEventsMessage(events=events) + + +def main() -> None: + # Any fixture for this version (any protocol) means the version was + # already released; a released version's serialized form is frozen. + existing = sorted(FIXTURES_DIR.glob(f"syft_client-{SYFT_CLIENT_VERSION}-protocol*")) + if existing: + sys.exit( + f"{existing[0]} already exists — fixtures are frozen once written. " + "Bump SYFT_CLIENT_VERSION before generating." + ) + target = FIXTURES_DIR / ( + f"syft_client-{SYFT_CLIENT_VERSION}-protocol{SYFT_CLIENT_PROTOCOL_VERSION}" + ) + target.mkdir(parents=True) + + (target / "SYFT_version.json").write_text(build_version_info().to_json()) + + proposed = build_proposed_message() + (target / proposed.message_filename.as_string()).write_bytes( + proposed.as_compressed_data() + ) + + events = build_events_message(proposed) + (target / events.message_filepath.as_string()).write_bytes( + events.as_compressed_data() + ) + + print(f"Wrote {target}") + + +if __name__ == "__main__": + main() diff --git a/syft_client/__init__.py b/syft_client/__init__.py index ca2fa8a09cd..f2a24c2906a 100644 --- a/syft_client/__init__.py +++ b/syft_client/__init__.py @@ -35,6 +35,11 @@ delete_syftbox, delete_local_syftbox, ) +from syft_client.migrations.history import register_historic_schemas # noqa: E402 + +# Historic schemas list object versions that must already be registered, which +# happens when the model modules are imported (transitively via sync.login). +register_historic_schemas() SYFT_CLIENT_DIR = Path(__file__).parent.parent CREDENTIALS_DIR = SYFT_CLIENT_DIR / "credentials" diff --git a/syft_client/migrations/__init__.py b/syft_client/migrations/__init__.py new file mode 100644 index 00000000000..fa713f56a40 --- /dev/null +++ b/syft_client/migrations/__init__.py @@ -0,0 +1,15 @@ +from .registry import ( + PROTOCOL_NAME, + SYFT_CLIENT_PROTOCOL_VERSION, + client_migration_service, + client_registry, + load_as_latest, +) + +__all__ = [ + "PROTOCOL_NAME", + "SYFT_CLIENT_PROTOCOL_VERSION", + "client_migration_service", + "client_registry", + "load_as_latest", +] diff --git a/syft_client/migrations/history.py b/syft_client/migrations/history.py new file mode 100644 index 00000000000..415d76a9b7e --- /dev/null +++ b/syft_client/migrations/history.py @@ -0,0 +1,32 @@ +from pathlib import Path + +from syft_migration import ReleasedPackageProtocolInfo, ReleasedProtocol + +from .registry import client_registry + +# Release artifacts of past syft-client releases: +# package-artifacts/syft-client-.json (every release) +# protocols/protocol-.json (only when the protocol changed) +# Generated by scripts/export_release_artifact.py; 0.1.117 / protocol 0 predate +# the artifact mechanism, so their files are hardcoded as if that release had +# emitted them. +HISTORY_DIR = Path(__file__).parent / "history" +PACKAGE_ARTIFACTS_DIR = HISTORY_DIR / "package-artifacts" +PROTOCOLS_DIR = HISTORY_DIR / "protocols" + + +def register_historic_schemas() -> None: + """Register the release artifacts of past releases into the client registry. + + Must run after the versioned models are imported: with + ``raise_for_unknown_objects`` an artifact listing an object version this + release cannot load fails at import time instead of at migration time. + """ + for path in sorted(PACKAGE_ARTIFACTS_DIR.glob("*.json")): + client_registry.register_released_package_protocol_info( + ReleasedPackageProtocolInfo.load(path), raise_for_unknown_objects=True + ) + for path in sorted(PROTOCOLS_DIR.glob("*.json")): + client_registry.register_released_protocol( + ReleasedProtocol.load(path), raise_for_unknown_objects=True + ) diff --git a/syft_client/migrations/history/package-artifacts/syft-client-0.1.117.json b/syft_client/migrations/history/package-artifacts/syft-client-0.1.117.json new file mode 100644 index 00000000000..c95eb8f75f2 --- /dev/null +++ b/syft_client/migrations/history/package-artifacts/syft-client-0.1.117.json @@ -0,0 +1,365 @@ +{ + "package_info": { + "package_name": "syft-client", + "version": "0.1.117", + "protocol_version": "0" + }, + "protocol_schema": { + "protocol_name": "syft-client", + "version": "0", + "supported_versions": { + "VersionInfo": [ + "1" + ], + "ProposedFileChangesMessage": [ + "1" + ], + "FileChangeEventsMessage": [ + "1" + ] + }, + "current_object_schemas": { + "VersionInfo": { + "description": "Model representing version information for a syft client.\n\nStored as SYFT_version.json in the peer-visible SyftBox folder. This file\nis the bootstrap channel for protocol negotiation (peers read it to learn\nwhat we speak), so its schema may only ever change additively: every\nsupported client version must be able to parse every newer version file.", + "properties": { + "canonical_name": { + "default": "VersionInfo", + "title": "Canonical Name", + "type": "string" + }, + "version": { + "default": "1", + "title": "Version", + "type": "string" + }, + "syft_client_version": { + "title": "Syft Client Version", + "type": "string" + }, + "min_supported_syft_client_version": { + "title": "Min Supported Syft Client Version", + "type": "string" + }, + "protocol_version": { + "title": "Protocol Version", + "type": "string" + }, + "min_supported_protocol_version": { + "title": "Min Supported Protocol Version", + "type": "string" + }, + "syft_client_install_source": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Syft Client Install Source" + }, + "updated_at": { + "format": "date-time", + "title": "Updated At", + "type": "string" + }, + "attestation_token": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Attestation Token" + } + }, + "required": [ + "syft_client_version", + "min_supported_syft_client_version", + "protocol_version", + "min_supported_protocol_version" + ], + "title": "VersionInfoV1", + "type": "object" + }, + "ProposedFileChangesMessage": { + "$defs": { + "MessageFileName": { + "properties": { + "submitted_timestamp": { + "title": "Submitted Timestamp", + "type": "number" + }, + "uid": { + "title": "Uid", + "type": "string" + } + }, + "title": "MessageFileName", + "type": "object" + }, + "ProposedFileChangeV1": { + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "old_hash": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Old Hash" + }, + "new_hash": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "New Hash" + }, + "submitted_timestamp": { + "title": "Submitted Timestamp", + "type": "number" + }, + "path_in_datasite": { + "format": "path", + "title": "Path In Datasite", + "type": "string" + }, + "content": { + "anyOf": [ + { + "type": "string" + }, + { + "format": "binary", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Content" + }, + "datasite_email": { + "title": "Datasite Email", + "type": "string" + }, + "is_deleted": { + "default": false, + "title": "Is Deleted", + "type": "boolean" + } + }, + "required": [ + "path_in_datasite", + "datasite_email" + ], + "title": "ProposedFileChangeV1", + "type": "object" + } + }, + "description": "The msgv2 wire envelope (DS -> DO). The envelope is the migratable unit;\nits items are pinned to the exact version class, never a floating alias.", + "properties": { + "canonical_name": { + "default": "ProposedFileChangesMessage", + "title": "Canonical Name", + "type": "string" + }, + "version": { + "default": "1", + "title": "Version", + "type": "string" + }, + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "sender_email": { + "title": "Sender Email", + "type": "string" + }, + "message_filename": { + "$ref": "#/$defs/MessageFileName" + }, + "proposed_file_changes": { + "items": { + "$ref": "#/$defs/ProposedFileChangeV1" + }, + "title": "Proposed File Changes", + "type": "array" + }, + "platform_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Platform Id" + } + }, + "required": [ + "sender_email", + "proposed_file_changes" + ], + "title": "ProposedFileChangesMessageV1", + "type": "object" + }, + "FileChangeEventsMessage": { + "$defs": { + "FileChangeEventV1": { + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "path_in_datasite": { + "format": "path", + "title": "Path In Datasite", + "type": "string" + }, + "datasite_email": { + "title": "Datasite Email", + "type": "string" + }, + "content": { + "anyOf": [ + { + "type": "string" + }, + { + "format": "binary", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Content" + }, + "old_hash": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Old Hash" + }, + "new_hash": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "New Hash" + }, + "is_deleted": { + "default": false, + "title": "Is Deleted", + "type": "boolean" + }, + "submitted_timestamp": { + "title": "Submitted Timestamp", + "type": "number" + }, + "timestamp": { + "title": "Timestamp", + "type": "number" + } + }, + "required": [ + "id", + "path_in_datasite", + "datasite_email", + "submitted_timestamp", + "timestamp" + ], + "title": "FileChangeEventV1", + "type": "object" + }, + "FileChangeEventsMessageFileName": { + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "timestamp": { + "title": "Timestamp", + "type": "number" + }, + "extension": { + "default": ".tar.gz", + "title": "Extension", + "type": "string" + } + }, + "title": "FileChangeEventsMessageFileName", + "type": "object" + } + }, + "description": "The events wire envelope (DO -> watchers). The envelope is the migratable\nunit; its items are pinned to the exact version class, never a floating alias.", + "properties": { + "canonical_name": { + "default": "FileChangeEventsMessage", + "title": "Canonical Name", + "type": "string" + }, + "version": { + "default": "1", + "title": "Version", + "type": "string" + }, + "events": { + "items": { + "$ref": "#/$defs/FileChangeEventV1" + }, + "title": "Events", + "type": "array" + }, + "message_filepath": { + "$ref": "#/$defs/FileChangeEventsMessageFileName" + } + }, + "required": [ + "events" + ], + "title": "FileChangeEventsMessageV1", + "type": "object" + } + } + } +} \ No newline at end of file diff --git a/syft_client/migrations/history/protocols/protocol-0.json b/syft_client/migrations/history/protocols/protocol-0.json new file mode 100644 index 00000000000..2b50bfc9b54 --- /dev/null +++ b/syft_client/migrations/history/protocols/protocol-0.json @@ -0,0 +1,360 @@ +{ + "protocol_schema": { + "protocol_name": "syft-client", + "version": "0", + "supported_versions": { + "VersionInfo": [ + "1" + ], + "ProposedFileChangesMessage": [ + "1" + ], + "FileChangeEventsMessage": [ + "1" + ] + }, + "current_object_schemas": { + "VersionInfo": { + "description": "Model representing version information for a syft client.\n\nStored as SYFT_version.json in the peer-visible SyftBox folder. This file\nis the bootstrap channel for protocol negotiation (peers read it to learn\nwhat we speak), so its schema may only ever change additively: every\nsupported client version must be able to parse every newer version file.", + "properties": { + "canonical_name": { + "default": "VersionInfo", + "title": "Canonical Name", + "type": "string" + }, + "version": { + "default": "1", + "title": "Version", + "type": "string" + }, + "syft_client_version": { + "title": "Syft Client Version", + "type": "string" + }, + "min_supported_syft_client_version": { + "title": "Min Supported Syft Client Version", + "type": "string" + }, + "protocol_version": { + "title": "Protocol Version", + "type": "string" + }, + "min_supported_protocol_version": { + "title": "Min Supported Protocol Version", + "type": "string" + }, + "syft_client_install_source": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Syft Client Install Source" + }, + "updated_at": { + "format": "date-time", + "title": "Updated At", + "type": "string" + }, + "attestation_token": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Attestation Token" + } + }, + "required": [ + "syft_client_version", + "min_supported_syft_client_version", + "protocol_version", + "min_supported_protocol_version" + ], + "title": "VersionInfoV1", + "type": "object" + }, + "ProposedFileChangesMessage": { + "$defs": { + "MessageFileName": { + "properties": { + "submitted_timestamp": { + "title": "Submitted Timestamp", + "type": "number" + }, + "uid": { + "title": "Uid", + "type": "string" + } + }, + "title": "MessageFileName", + "type": "object" + }, + "ProposedFileChangeV1": { + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "old_hash": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Old Hash" + }, + "new_hash": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "New Hash" + }, + "submitted_timestamp": { + "title": "Submitted Timestamp", + "type": "number" + }, + "path_in_datasite": { + "format": "path", + "title": "Path In Datasite", + "type": "string" + }, + "content": { + "anyOf": [ + { + "type": "string" + }, + { + "format": "binary", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Content" + }, + "datasite_email": { + "title": "Datasite Email", + "type": "string" + }, + "is_deleted": { + "default": false, + "title": "Is Deleted", + "type": "boolean" + } + }, + "required": [ + "path_in_datasite", + "datasite_email" + ], + "title": "ProposedFileChangeV1", + "type": "object" + } + }, + "description": "The msgv2 wire envelope (DS -> DO). The envelope is the migratable unit;\nits items are pinned to the exact version class, never a floating alias.", + "properties": { + "canonical_name": { + "default": "ProposedFileChangesMessage", + "title": "Canonical Name", + "type": "string" + }, + "version": { + "default": "1", + "title": "Version", + "type": "string" + }, + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "sender_email": { + "title": "Sender Email", + "type": "string" + }, + "message_filename": { + "$ref": "#/$defs/MessageFileName" + }, + "proposed_file_changes": { + "items": { + "$ref": "#/$defs/ProposedFileChangeV1" + }, + "title": "Proposed File Changes", + "type": "array" + }, + "platform_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Platform Id" + } + }, + "required": [ + "sender_email", + "proposed_file_changes" + ], + "title": "ProposedFileChangesMessageV1", + "type": "object" + }, + "FileChangeEventsMessage": { + "$defs": { + "FileChangeEventV1": { + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "path_in_datasite": { + "format": "path", + "title": "Path In Datasite", + "type": "string" + }, + "datasite_email": { + "title": "Datasite Email", + "type": "string" + }, + "content": { + "anyOf": [ + { + "type": "string" + }, + { + "format": "binary", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Content" + }, + "old_hash": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Old Hash" + }, + "new_hash": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "New Hash" + }, + "is_deleted": { + "default": false, + "title": "Is Deleted", + "type": "boolean" + }, + "submitted_timestamp": { + "title": "Submitted Timestamp", + "type": "number" + }, + "timestamp": { + "title": "Timestamp", + "type": "number" + } + }, + "required": [ + "id", + "path_in_datasite", + "datasite_email", + "submitted_timestamp", + "timestamp" + ], + "title": "FileChangeEventV1", + "type": "object" + }, + "FileChangeEventsMessageFileName": { + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "timestamp": { + "title": "Timestamp", + "type": "number" + }, + "extension": { + "default": ".tar.gz", + "title": "Extension", + "type": "string" + } + }, + "title": "FileChangeEventsMessageFileName", + "type": "object" + } + }, + "description": "The events wire envelope (DO -> watchers). The envelope is the migratable\nunit; its items are pinned to the exact version class, never a floating alias.", + "properties": { + "canonical_name": { + "default": "FileChangeEventsMessage", + "title": "Canonical Name", + "type": "string" + }, + "version": { + "default": "1", + "title": "Version", + "type": "string" + }, + "events": { + "items": { + "$ref": "#/$defs/FileChangeEventV1" + }, + "title": "Events", + "type": "array" + }, + "message_filepath": { + "$ref": "#/$defs/FileChangeEventsMessageFileName" + } + }, + "required": [ + "events" + ], + "title": "FileChangeEventsMessageV1", + "type": "object" + } + } + } +} \ No newline at end of file diff --git a/syft_client/migrations/registry.py b/syft_client/migrations/registry.py new file mode 100644 index 00000000000..c071e4c6da8 --- /dev/null +++ b/syft_client/migrations/registry.py @@ -0,0 +1,38 @@ +from syft_migration import MigrationRegistry, MigrationService + +from syft_client.version import SYFT_CLIENT_VERSION + +PACKAGE_NAME = "syft-client" + +# Hardcoded, language-agnostic identifier for the syft-client protocol; +# intentionally distinct from the package name. +PROTOCOL_NAME = "syft-client" + +# Incrementing version of the syft-client protocol. Protocol 0 is the last +# release without per-object versioning (<= 0.1.117, files carry no +# canonical_name/version identity fields); protocol >= 1 serializes identity +# fields on every versioned object. +SYFT_CLIENT_PROTOCOL_VERSION = "1" + +# Package-local registry for all versioned syft-client objects. The current +# protocol schema is computed from the objects registered into it. +client_registry = MigrationRegistry( + protocol_name=PROTOCOL_NAME, + package_name=PACKAGE_NAME, + package_version=SYFT_CLIENT_VERSION, + protocol_version=SYFT_CLIENT_PROTOCOL_VERSION, +) + +# Shared service for loading/migrating syft-client objects. +client_migration_service = MigrationService(registry=client_registry) + + +def load_as_latest(data: dict, canonical_name: str) -> object: + """Load ``data`` (defaulting identity fields for protocol-0 files, which + predate them and are all version 1) and migrate to the latest version.""" + data.setdefault("canonical_name", canonical_name) + data.setdefault("version", "1") + obj = client_migration_service.load(data) + return client_migration_service.migrate( + obj, client_registry.latest_version(canonical_name) + ) diff --git a/syft_client/sync/events/file_change_event.py b/syft_client/sync/events/file_change_event.py index 3c5d3092589..f5389c09dcc 100644 --- a/syft_client/sync/events/file_change_event.py +++ b/syft_client/sync/events/file_change_event.py @@ -2,6 +2,7 @@ from pathlib import Path from uuid import UUID, uuid4 import base64 +import json from pydantic import ( BaseModel, Field, @@ -9,6 +10,9 @@ field_serializer, computed_field, ) +from syft_migration import MigratableObject + +from syft_client.migrations import client_registry, load_as_latest from syft_client.sync.messages.proposed_filechange import ProposedFileChange from syft_client.sync.utils.syftbox_utils import create_event_timestamp from syft_client.sync.utils.syftbox_utils import compress_data @@ -53,7 +57,7 @@ def from_string(cls, filename: str) -> "FileChangeEventsMessageFileName": raise ValueError(f"Invalid filename: {filename}") from e -class FileChangeEvent(BaseModel): +class FileChangeEventV1(BaseModel): id: UUID path_in_datasite: Path datasite_email: str @@ -130,13 +134,19 @@ def __hash__(self) -> int: return hash(self.id) def __eq__(self, other: Any) -> bool: - if not isinstance(other, FileChangeEvent): + if not isinstance(other, FileChangeEventV1): return False return self.id == other.id -class FileChangeEventsMessage(BaseModel): - events: List[FileChangeEvent] +class FileChangeEventsMessageV1(MigratableObject, registry=client_registry): + """The events wire envelope (DO -> watchers). The envelope is the migratable + unit; its items are pinned to the exact version class, never a floating alias.""" + + canonical_name: str = "FileChangeEventsMessage" + version: str = "1" + + events: List[FileChangeEventV1] message_filepath: FileChangeEventsMessageFileName = Field( default_factory=lambda: FileChangeEventsMessageFileName() ) @@ -149,6 +159,16 @@ def as_compressed_data(self) -> bytes: return compress_data(self.model_dump_json().encode("utf-8")) @classmethod - def from_compressed_data(cls, data: bytes) -> "FileChangeEvent": + def from_compressed_data(cls, data: bytes) -> "FileChangeEventsMessage": + """Decompress and load, upgraded to the latest version. + + Blobs written by protocol-0 clients (<= 0.1.117) predate the identity + fields; they are all version 1. + """ uncompressed_data = uncompress_data(data) - return cls.model_validate_json(uncompressed_data) + return load_as_latest(json.loads(uncompressed_data), "FileChangeEventsMessage") + + +# Current-version aliases: callers always work with the latest versions. +FileChangeEvent = FileChangeEventV1 +FileChangeEventsMessage = FileChangeEventsMessageV1 diff --git a/syft_client/sync/messages/proposed_filechange.py b/syft_client/sync/messages/proposed_filechange.py index 136af0831f4..5c6244e6e63 100644 --- a/syft_client/sync/messages/proposed_filechange.py +++ b/syft_client/sync/messages/proposed_filechange.py @@ -1,11 +1,15 @@ from typing import List, Any, Literal from uuid import UUID, uuid4 from pathlib import Path +import json import uuid import time import base64 from pydantic import Field, model_validator, field_serializer, computed_field from pydantic.main import BaseModel +from syft_migration import MigratableObject + +from syft_client.migrations import client_registry, load_as_latest from syft_client.sync.utils.syftbox_utils import compress_data, uncompress_data from syft_client.sync.utils.syftbox_utils import create_event_timestamp from syft_client.sync.utils.syftbox_utils import get_event_hash_from_content @@ -15,7 +19,7 @@ MESSAGE_FILENAME_EXTENSION = ".tar.gz" -class ProposedFileChange(BaseModel): +class ProposedFileChangeV1(BaseModel): id: UUID = Field(default_factory=lambda: uuid4()) old_hash: str | None = None new_hash: str | None = None # None for deletions @@ -94,20 +98,38 @@ def from_string(cls, filename: str) -> "MessageFileName": return cls(submitted_timestamp=submitted_timestamp, uid=uid) -class ProposedFileChangesMessage(BaseModel): +class ProposedFileChangesMessageV1(MigratableObject, registry=client_registry): + """The msgv2 wire envelope (DS -> DO). The envelope is the migratable unit; + its items are pinned to the exact version class, never a floating alias.""" + + canonical_name: str = "ProposedFileChangesMessage" + version: str = "1" + id: UUID = Field(default_factory=lambda: uuid4()) sender_email: str message_filename: MessageFileName = Field(default_factory=lambda: MessageFileName()) - proposed_file_changes: List[ProposedFileChange] + proposed_file_changes: List[ProposedFileChangeV1] # Platform-specific ID (e.g., Google Drive file ID) - set when retrieving message # Used to avoid re-querying the platform when removing the message platform_id: str | None = Field(default=None, exclude=True) @classmethod def from_compressed_data(cls, data: bytes) -> "ProposedFileChangesMessage": + """Decompress and load, upgraded to the latest version. + + Blobs written by protocol-0 clients (<= 0.1.117) predate the identity + fields; they are all version 1. + """ uncompressed_data = uncompress_data(data) - return cls.model_validate_json(uncompressed_data) + return load_as_latest( + json.loads(uncompressed_data), "ProposedFileChangesMessage" + ) def as_compressed_data(self) -> bytes: data = self.model_dump_json(indent=2).encode("utf-8") return compress_data(data) + + +# Current-version aliases: callers always work with the latest versions. +ProposedFileChange = ProposedFileChangeV1 +ProposedFileChangesMessage = ProposedFileChangesMessageV1 diff --git a/syft_client/sync/syftbox_manager.py b/syft_client/sync/syftbox_manager.py index 4e9837d8a79..923f1a49691 100644 --- a/syft_client/sync/syftbox_manager.py +++ b/syft_client/sync/syftbox_manager.py @@ -511,8 +511,20 @@ def from_config(cls, config: SyftboxManagerConfig): datasite_watcher_syncer = None job_runner = None - dataset_manager = SyftDatasetManager.from_config(config.dataset_manager_config) - job_client = JobClient.from_config(config.job_client_config) + # Created before the job and dataset clients so its live peer-schema + # maps (filled as peer version files load) can drive their protocol + # negotiation. + peer_manager = PeerManager.from_config( + config.peer_manager_config, email=config.email + ) + dataset_manager = SyftDatasetManager.from_config( + config.dataset_manager_config, + peer_schemas=peer_manager.live_peer_schemas("syft-dataset"), + ) + job_client = JobClient.from_config( + config.job_client_config, + peer_schemas=peer_manager.live_peer_schemas("syft-job"), + ) if config.has_do_role: datasite_owner_syncer = DatasiteOwnerSyncer.from_config( @@ -527,10 +539,6 @@ def from_config(cls, config: SyftboxManagerConfig): config.datasite_watcher_syncer_config ) - peer_manager = PeerManager.from_config( - config.peer_manager_config, email=config.email - ) - manager_res = cls( syftbox_folder=config.syftbox_folder, email=config.email, diff --git a/syft_client/sync/version/peer_manager.py b/syft_client/sync/version/peer_manager.py index cc4fb6a4b58..fa635cb3e96 100644 --- a/syft_client/sync/version/peer_manager.py +++ b/syft_client/sync/version/peer_manager.py @@ -11,6 +11,7 @@ from typing import Dict, List, Optional from pydantic import BaseModel, ConfigDict, PrivateAttr, model_validator +from syft_migration import ProtocolSchema from syft_client.sync.connections.base_connection import ConnectionConfig from syft_client.sync.connections.connection_router import ConnectionRouter @@ -148,6 +149,17 @@ class PeerManager(BaseModel): _own_version: Optional[VersionInfo] = PrivateAttr(default=None) _executor: Optional[ThreadPoolExecutor] = PrivateAttr(default=None) + # protocol name -> {peer email -> ProtocolSchema}. Inner dicts are handed + # out by live_peer_schemas() and mutated in place as peer versions load, + # so consumers (JobStorage, DatasetStorage) always see current knowledge. + _peer_protocol_schemas: Dict[str, Dict[str, ProtocolSchema]] = PrivateAttr( + default_factory=dict + ) + # peer email -> last loaded VersionInfo (or None); lets a protocol map + # registered AFTER versions were loaded backfill instead of starting empty. + _loaded_peer_versions: Dict[str, Optional[VersionInfo]] = PrivateAttr( + default_factory=dict + ) # ========== Peer List Properties ========== @@ -213,6 +225,53 @@ def model_post_init(self, __context) -> None: """Initialize the thread pool executor.""" self._executor = ThreadPoolExecutor(max_workers=self.n_threads) + def live_peer_schemas(self, protocol_name: str) -> Dict[str, ProtocolSchema]: + """The live {peer email -> ProtocolSchema} map for ``protocol_name``. + + The returned dict is owned by this PeerManager and updated in place + whenever a peer's version file is (re)loaded: peers advertising the + protocol appear, peers that stop advertising it (or whose version is + cleared) disappear. Hand it to JobStorage/DatasetStorage as + ``peer_schemas`` so negotiation always uses current knowledge. A map + registered after peer versions were already loaded is backfilled from + them. + """ + per_peer = self._peer_protocol_schemas.get(protocol_name) + if per_peer is None: + per_peer = self._peer_protocol_schemas[protocol_name] = {} + for email, version_info in list(self._loaded_peer_versions.items()): + self._sync_one(per_peer, protocol_name, email, version_info) + return per_peer + + @staticmethod + def _sync_one( + per_peer: Dict[str, ProtocolSchema], + protocol_name: str, + peer_email: str, + version_info: Optional[VersionInfo], + ) -> None: + # getattr: a peer's version may be a V1 object (no schemas attribute). + advertised = getattr(version_info, "protocol_schemas", None) or {} + schema = advertised.get(protocol_name) + if schema is not None: + per_peer[peer_email] = schema + else: + per_peer.pop(peer_email, None) + + def _update_peer_schemas( + self, peer_email: str, version_info: Optional[VersionInfo] + ) -> None: + """Sync the live schema maps with a freshly loaded peer version. + + A None version or a pre-V2 version file (empty ``protocol_schemas``) + removes the peer: it is an unknown speaker, and consumers apply their + own unknown-peer defaults. + """ + self._loaded_peer_versions[peer_email] = version_info + # list(): guard against a concurrent live_peer_schemas() registration. + for protocol_name, per_peer in list(self._peer_protocol_schemas.items()): + self._sync_one(per_peer, protocol_name, peer_email, version_info) + def get_own_version(self) -> VersionInfo: """Get current client's version info.""" if self._own_version is None: @@ -241,6 +300,7 @@ def load_peer_version(self, peer_email: str) -> Optional[VersionInfo]: cached_peer = self.get_cached_peer(peer_email) if cached_peer: cached_peer.version = version_info + self._update_peer_schemas(peer_email, version_info) return version_info def _load_single_peer_version( @@ -275,6 +335,7 @@ def load_peer_versions_parallel( peer = self.get_cached_peer(email) if peer: peer.version = version + self._update_peer_schemas(email, version) return {email: version for email, version in results} @@ -288,6 +349,7 @@ def clear_peer_version(self, peer_email: str) -> None: peer = self.get_cached_peer(peer_email) if peer: peer.version = None + self._update_peer_schemas(peer_email, None) def peer_compatibility_status(self, peer_email: str) -> CompatibilityStatus: """Get the CompatibilityStatus for a peer (UNKNOWN if version not loaded).""" @@ -490,6 +552,7 @@ def add_peer(self, peer_email: str, force: bool = False, verbose: bool = True): ) self.share_version_with_peer(peer_email) version_info = self.connection_router.read_peer_version_file(peer_email) + self._update_peer_schemas(peer_email, version_info) new_peer_obj.version = version_info new_peer_obj.public_encryption_bundle = peer_bundle diff --git a/syft_client/sync/version/version_info.py b/syft_client/sync/version/version_info.py index aac9d9aed4a..28f1453c67d 100644 --- a/syft_client/sync/version/version_info.py +++ b/syft_client/sync/version/version_info.py @@ -4,13 +4,16 @@ from __future__ import annotations +import json import logging from datetime import datetime, timezone from enum import Enum from typing import Optional -from pydantic import BaseModel, Field +from pydantic import Field +from syft_migration import MigratableObject, ProtocolSchema +from syft_client.migrations import client_registry, load_as_latest from syft_client.version import ( MIN_SUPPORTED_PROTOCOL_VERSION, MIN_SUPPORTED_SYFT_CLIENT_VERSION, @@ -36,8 +39,17 @@ def _parse_semver(version_str: str) -> tuple[int, int, int]: return (int(parts[0]), int(parts[1]), int(parts[2])) -class VersionInfo(BaseModel): - """Model representing version information for a syft client.""" +class VersionInfoV1(MigratableObject, registry=client_registry): + """Model representing version information for a syft client. + + Stored as SYFT_version.json in the peer-visible SyftBox folder. This file + is the bootstrap channel for protocol negotiation (peers read it to learn + what we speak), so its schema may only ever change additively: every + supported client version must be able to parse every newer version file. + """ + + canonical_name: str = "VersionInfo" + version: str = "1" syft_client_version: str min_supported_syft_client_version: str @@ -126,5 +138,91 @@ def to_json(self) -> str: @classmethod def from_json(cls, json_str: str) -> "VersionInfo": - """Deserialize from JSON string.""" - return cls.model_validate_json(json_str) + """Deserialize from JSON string, upgraded to the latest version. + + Files written by protocol-0 clients (<= 0.1.117) predate the identity + fields; they are all version 1. + """ + return load_as_latest(json.loads(json_str), "VersionInfo") + + +def _slim_schema_of(registry) -> ProtocolSchema: + """The registry's protocol schema without the embedded object JSON schemas. + + Negotiation needs only ``version`` and ``supported_versions``; skipping + ``current_object_schemas`` keeps the published version file small (the + full frozen schemas live in the release artifacts, not on the wire) and + avoids computing every object's JSON schema just to discard it. + """ + return ProtocolSchema( + protocol_name=registry.protocol_name, + version=registry.protocol_version, + supported_versions={ + canonical_name: sorted(versions) + for canonical_name, versions in registry.objects.items() + }, + ) + + +def _gather_protocol_schemas() -> dict[str, ProtocolSchema]: + """Slim protocol schemas of every syft package present in this install. + + Keyed by protocol name. syft-job/syft-dataset are optional dependencies; + a missing or broken package simply means its schema is not advertised and + peers treat this client as an unknown speaker of that protocol (same + failure-tolerant pattern as the install-source detection in ``current``). + """ + logger = logging.getLogger(__name__) + schemas = {client_registry.protocol_name: _slim_schema_of(client_registry)} + try: + from syft_job.migrations import job_registry + + schemas[job_registry.protocol_name] = _slim_schema_of(job_registry) + except Exception as e: + logger.debug(f"Not advertising a syft-job protocol schema: {e}") + try: + from syft_datasets.migrations.registry import dataset_registry + + schemas[dataset_registry.protocol_name] = _slim_schema_of(dataset_registry) + except Exception as e: + logger.debug(f"Not advertising a syft-dataset protocol schema: {e}") + return schemas + + +class VersionInfoV2(VersionInfoV1): + """V2 adds the protocol schemas this client speaks (client, job, dataset). + + Purely additive over V1 (see the bootstrap-channel rule in the V1 + docstring): protocol-0/1 readers ignore the extra key. + """ + + version: str = "2" + + # protocol name -> slim ProtocolSchema (no embedded object JSON schemas). + protocol_schemas: dict[str, ProtocolSchema] = Field(default_factory=dict) + + @classmethod + def current(cls) -> "VersionInfo": + info = super().current() + info.protocol_schemas = _gather_protocol_schemas() + return info + + +@client_registry.migration("VersionInfo", "1", "2") +def _version_info_v1_to_v2(obj: VersionInfoV1) -> VersionInfoV2: + # A v1 file says nothing about package protocols: empty schemas, meaning + # "unknown speaker" to consumers. + return VersionInfoV2.model_validate( + obj.model_dump(exclude={"canonical_name", "version"}) + ) + + +@client_registry.migration("VersionInfo", "2", "1") +def _version_info_v2_to_v1(obj: VersionInfoV2) -> VersionInfoV1: + return VersionInfoV1.model_validate( + obj.model_dump(exclude={"canonical_name", "version", "protocol_schemas"}) + ) + + +# Current-version alias: callers always work with the latest VersionInfo. +VersionInfo = VersionInfoV2 diff --git a/tests/migrations/__init__.py b/tests/migrations/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/migrations/p2p/__init__.py b/tests/migrations/p2p/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/migrations/p2p/fixtures/syft_client-0.1.117-protocol0/SYFT_version.json b/tests/migrations/p2p/fixtures/syft_client-0.1.117-protocol0/SYFT_version.json new file mode 100644 index 00000000000..05bcb23b59b --- /dev/null +++ b/tests/migrations/p2p/fixtures/syft_client-0.1.117-protocol0/SYFT_version.json @@ -0,0 +1,9 @@ +{ + "syft_client_version": "0.1.117", + "min_supported_syft_client_version": "0.1.93", + "protocol_version": "1.0.0", + "min_supported_protocol_version": "1.0.0", + "syft_client_install_source": "pip", + "updated_at": "2026-07-20T10:15:30.123456Z", + "attestation_token": null +} \ No newline at end of file diff --git a/tests/migrations/p2p/fixtures/syft_client-0.1.117-protocol0/msgv2_1752900000.0_6f9d5f57-31f7-4302-8746-9ba030e88961.tar.gz b/tests/migrations/p2p/fixtures/syft_client-0.1.117-protocol0/msgv2_1752900000.0_6f9d5f57-31f7-4302-8746-9ba030e88961.tar.gz new file mode 100644 index 00000000000..834a33e2fc5 Binary files /dev/null and b/tests/migrations/p2p/fixtures/syft_client-0.1.117-protocol0/msgv2_1752900000.0_6f9d5f57-31f7-4302-8746-9ba030e88961.tar.gz differ diff --git a/tests/migrations/p2p/fixtures/syft_client-0.1.117-protocol0/syfteventsmessagev3_1752900001.0_b3d3e5c1-89a4-4f7e-9f26-4a3f0e2d1c0a.tar.gz b/tests/migrations/p2p/fixtures/syft_client-0.1.117-protocol0/syfteventsmessagev3_1752900001.0_b3d3e5c1-89a4-4f7e-9f26-4a3f0e2d1c0a.tar.gz new file mode 100644 index 00000000000..092ef6584dd Binary files /dev/null and b/tests/migrations/p2p/fixtures/syft_client-0.1.117-protocol0/syfteventsmessagev3_1752900001.0_b3d3e5c1-89a4-4f7e-9f26-4a3f0e2d1c0a.tar.gz differ diff --git a/tests/migrations/p2p/test_dataset_schema_negotiation.py b/tests/migrations/p2p/test_dataset_schema_negotiation.py new file mode 100644 index 00000000000..12c3e461608 --- /dev/null +++ b/tests/migrations/p2p/test_dataset_schema_negotiation.py @@ -0,0 +1,82 @@ +"""Peer-advertised dataset schemas drive DatasetStorage protocol negotiation.""" + +from pathlib import Path + +from syft_datasets.config import SyftBoxConfig +from syft_datasets.dataset_manager import SyftDatasetManager +from syft_datasets.dataset_storage import DatasetStorage +from syft_datasets.migrations.registry import DATASET_PROTOCOL_VERSION +from syft_migration import ProtocolSchema + +OWNER_EMAIL = "do@test.org" +OLD_PEER = "old@test.org" +NEW_PEER = "new@test.org" +UNKNOWN_PEER = "unknown@test.org" + + +def _dataset_schema(protocol_version: str) -> ProtocolSchema: + # The slim form a peer advertises in its VersionInfo. + return ProtocolSchema( + protocol_name="syft-dataset", + version=protocol_version, + supported_versions={"Dataset": ["1"], "PrivateDatasetConfig": ["1"]}, + ) + + +def _storage(tmp_path: Path, peer_schemas: dict) -> DatasetStorage: + config = SyftBoxConfig(syftbox_folder=tmp_path / "SyftBox", email=OWNER_EMAIL) + (tmp_path / "SyftBox" / OWNER_EMAIL).mkdir(parents=True, exist_ok=True) + return DatasetStorage(config=config, peer_schemas=peer_schemas) + + +def test_mixed_audience_writes_both_versions(tmp_path): + storage = _storage( + tmp_path, + { + OLD_PEER: _dataset_schema("0"), + NEW_PEER: _dataset_schema(DATASET_PROTOCOL_VERSION), + }, + ) + versions = storage.target_protocol_versions_for_peers([OLD_PEER, NEW_PEER]) + assert versions == {"0", DATASET_PROTOCOL_VERSION} + + +def test_all_current_audience_drops_legacy_layout(tmp_path): + storage = _storage(tmp_path, {NEW_PEER: _dataset_schema(DATASET_PROTOCOL_VERSION)}) + assert storage.target_protocol_versions_for_peers([NEW_PEER]) == { + DATASET_PROTOCOL_VERSION + } + + +def test_unknown_peer_gets_widest_protocol(tmp_path): + storage = _storage(tmp_path, {}) + versions = storage.target_protocol_versions_for_peers([UNKNOWN_PEER]) + assert versions == {storage._widest_protocol_version} + + +def test_live_map_updates_are_seen_by_storage(tmp_path): + live: dict = {} + storage = _storage(tmp_path, live) + assert storage.target_protocol_versions_for_peers([NEW_PEER]) == { + storage._widest_protocol_version + } + live[NEW_PEER] = _dataset_schema(DATASET_PROTOCOL_VERSION) + assert storage.target_protocol_versions_for_peers([NEW_PEER]) == { + DATASET_PROTOCOL_VERSION + } + + +def test_manager_from_config_passes_schemas_through(tmp_path): + config = SyftBoxConfig(syftbox_folder=tmp_path / "SyftBox", email=OWNER_EMAIL) + (tmp_path / "SyftBox" / OWNER_EMAIL).mkdir(parents=True, exist_ok=True) + live = {OLD_PEER: _dataset_schema("0")} + manager = SyftDatasetManager.from_config(config, peer_schemas=live) + assert manager.storage.peer_schemas is live + + +def test_newer_peer_clamps_to_our_protocol(tmp_path): + # A peer speaking a future protocol contributes min(ours, theirs) = ours. + storage = _storage(tmp_path, {NEW_PEER: _dataset_schema("99")}) + assert storage.target_protocol_versions_for_peers([NEW_PEER]) == { + DATASET_PROTOCOL_VERSION + } diff --git a/tests/migrations/p2p/test_job_schema_negotiation.py b/tests/migrations/p2p/test_job_schema_negotiation.py new file mode 100644 index 00000000000..f1051e5f8c7 --- /dev/null +++ b/tests/migrations/p2p/test_job_schema_negotiation.py @@ -0,0 +1,105 @@ +"""Peer-advertised job schemas drive JobStorage protocol negotiation.""" + +from pathlib import Path + +from syft_job import SyftJobConfig +from syft_job.client import JobClient +from syft_job.job_storage import JobStorage +from syft_job.migrations.registry import JOB_PROTOCOL_VERSION +from syft_migration import ProtocolSchema + +DO_EMAIL = "do@test.org" +DS_EMAIL = "ds@test.org" + + +def _job_schema(protocol_version: str) -> ProtocolSchema: + # The slim form a peer advertises in its VersionInfo (see + # version_info._slim_schema_of): no embedded object schemas. + return ProtocolSchema( + protocol_name="syft-job", + version=protocol_version, + supported_versions={"JobState": ["1"], "JobSubmissionMetadata": ["1"]}, + ) + + +def _storage(tmp_path: Path, peer_schemas: dict) -> JobStorage: + config = SyftJobConfig( + syftbox_folder=tmp_path / "SyftBox", current_user_email=DS_EMAIL + ) + (tmp_path / "SyftBox" / DS_EMAIL).mkdir(parents=True, exist_ok=True) + return JobStorage(config=config, peer_schemas=peer_schemas) + + +def test_protocol0_peer_negotiates_down(tmp_path): + storage = _storage(tmp_path, {DO_EMAIL: _job_schema("0")}) + assert storage.negotiated_protocol_version_for_peer(DO_EMAIL) == "0" + ref = storage.new_submission_ref(DO_EMAIL, "legacy.job") + assert ref.protocol_version == "0" + # Protocol 0 = flat pre-versioning layout: no v path segment. + assert f"v{JOB_PROTOCOL_VERSION}" not in storage.submission_dir(ref).parts + + +def test_current_peer_negotiates_current(tmp_path): + storage = _storage(tmp_path, {DO_EMAIL: _job_schema(JOB_PROTOCOL_VERSION)}) + assert ( + storage.negotiated_protocol_version_for_peer(DO_EMAIL) == JOB_PROTOCOL_VERSION + ) + ref = storage.new_submission_ref(DO_EMAIL, "current.job") + assert ref.protocol_version == JOB_PROTOCOL_VERSION + assert f"v{JOB_PROTOCOL_VERSION}" in storage.submission_dir(ref).parts + + +def test_unknown_peer_keeps_current_protocol_assumption(tmp_path): + storage = _storage(tmp_path, {}) + ref = storage.new_submission_ref(DO_EMAIL, "unknown.job") + assert ref.protocol_version == JOB_PROTOCOL_VERSION + + +def test_live_map_updates_are_seen_by_storage(tmp_path): + # JobStorage holds the dict by reference: schemas arriving after + # construction (peer version files loading) change negotiation. + live: dict = {} + storage = _storage(tmp_path, live) + assert ( + storage.new_submission_ref(DO_EMAIL, "before.job").protocol_version + == JOB_PROTOCOL_VERSION + ) + live[DO_EMAIL] = _job_schema("0") + assert storage.new_submission_ref(DO_EMAIL, "after.job").protocol_version == "0" + + +def test_job_client_from_config_passes_schemas_through(tmp_path): + config = SyftJobConfig( + syftbox_folder=tmp_path / "SyftBox", current_user_email=DS_EMAIL + ) + (tmp_path / "SyftBox" / DS_EMAIL).mkdir(parents=True, exist_ok=True) + live = {DO_EMAIL: _job_schema("0")} + client = JobClient.from_config(config, peer_schemas=live) + assert client.manager.peer_schemas is live + + +def test_newer_peer_clamps_to_our_protocol(tmp_path): + # A peer speaking a future protocol negotiates down to ours (min). + storage = _storage(tmp_path, {DO_EMAIL: _job_schema("99")}) + assert ( + storage.negotiated_protocol_version_for_peer(DO_EMAIL) == JOB_PROTOCOL_VERSION + ) + + +def test_downgrade_write_uses_slim_peer_schema(tmp_path): + # The write path's downgrade target comes from the slim advertised schema + # (supported_versions only) — no dependency on current_object_schemas. + from syft_job.models import JobSubmissionMetadataV1 + from datetime import datetime, timezone + + storage = _storage(tmp_path, {DO_EMAIL: _job_schema("0")}) + ref = storage.new_submission_ref(DO_EMAIL, "legacy.job") + metadata = JobSubmissionMetadataV1( + name="legacy.job", + submitted_by=DS_EMAIL, + datasite_email=DO_EMAIL, + submitted_at=datetime.now(tz=timezone.utc), + ) + path = storage.write_submission(ref, metadata) + assert path.exists() + assert storage.read_submission(ref).name == "legacy.job" diff --git a/tests/migrations/p2p/test_older_protocol_compatibility.py b/tests/migrations/p2p/test_older_protocol_compatibility.py new file mode 100644 index 00000000000..bb8922fbc35 --- /dev/null +++ b/tests/migrations/p2p/test_older_protocol_compatibility.py @@ -0,0 +1,121 @@ +"""Reading every released syft-client serialized format with the current code. + +Each fixture directory under fixtures/ is named ``syft_client--protocol

`` +and holds the serialized artifacts exactly as that release produced them: the +published SYFT_version.json, one msgv2 proposed-changes blob and one events blob. +The current code must still read, upgrade and round-trip all of them. + +Protocol 0 is the last release (0.1.117) without canonical_name/version identity +fields; protocol >= 1 writes them. Fixtures are generated by +scripts/generate_release_fixture.py (protocol 0 / 0.1.117 predates it and is +hand-authored). +""" + +import json +import re +from pathlib import Path + +import pytest + +from syft_client.migrations import client_registry +from syft_client.sync.events.file_change_event import FileChangeEventsMessage +from syft_client.sync.messages.proposed_filechange import ProposedFileChangesMessage +from syft_client.sync.utils.syftbox_utils import uncompress_data +from syft_client.sync.version.version_info import VersionInfo + +FIXTURES_DIR = Path(__file__).parent / "fixtures" +RELEASE_FIXTURES = sorted(FIXTURES_DIR.glob("syft_client-*-protocol*")) + +released_fixtures = pytest.mark.parametrize( + "fixture", RELEASE_FIXTURES, ids=lambda f: f.name +) + + +def _protocol_of(fixture: Path) -> str: + return re.search(r"-protocol(\d+)$", fixture.name).group(1) + + +def _has_identity(protocol: str) -> bool: + """canonical_name/version are written only from protocol 1 onwards.""" + return protocol != "0" + + +def _single(fixture: Path, pattern: str) -> Path: + matches = list(fixture.glob(pattern)) + assert len(matches) == 1, f"expected one {pattern} in {fixture.name}" + return matches[0] + + +def test_fixtures_exist(): + assert RELEASE_FIXTURES, "no release fixtures found" + + +@released_fixtures +def test_version_file_reads_and_round_trips(fixture: Path): + raw = _single(fixture, "SYFT_version.json").read_text() + assert ("canonical_name" in json.loads(raw)) == _has_identity(_protocol_of(fixture)) + + info = VersionInfo.from_json(raw) + assert info.version == client_registry.latest_version("VersionInfo") + assert info.syft_client_version + # Round-trip: what the current code writes must load again. + assert VersionInfo.from_json(info.to_json()) == info + + +@released_fixtures +def test_proposed_message_reads_and_round_trips(fixture: Path): + blob = _single(fixture, "msgv2_*.tar.gz").read_bytes() + assert ("canonical_name" in json.loads(uncompress_data(blob))) == _has_identity( + _protocol_of(fixture) + ) + + message = ProposedFileChangesMessage.from_compressed_data(blob) + assert message.version == client_registry.latest_version( + "ProposedFileChangesMessage" + ) + changes = {c.path_in_datasite.name: c for c in message.proposed_file_changes} + assert changes["notes.txt"].content == "hello from the release fixture" + assert changes["blob.bin"].content == b"\x00\x01\x02fixture-binary" + deletion = changes["removed.txt"] + assert deletion.is_deleted and deletion.content is None + assert deletion.new_hash is None and deletion.content_type is None + + # Upgrade-on-write: what the current code re-emits carries identity fields. + rewritten = message.as_compressed_data() + assert b'"canonical_name"' in uncompress_data(rewritten) + restored = ProposedFileChangesMessage.from_compressed_data(rewritten) + assert restored.proposed_file_changes == message.proposed_file_changes + + +@released_fixtures +def test_events_message_reads_and_round_trips(fixture: Path): + blob = _single(fixture, "syfteventsmessagev3_*.tar.gz").read_bytes() + assert ("canonical_name" in json.loads(uncompress_data(blob))) == _has_identity( + _protocol_of(fixture) + ) + + message = FileChangeEventsMessage.from_compressed_data(blob) + assert message.version == client_registry.latest_version("FileChangeEventsMessage") + events = {e.path_in_datasite.name: e for e in message.events} + assert events["notes.txt"].content == "hello from the release fixture" + assert events["blob.bin"].content == b"\x00\x01\x02fixture-binary" + assert events["removed.txt"].is_deleted and events["removed.txt"].content is None + + # Upgrade-on-write: what the current code re-emits carries identity fields. + rewritten = message.as_compressed_data() + assert b'"canonical_name"' in uncompress_data(rewritten) + restored = FileChangeEventsMessage.from_compressed_data(rewritten) + assert restored.events == message.events + + +@released_fixtures +def test_fixture_protocol_is_in_registry_history(fixture: Path): + """Every fixture's protocol is either the current one or a released one + the registry knows the schema for (so downgrades can target it).""" + protocol = _protocol_of(fixture) + schema = client_registry.schema_for_protocol_version(protocol) + assert set(schema.supported_versions) == { + "VersionInfo", + "ProposedFileChangesMessage", + "FileChangeEventsMessage", + } diff --git a/tests/migrations/unit/__init__.py b/tests/migrations/unit/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/migrations/unit/fixtures/version_info/SYFT_version-0.1.117.json b/tests/migrations/unit/fixtures/version_info/SYFT_version-0.1.117.json new file mode 100644 index 00000000000..b6a40d2e192 --- /dev/null +++ b/tests/migrations/unit/fixtures/version_info/SYFT_version-0.1.117.json @@ -0,0 +1,9 @@ +{ + "syft_client_version": "0.1.117", + "min_supported_syft_client_version": "0.1.93", + "protocol_version": "1.0.0", + "min_supported_protocol_version": "1.0.0", + "syft_client_install_source": "pip", + "updated_at": "2026-07-20T10:15:30.123456Z", + "attestation_token": null +} diff --git a/tests/migrations/unit/test_file_change_events_serialization.py b/tests/migrations/unit/test_file_change_events_serialization.py new file mode 100644 index 00000000000..6b87828b48d --- /dev/null +++ b/tests/migrations/unit/test_file_change_events_serialization.py @@ -0,0 +1,97 @@ +"""The events envelope round-trips and legacy protocol-0 blobs still decode.""" + +import base64 +import json +from uuid import uuid4 + +from syft_client.migrations import client_registry +from syft_client.sync.events.file_change_event import ( + FileChangeEvent, + FileChangeEventsMessage, + FileChangeEventsMessageV1, + FileChangeEventV1, +) +from syft_client.sync.messages.proposed_filechange import ProposedFileChange +from syft_client.sync.utils.syftbox_utils import compress_data + + +def _make_event(content) -> FileChangeEvent: + return FileChangeEvent( + id=uuid4(), + path_in_datasite="data/file.txt", + datasite_email="do@test.org", + content=content, + submitted_timestamp=1752900000.0, + timestamp=1752900001.0, + ) + + +def test_envelope_registered_items_not(): + assert client_registry.versions("FileChangeEventsMessage") + assert not client_registry.versions("FileChangeEvent") + assert FileChangeEventsMessage is FileChangeEventsMessageV1 + assert FileChangeEvent is FileChangeEventV1 + + +def test_round_trip_text_binary_and_deletion(): + for content in ["hello", b"\x00\x01binary", None]: + original = FileChangeEventsMessage(events=[_make_event(content)]) + restored = FileChangeEventsMessage.from_compressed_data( + original.as_compressed_data() + ) + assert restored.events[0].content == content + assert restored.events[0].content_type == original.events[0].content_type + assert restored.message_filepath == original.message_filepath + + +def test_identity_fields_on_the_wire(): + # load_as_latest would setdefault them back, so the round-trip alone + # cannot catch a silently broken serialization of the identity fields. + data = json.loads( + FileChangeEventsMessage(events=[_make_event("x")]).model_dump_json() + ) + assert data["canonical_name"] == "FileChangeEventsMessage" + assert data["version"] == "1" + + +def test_legacy_protocol0_blob_decodes_as_latest(): + # A blob exactly as a <= 0.1.117 client writes it: no identity fields. + legacy = { + "events": [ + { + "id": "9c1a2e75-8a45-4a17-b7f2-0d94d13d3c60", + "path_in_datasite": "data/blob.bin", + "datasite_email": "do@test.org", + "content": base64.b64encode(b"\x00\x01binary").decode("utf-8"), + "content_type": "binary", + "old_hash": None, + "new_hash": "abc123", + "is_deleted": False, + "submitted_timestamp": 1752900000.0, + "timestamp": 1752900001.0, + } + ], + "message_filepath": { + "id": "6f9d5f57-31f7-4302-8746-9ba030e88961", + "timestamp": 1752900001.0, + "extension": ".tar.gz", + }, + } + blob = compress_data(json.dumps(legacy).encode("utf-8")) + + message = FileChangeEventsMessage.from_compressed_data(blob) + assert message.version == client_registry.latest_version("FileChangeEventsMessage") + assert message.events[0].content == b"\x00\x01binary" + + +def test_from_proposed_filechange_carries_identity_free_items(): + proposed = ProposedFileChange( + path_in_datasite="data/file.txt", + content="hello", + datasite_email="do@test.org", + ) + event = FileChangeEvent.from_proposed_filechange(proposed) + assert event.id == proposed.id + assert event.new_hash == proposed.new_hash + # Items carry no identity fields on the wire; only envelopes do. + assert "canonical_name" not in json.loads(event.model_dump_json()) diff --git a/tests/migrations/unit/test_history_artifacts.py b/tests/migrations/unit/test_history_artifacts.py new file mode 100644 index 00000000000..6d9adcfeadb --- /dev/null +++ b/tests/migrations/unit/test_history_artifacts.py @@ -0,0 +1,108 @@ +"""Past release artifacts register cleanly and guard against schema drift.""" + +import pytest +import syft_client # noqa: F401 -- imports models and registers history +from syft_migration import MigrationError, MigrationRegistry, ReleasedProtocol + +from syft_client.migrations import client_registry +from syft_client.migrations.history import ( + PACKAGE_ARTIFACTS_DIR, + PROTOCOLS_DIR, + register_historic_schemas, +) + +PROTOCOL_0_PATH = PROTOCOLS_DIR / "protocol-0.json" + + +def test_protocol0_artifacts_registered(): + # importing syft_client registered the hardcoded 0.1.117 artifacts. + assert "0" in client_registry.protocol_version_history + package_info = client_registry.package_version_history["0"] + assert package_info.package_name == "syft-client" + assert package_info.version == "0.1.117" + + schema = client_registry.protocol_version_history["0"] + assert schema.supported_versions == { + "VersionInfo": ["1"], + "ProposedFileChangesMessage": ["1"], + "FileChangeEventsMessage": ["1"], + } + + +def test_registering_again_is_idempotent(): + register_historic_schemas() + assert "0" in client_registry.protocol_version_history + + +def test_all_protocol_artifacts_well_formed(): + # Filename encodes the frozen protocol version, and every supported + # canonical name freezes a current-object schema (catches a mis-named or + # hand-edited artifact). + paths = sorted(PROTOCOLS_DIR.glob("*.json")) + assert paths, "no released protocol artifacts found" + for path in paths: + schema = ReleasedProtocol.load(path).protocol_schema + assert path.name == f"protocol-{schema.version}.json" + assert set(schema.current_object_schemas) == set(schema.supported_versions) + + +def test_no_schema_drift_against_released_protocols(): + # Every schema frozen by a released protocol must still be produced + # byte-identically by the class registered for that version. + assert client_registry.find_schema_drift() == [], ( + "A released object schema drifted. Fix by either: (1) reverting the " + "model change; or (2) adding a new V model class, registering " + "migrations in both directions, and bumping " + "SYFT_CLIENT_PROTOCOL_VERSION in syft_client/migrations/registry.py. " + "If the drift comes from a pydantic upgrade changing JSON-schema " + "output only, regenerate the artifacts instead." + ) + + +def test_protocol_not_changed_without_bump(): + assert not client_registry.protocol_changed_without_bump() + + +def test_bump_guard_trips_on_protocol_change(): + # A registry claiming the same protocol version as a released schema but + # supporting different object versions must trip the guard. + stale = MigrationRegistry( + protocol_name=client_registry.protocol_name, + package_name=client_registry.package_name, + package_version=client_registry.package_version, + protocol_version="0", # pretend we still speak the released protocol 0 + ) + # Register only a subset of protocol-0's objects, then load its schema. + stale.register_object_version(client_registry.get_class("VersionInfo", "1")) + stale.register_historic_protocol_schema( + ReleasedProtocol.load(PROTOCOL_0_PATH).protocol_schema + ) + assert stale.protocol_changed_without_bump() + + +def test_unknown_object_version_in_artifact_raises(): + # The fail-at-import guarantee syft_client/__init__.py relies on: an + # artifact naming an object version this release cannot load must raise. + schema = ReleasedProtocol.load(PROTOCOL_0_PATH).protocol_schema + schema = schema.model_copy( + update={ + "supported_versions": { + **schema.supported_versions, + "VersionInfo": ["1", "99"], + } + } + ) + empty = MigrationRegistry( + protocol_name=client_registry.protocol_name, + package_name=client_registry.package_name, + package_version=client_registry.package_version, + protocol_version=client_registry.protocol_version, + ) + empty.register_object_version(client_registry.get_class("VersionInfo", "1")) + with pytest.raises(MigrationError): + empty.register_historic_protocol_schema(schema, raise_for_unknown_objects=True) + + +def test_artifact_files_exist(): + assert (PACKAGE_ARTIFACTS_DIR / "syft-client-0.1.117.json").exists() + assert PROTOCOL_0_PATH.exists() diff --git a/tests/migrations/unit/test_peer_manager_schemas.py b/tests/migrations/unit/test_peer_manager_schemas.py new file mode 100644 index 00000000000..86c41f8e6e3 --- /dev/null +++ b/tests/migrations/unit/test_peer_manager_schemas.py @@ -0,0 +1,64 @@ +"""PeerManager's live peer-schema maps track loaded peer versions.""" + +from syft_client.sync.version.peer_manager import PeerManager +from syft_client.sync.version.version_info import VersionInfo, VersionInfoV1 + + +def _peer_manager() -> PeerManager: + # Construct without model_validate side effects: only the private schema + # map and _update_peer_schemas are exercised here. + return PeerManager.model_construct() + + +def _v2_with_schemas() -> VersionInfo: + return VersionInfo.current() + + +def _v1() -> VersionInfoV1: + return VersionInfoV1( + syft_client_version="0.1.117", + min_supported_syft_client_version="0.1.93", + protocol_version="1.0.0", + min_supported_protocol_version="1.0.0", + ) + + +def test_advertising_peer_appears_in_live_map(): + pm = _peer_manager() + live = pm.live_peer_schemas("syft-job") + pm._update_peer_schemas("do@test.org", _v2_with_schemas()) + assert "do@test.org" in live + assert live["do@test.org"].protocol_name == "syft-job" + + +def test_pre_v2_peer_is_an_unknown_speaker(): + pm = _peer_manager() + live = pm.live_peer_schemas("syft-job") + pm._update_peer_schemas("do@test.org", _v2_with_schemas()) + # A reloaded version file from an old client (upgraded V1: no schemas) + # must remove the stale entry. + pm._update_peer_schemas("do@test.org", _v1()) + assert live == {} + + +def test_cleared_version_removes_peer(): + pm = _peer_manager() + live = pm.live_peer_schemas("syft-client") + pm._update_peer_schemas("do@test.org", _v2_with_schemas()) + pm._update_peer_schemas("do@test.org", None) + assert live == {} + + +def test_map_identity_is_stable(): + # live_peer_schemas must always return the same dict object so consumers + # holding a reference see updates. + pm = _peer_manager() + assert pm.live_peer_schemas("syft-job") is pm.live_peer_schemas("syft-job") + + +def test_late_registered_map_backfills_from_loaded_versions(): + pm = _peer_manager() + pm._update_peer_schemas("do@test.org", _v2_with_schemas()) + # Registering the protocol AFTER the version loaded must not start empty. + live = pm.live_peer_schemas("syft-job") + assert "do@test.org" in live diff --git a/tests/migrations/unit/test_proposed_filechange_serialization.py b/tests/migrations/unit/test_proposed_filechange_serialization.py new file mode 100644 index 00000000000..9e8f6ce62a5 --- /dev/null +++ b/tests/migrations/unit/test_proposed_filechange_serialization.py @@ -0,0 +1,91 @@ +"""The msgv2 envelope round-trips and legacy protocol-0 blobs still decode.""" + +import base64 +import json + +from syft_client.migrations import client_registry +from syft_client.sync.messages.proposed_filechange import ( + ProposedFileChange, + ProposedFileChangesMessage, + ProposedFileChangesMessageV1, + ProposedFileChangeV1, +) +from syft_client.sync.utils.syftbox_utils import compress_data + + +def _make_message(content) -> ProposedFileChangesMessage: + return ProposedFileChangesMessage( + sender_email="ds@test.org", + proposed_file_changes=[ + ProposedFileChange( + path_in_datasite="data/file.txt", + content=content, + datasite_email="do@test.org", + ) + ], + ) + + +def test_envelope_registered_items_not(): + # The envelope is the migratable unit; items ride inside it. + assert client_registry.versions("ProposedFileChangesMessage") + assert not client_registry.versions("ProposedFileChange") + assert ProposedFileChangesMessage is ProposedFileChangesMessageV1 + assert ProposedFileChange is ProposedFileChangeV1 + + +def test_round_trip_text_and_binary(): + for content in ["hello", b"\x00\x01binary"]: + original = _make_message(content) + restored = ProposedFileChangesMessage.from_compressed_data( + original.as_compressed_data() + ) + assert restored.sender_email == original.sender_email + assert restored.proposed_file_changes[0].content == content + assert ( + restored.proposed_file_changes[0].new_hash + == original.proposed_file_changes[0].new_hash + ) + + +def test_legacy_protocol0_blob_decodes_as_latest(): + # A blob exactly as a <= 0.1.117 client writes it: no identity fields + # on the envelope, base64 binary content on the item. + legacy = { + "id": "8be509b2-4340-44db-a3a4-b0ecf8c463f4", + "sender_email": "ds@test.org", + "message_filename": { + "submitted_timestamp": 1752900000.0, + "uid": "6f9d5f57-31f7-4302-8746-9ba030e88961", + }, + "proposed_file_changes": [ + { + "id": "9c1a2e75-8a45-4a17-b7f2-0d94d13d3c60", + "old_hash": None, + "submitted_timestamp": 1752900000.0, + "path_in_datasite": "data/blob.bin", + "content": base64.b64encode(b"\x00\x01binary").decode("utf-8"), + "content_type": "binary", + "datasite_email": "do@test.org", + "is_deleted": False, + } + ], + } + blob = compress_data(json.dumps(legacy).encode("utf-8")) + + message = ProposedFileChangesMessage.from_compressed_data(blob) + assert message.version == client_registry.latest_version( + "ProposedFileChangesMessage" + ) + change = message.proposed_file_changes[0] + assert change.content == b"\x00\x01binary" + assert change.content_type == "binary" + # pre_init derived the hash from the payload content only. + assert change.new_hash + + +def test_identity_fields_on_wire_but_platform_id_excluded(): + data = json.loads(_make_message("x").model_dump_json()) + assert data["canonical_name"] == "ProposedFileChangesMessage" + assert data["version"] == "1" + assert "platform_id" not in data diff --git a/tests/migrations/unit/test_registry.py b/tests/migrations/unit/test_registry.py new file mode 100644 index 00000000000..aceeba7a005 --- /dev/null +++ b/tests/migrations/unit/test_registry.py @@ -0,0 +1,25 @@ +"""The syft-client migration registry exists and computes a valid protocol schema.""" + +from syft_client.migrations import ( + PROTOCOL_NAME, + SYFT_CLIENT_PROTOCOL_VERSION, + client_registry, +) +from syft_client.version import SYFT_CLIENT_VERSION + + +def test_registry_identity(): + assert client_registry.protocol_name == PROTOCOL_NAME + assert client_registry.package_name == "syft-client" + assert client_registry.package_version == SYFT_CLIENT_VERSION + assert client_registry.protocol_version == SYFT_CLIENT_PROTOCOL_VERSION + + +def test_registry_computes_protocol_schema(): + schema = client_registry.compute_protocol_schema() + assert schema.protocol_name == PROTOCOL_NAME + assert schema.version == SYFT_CLIENT_PROTOCOL_VERSION + # Every registered object resolves a current version and a frozen schema. + for canonical_name in schema.supported_versions: + assert schema.current_schema(canonical_name=canonical_name) + assert canonical_name in schema.current_object_schemas diff --git a/tests/migrations/unit/test_version_info_migrations.py b/tests/migrations/unit/test_version_info_migrations.py new file mode 100644 index 00000000000..bd0adbc9abf --- /dev/null +++ b/tests/migrations/unit/test_version_info_migrations.py @@ -0,0 +1,76 @@ +"""The first real migrations: VersionInfo v1 <-> v2 (protocol schemas).""" + +import json +from pathlib import Path + +from syft_client.migrations import client_migration_service, client_registry +from syft_client.sync.version.version_info import ( + VersionInfo, + VersionInfoV1, + VersionInfoV2, +) + +LEGACY_FILE = ( + Path(__file__).parent / "fixtures" / "version_info" / "SYFT_version-0.1.117.json" +) + + +def _v1() -> VersionInfoV1: + return VersionInfoV1.model_validate(json.loads(LEGACY_FILE.read_text())) + + +def test_both_versions_registered_with_paths_both_ways(): + assert client_registry.versions("VersionInfo") == ["1", "2"] + assert client_registry.has_migration_path("VersionInfo", "1", "2") + assert client_registry.has_migration_path("VersionInfo", "2", "1") + + +def test_v1_upgrades_to_v2_with_empty_schemas(): + upgraded = client_migration_service.migrate(_v1(), "2") + assert type(upgraded) is VersionInfoV2 + assert upgraded.version == "2" + # A v1 file says nothing about package protocols. + assert upgraded.protocol_schemas == {} + assert upgraded.syft_client_version == "0.1.117" + assert upgraded.updated_at == _v1().updated_at + + +def test_v2_downgrades_to_v1_dropping_schemas(): + current = VersionInfo.current() + assert current.protocol_schemas # populated before the downgrade + downgraded = client_migration_service.migrate(current, "1") + assert type(downgraded) is VersionInfoV1 + assert downgraded.version == "1" + assert "protocol_schemas" not in downgraded.model_dump() + assert downgraded.syft_client_version == current.syft_client_version + + +def test_downgrade_for_protocol_0_peer(): + # A protocol-0 peer's schema only lists VersionInfo v1. + downgraded = client_migration_service.downgrade_for_protocol_version( + VersionInfo.current(), "0" + ) + assert type(downgraded) is VersionInfoV1 + + +def test_current_advertises_slim_schemas(): + schemas = VersionInfo.current().protocol_schemas + # The client's own schema is always present; job/dataset only when the + # optional packages are importable (both are in the workspace env, but + # the code must degrade on client-only installs). + assert "syft-client" in schemas + assert set(schemas) <= {"syft-client", "syft-job", "syft-dataset"} + client_schema = schemas["syft-client"] + assert client_schema.version == client_registry.protocol_version + assert client_schema.supported_versions == ( + client_registry.compute_protocol_schema().supported_versions + ) + # Slim on the wire: no embedded per-object JSON schemas. + for schema in schemas.values(): + assert schema.current_object_schemas == {} + + +def test_legacy_file_loads_all_the_way_to_v2(): + info = VersionInfo.from_json(LEGACY_FILE.read_text()) + assert type(info) is VersionInfoV2 + assert info.protocol_schemas == {} diff --git a/tests/migrations/unit/test_version_info_serialization.py b/tests/migrations/unit/test_version_info_serialization.py new file mode 100644 index 00000000000..5a876aaa15d --- /dev/null +++ b/tests/migrations/unit/test_version_info_serialization.py @@ -0,0 +1,62 @@ +"""VersionInfo round-trips through JSON and legacy protocol-0 files still load.""" + +import json +from pathlib import Path + +from syft_client.migrations import client_registry +from syft_client.sync.version.version_info import ( + VersionInfo, + VersionInfoV1, + VersionInfoV2, +) + +FIXTURES_DIR = Path(__file__).parent / "fixtures" / "version_info" +LEGACY_FILE = FIXTURES_DIR / "SYFT_version-0.1.117.json" + + +def test_version_info_registered_and_aliased(): + assert client_registry.versions("VersionInfo") + assert VersionInfo is VersionInfoV2 + + schema = client_registry.compute_protocol_schema() + assert "VersionInfo" in schema.supported_versions + assert schema.current_schema(canonical_name="VersionInfo") + + +def test_current_serializes_identity_fields(): + data = json.loads(VersionInfo.current().to_json()) + assert data["canonical_name"] == "VersionInfo" + assert data["version"] == "2" + + +def test_json_round_trip(): + original = VersionInfo.current() + restored = VersionInfo.from_json(original.to_json()) + assert restored == original + + +def test_legacy_protocol0_file_loads_as_latest(): + # Written by a <= 0.1.117 client: no canonical_name/version fields. + legacy_json = LEGACY_FILE.read_text() + assert "canonical_name" not in json.loads(legacy_json) + + info = VersionInfo.from_json(legacy_json) + # type() not isinstance(): V2 subclasses V1, so isinstance is vacuous. + assert type(info) is VersionInfoV2 + assert info.version == client_registry.latest_version("VersionInfo") + assert info.syft_client_version == "0.1.117" + assert info.syft_client_install_source == "pip" + + +def test_legacy_reader_tolerates_identity_fields(): + # A protocol-0 client parses with pydantic's default extra="ignore"; the + # closest stand-in we have is validating minus the identity defaults. + data = json.loads(VersionInfo.current().to_json()) + # Legacy clients see unknown keys and ignore them; simulate by checking + # the payload minus identity fields is exactly the legacy shape. + data.pop("canonical_name") + data.pop("version") + data.pop("protocol_schemas") + # Additive-only invariant: current output minus the added fields is + # exactly the legacy shape a 0.1.117 reader expects. + assert set(data) == set(json.loads(LEGACY_FILE.read_text())) diff --git a/uv.lock b/uv.lock index 51ca3f63bbc..3ca38281bc4 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.10" resolution-markers = [ "python_full_version >= '3.15' and sys_platform == 'win32'", @@ -986,7 +986,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -1506,17 +1506,17 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "colorama", marker = "python_full_version < '3.11' and sys_platform == 'win32'" }, - { name = "decorator", marker = "python_full_version < '3.11'" }, - { name = "exceptiongroup", marker = "python_full_version < '3.11'" }, - { name = "jedi", marker = "python_full_version < '3.11'" }, - { name = "matplotlib-inline", marker = "python_full_version < '3.11'" }, - { name = "pexpect", marker = "python_full_version < '3.11' and sys_platform != 'emscripten' and sys_platform != 'win32'" }, - { name = "prompt-toolkit", marker = "python_full_version < '3.11'" }, - { name = "pygments", marker = "python_full_version < '3.11'" }, - { name = "stack-data", marker = "python_full_version < '3.11'" }, - { name = "traitlets", marker = "python_full_version < '3.11'" }, - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "decorator" }, + { name = "exceptiongroup" }, + { name = "jedi" }, + { name = "matplotlib-inline" }, + { name = "pexpect", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, + { name = "prompt-toolkit" }, + { name = "pygments" }, + { name = "stack-data" }, + { name = "traitlets" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/40/18/f8598d287006885e7136451fdea0755af4ebcbfe342836f24deefaed1164/ipython-8.39.0.tar.gz", hash = "sha256:4110ae96012c379b8b6db898a07e186c40a2a1ef5d57a7fa83166047d9da7624", size = 5513971, upload-time = "2026-03-27T10:02:13.94Z" } wheels = [ @@ -1542,18 +1542,18 @@ resolution-markers = [ "python_full_version >= '3.11' and python_full_version < '3.13' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "colorama", marker = "python_full_version >= '3.11' and sys_platform == 'win32'" }, - { name = "decorator", marker = "python_full_version >= '3.11'" }, - { name = "ipython-pygments-lexers", marker = "python_full_version >= '3.11'" }, - { name = "jedi", marker = "python_full_version >= '3.11'" }, - { name = "matplotlib-inline", marker = "python_full_version >= '3.11'" }, - { name = "pexpect", marker = "python_full_version >= '3.11' and sys_platform != 'emscripten' and sys_platform != 'win32'" }, - { name = "prompt-toolkit", marker = "python_full_version >= '3.11'" }, - { name = "psutil", marker = "python_full_version >= '3.11'" }, - { name = "pygments", marker = "python_full_version >= '3.11'" }, - { name = "stack-data", marker = "python_full_version >= '3.11'" }, - { name = "traitlets", marker = "python_full_version >= '3.11'" }, - { name = "typing-extensions", marker = "python_full_version == '3.11.*'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "decorator" }, + { name = "ipython-pygments-lexers" }, + { name = "jedi" }, + { name = "matplotlib-inline" }, + { name = "pexpect", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, + { name = "prompt-toolkit" }, + { name = "psutil" }, + { name = "pygments" }, + { name = "stack-data" }, + { name = "traitlets" }, + { name = "typing-extensions", marker = "python_full_version < '3.12'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/cd/c4/87cda5842cf5c31837c06ddb588e11c3c35d8ece89b7a0108c06b8c9b00a/ipython-9.13.0.tar.gz", hash = "sha256:7e834b6afc99f020e3f05966ced34792f40267d64cb1ea9043886dab0dde5967", size = 4430549, upload-time = "2026-04-24T12:24:55.221Z" } wheels = [ @@ -1565,7 +1565,7 @@ name = "ipython-pygments-lexers" version = "1.1.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "pygments", marker = "python_full_version >= '3.11'" }, + { name = "pygments" }, ] sdist = { url = "https://files.pythonhosted.org/packages/ef/4c/5dd1d8af08107f88c7f741ead7a40854b8ac24ddf9ae850afbcf698aa552/ipython_pygments_lexers-1.1.1.tar.gz", hash = "sha256:09c0138009e56b6854f9535736f4171d855c8c08a563a0dcd8022f78355c7e81", size = 8393, upload-time = "2025-01-17T11:24:34.505Z" } wheels = [ @@ -2840,10 +2840,10 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "python-dateutil", marker = "python_full_version < '3.11'" }, - { name = "pytz", marker = "python_full_version < '3.11'" }, - { name = "tzdata", marker = "python_full_version < '3.11'" }, + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } }, + { name = "python-dateutil" }, + { name = "pytz" }, + { name = "tzdata" }, ] sdist = { url = "https://files.pythonhosted.org/packages/33/01/d40b85317f86cf08d853a4f495195c73815fdf205eef3993821720274518/pandas-2.3.3.tar.gz", hash = "sha256:e05e1af93b977f7eafa636d043f9f94c7ee3ac81af99c13508215942e64c993b", size = 4495223, upload-time = "2025-09-29T23:34:51.853Z" } wheels = [ @@ -2915,9 +2915,9 @@ resolution-markers = [ "python_full_version >= '3.11' and python_full_version < '3.13' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] dependencies = [ - { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "python-dateutil", marker = "python_full_version >= '3.11'" }, - { name = "tzdata", marker = "(python_full_version >= '3.11' and sys_platform == 'emscripten') or (python_full_version >= '3.11' and sys_platform == 'win32')" }, + { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" } }, + { name = "python-dateutil" }, + { name = "tzdata", marker = "sys_platform == 'emscripten' or sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/f8/87/4341c6252d1c47b08768c3d25ac487362bf403f0313ddae4a2a26c9b1b4c/pandas-3.0.3.tar.gz", hash = "sha256:696a4a00a2a2a35d4e5deb3fc946641b96c944f02230e4f76137fe35d806c4fc", size = 4651414, upload-time = "2026-05-11T18:54:29.21Z" } wheels = [ @@ -3002,7 +3002,7 @@ name = "pexpect" version = "4.9.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "ptyprocess", marker = "(python_full_version < '3.11' and sys_platform == 'emscripten') or (python_full_version < '3.11' and sys_platform == 'win32') or (sys_platform != 'emscripten' and sys_platform != 'win32')" }, + { name = "ptyprocess" }, ] sdist = { url = "https://files.pythonhosted.org/packages/42/92/cc564bf6381ff43ce1f4d06852fc19a2f11d180f23dc32d9588bee2f149d/pexpect-4.9.0.tar.gz", hash = "sha256:ee7d41123f3c9911050ea2c2dac107568dc43b2d3b0c7557a33212c398ead30f", size = 166450, upload-time = "2023-11-25T09:07:26.339Z" } wheels = [ @@ -4509,6 +4509,7 @@ dependencies = [ { name = "syft-crypto-python" }, { name = "syft-dataset" }, { name = "syft-job" }, + { name = "syft-migration" }, { name = "syft-permissions" }, { name = "syft-perms" }, ] @@ -4573,6 +4574,7 @@ requires-dist = [ { name = "syft-dataset", marker = "extra == 'datasets'", editable = "packages/syft-datasets" }, { name = "syft-job", editable = "packages/syft-job" }, { name = "syft-job", marker = "extra == 'job'", editable = "packages/syft-job" }, + { name = "syft-migration", editable = "packages/syft-migration" }, { name = "syft-permissions", editable = "packages/syft-permissions" }, { name = "syft-perms", editable = "packages/syft-perms" }, ]