From f6ac748f208d10ba5c16ac4c16c9a8c5e6c99fd8 Mon Sep 17 00:00:00 2001 From: rasswanth-s <43314053+rasswanth-s@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:07:43 +0530 Subject: [PATCH 1/2] refactor(transport): implement the connections on Drive and route the router through them Add DriveP2PConnection (sending), DriveOwnerStore, DriveCollections, DrivePeering and a DriveWatcherStore placeholder over GDriveConnection, and route ConnectionRouter through them; it no longer imports Drive code. Receiving stays on the Drive methods until send-and-forget. Fix: the event log is read oldest first, so restoring from it keeps the newest version of a file that changed twice. --- syft/sync/connections/base_connection.py | 17 +- syft/sync/connections/connection_router.py | 159 ++++----- .../connections/drive/drive_connections.py | 267 +++++++++++++++ .../connections/drive/gdrive_transport.py | 78 +++-- syft/sync/version/peer_manager.py | 6 +- tests/unit/test_drive_connections.py | 317 ++++++++++++++++++ tests/unit/test_peers_json_version.py | 2 +- 7 files changed, 705 insertions(+), 141 deletions(-) create mode 100644 syft/sync/connections/drive/drive_connections.py create mode 100644 tests/unit/test_drive_connections.py diff --git a/syft/sync/connections/base_connection.py b/syft/sync/connections/base_connection.py index da97faf303a..54c5cc9eec1 100644 --- a/syft/sync/connections/base_connection.py +++ b/syft/sync/connections/base_connection.py @@ -1,7 +1,10 @@ -from typing import ClassVar, Type +from typing import TYPE_CHECKING, ClassVar, Type from pydantic import BaseModel from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage +if TYPE_CHECKING: + from syft.sync.transport.protocols import Transport + class FileCollection(BaseModel): """A collection of files that can be shared and synced.""" @@ -36,6 +39,18 @@ def watcher_send_proposed_file_changes_message( def from_config(cls, config: ConnectionConfig): return config.connection_type.from_config(config) + def as_transport(self) -> "Transport": + """This connection as the transport jobs: P2P messages, owner and + watcher storage, collections and peering.""" + raise NotImplementedError() + + def clone_for_thread(self) -> "SyftboxPlatformConnection": + """A connection that another thread may use. + + This one, unless the backend's client cannot be shared across threads. + """ + return self + # ========================================================================= # GENERIC PREFIX-PARAMETERIZED COLLECTION METHODS # ========================================================================= diff --git a/syft/sync/connections/connection_router.py b/syft/sync/connections/connection_router.py index e45fc214476..35114df99fc 100644 --- a/syft/sync/connections/connection_router.py +++ b/syft/sync/connections/connection_router.py @@ -16,17 +16,15 @@ FileCollection, SyftboxPlatformConnection, ) -from syft.sync.connections.drive.gdrive_transport import ( - PEERS_META_KEY, - GDriveConnection, -) from syft.sync.events.file_change_event import ( FileChangeEventsMessage, ) from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage from syft.sync.peers.peer import Peer, PeerState from syft.sync.peers.peer_store import PeerStore -from syft.sync.platforms.gdrive_files_platform import GdriveFilesPlatform +from syft.sync.transport.naming import PEERS_META_KEY +from syft.sync.transport.protocols import Transport +from syft.sync.transport.types import MessageKind from syft.sync.utils.print_utils import ( print_peer_added_to_platform, print_peer_adding_to_platform, @@ -48,6 +46,18 @@ class ConnectionRouter(BaseModel): # live map here (updated in place as peer version files load), so outgoing # messages downgrade to what each peer reads. _peer_schemas: Dict[str, ProtocolSchema] = PrivateAttr(default_factory=dict) + # The transport jobs over connections[0], rebuilt if that connection is + # replaced (tests swap in a mock connection with add_connection). + _transport: Optional[Transport] = PrivateAttr(default=None) + _transport_source: Optional[SyftboxPlatformConnection] = PrivateAttr(default=None) + + @property + def transport(self) -> Transport: + connection = self.connections[0] + if self._transport is None or self._transport_source is not connection: + self._transport = connection.as_transport() + self._transport_source = connection + return self._transport def set_peer_schemas(self, peer_schemas: Dict[str, ProtocolSchema]) -> None: """Adopt the live {peer email -> syft ProtocolSchema} map.""" @@ -78,10 +88,7 @@ def connection_for_receive_message(self) -> SyftboxPlatformConnection: def copy_connection( self, connection: SyftboxPlatformConnection ) -> SyftboxPlatformConnection: - if isinstance(connection, GDriveConnection): - return connection.copy() - else: - return connection + return connection.clone_for_thread() def connection_for_eventlog( self, create_new: bool = False @@ -149,9 +156,7 @@ def watcher_send_proposed_file_changes_message( data = proposed_file_changes_message.as_compressed_data() data = self.peer_store.encrypt_if_needed(recipient, data) filename = proposed_file_changes_message.message_filename.as_string() - self.connection_for_send_message().watcher_send_raw_bytes_to_inbox( - recipient, filename, data - ) + self.transport.p2p.send(recipient, filename, data, MessageKind.PROPOSED_CHANGES) def owner_get_next_proposed_filechange_message( self, sender_email: str @@ -175,9 +180,7 @@ def owner_write_event_messages_to_outbox( data = events_message.as_compressed_data() data = self.peer_store.encrypt_if_needed(recipient_email, data) fname = events_message.message_filepath.as_string() - self.connection_for_outbox().owner_write_raw_bytes_to_outbox( - recipient_email, fname, data - ) + self.transport.p2p.send(recipient_email, fname, data, MessageKind.FILE_EVENTS) def watcher_get_events_messages( self, peer_email: str, since_timestamp: float | None @@ -213,11 +216,10 @@ def owner_write_events_message_to_syftbox( data = events_message.as_compressed_data() data = self.peer_store.encrypt_for_self_if_needed(data) filename = events_message.message_filepath.as_string() - self.connection_for_eventlog().owner_write_raw_bytes_to_syftbox(filename, data) + self.transport.owner_store.append_event(filename, data) def owner_get_all_accepted_events_messages(self) -> List[FileChangeEventsMessage]: - connection = self.connection_for_eventlog() - raw_list = connection.owner_download_all_raw_events_from_syftbox() + raw_list = self.transport.owner_store.fetch_all_events() return [ FileChangeEventsMessage.from_compressed_data( self.peer_store.decrypt_and_verify_for_self_if_needed(raw) @@ -230,7 +232,7 @@ def download_events_message_by_id( ) -> FileChangeEventsMessage: """Download and decrypt a single events message (thread-safe).""" connection = self.connection_for_eventlog(create_new=True) - raw = connection.owner_download_raw_bytes_by_id(events_message_id) + raw = connection.as_transport().owner_store.fetch(events_message_id) raw = self.peer_store.decrypt_and_verify_for_self_if_needed(raw) return FileChangeEventsMessage.from_compressed_data(raw) @@ -239,13 +241,12 @@ def download_events_message_by_id( # ========================================================================= def add_peer(self, peer_email: str, verbose: bool = True) -> Peer: - connection = self.connection_for_receive_message() - platform = GdriveFilesPlatform() + platform = self.transport.platform if verbose: print_peer_adding_to_platform(peer_email, platform.module_path) - connection.add_peer(peer_email=peer_email) + self.transport.peering.open_channels(peer_email) if verbose: print_peer_added_to_platform(peer_email, platform.module_path) @@ -253,8 +254,9 @@ def add_peer(self, peer_email: str, verbose: bool = True) -> Peer: def get_all_peers_from_json(self, force_download: bool = False) -> List[Peer]: """Get all peers from SYFT_peers.json with their stored state.""" - connection = self.connection_for_send_message() - peers_data = connection._get_peers_json(force_download=force_download) + peers_data = self.transport.peering.read_peer_registry( + force_download=force_download + ) peers = [] for email, data in peers_data.items(): if email == PEERS_META_KEY: @@ -273,7 +275,7 @@ def get_all_peers_from_json(self, force_download: bool = False) -> List[Peer]: continue peer = Peer( email=email, - platforms=[GdriveFilesPlatform()], + platforms=[self.transport.platform], state=state, public_encryption_bundle=data.get("public_encryption_bundle"), ) @@ -282,12 +284,11 @@ def get_all_peers_from_json(self, force_download: bool = False) -> List[Peer]: def get_peer_requests(self) -> List[Peer]: """Get pending peer requests""" - connection = self.connection_for_send_message() - peer_emails = connection.get_peer_requests() + peer_emails = self.transport.peering.incoming_requests() return [ Peer( email=peer_email, - platforms=[GdriveFilesPlatform()], + platforms=[self.transport.platform], state=PeerState.REQUESTED_BY_PEER, ) for peer_email in peer_emails @@ -300,8 +301,9 @@ def update_peer_state( public_encryption_bundle: dict | None = None, ): """Update peer state in storage""" - connection = self.connection_for_send_message() - connection._update_peer_state(peer_email, state, public_encryption_bundle) + self.transport.peering.update_peer_registry( + peer_email, state, public_encryption_bundle + ) def owner_remove_proposed_filechange_from_inbox( self, @@ -319,18 +321,15 @@ def owner_remove_proposed_filechange_from_inbox( def write_encryption_bundle(self, peer_email: str, bundle_json: str) -> None: """Write own encryption bundle for a peer.""" - connection = self.connection_for_own_syftbox() - connection.write_encryption_bundle(peer_email, bundle_json) + self.transport.peering.publish_key_bundle(peer_email, bundle_json) def share_encryption_bundles_folder(self, peer_email: str) -> None: """Share the bundles folder with a peer.""" - connection = self.connection_for_own_syftbox() - connection.share_encryption_bundles_folder(peer_email) + self.transport.peering.grant_key_bundle_read(peer_email) def read_peer_encryption_bundle(self, peer_email: str) -> str | None: """Read encryption bundle that peer wrote for us.""" - connection = self.connection_for_datasite_watcher() - return connection.read_peer_encryption_bundle(peer_email) + return self.transport.peering.read_peer_key_bundle(peer_email) # ========================================================================= # GENERIC PREFIX-PARAMETERIZED COLLECTION METHODS (with encryption) @@ -339,22 +338,17 @@ def read_peer_encryption_bundle(self, peer_email: str) -> str | None: def owner_create_collection_folder( self, prefix: str, tag: str, content_hash: str, owner_email: str ) -> str: - connection = self.connection_for_send_message() - return connection.owner_create_collection_folder( - prefix, tag, content_hash, owner_email - ) + return self.transport.collections.create(prefix, tag, content_hash, owner_email) def owner_tag_collection_as_any( self, prefix: str, tag: str, content_hash: str ) -> None: - connection = self.connection_for_send_message() - connection.owner_tag_collection_as_any(prefix, tag, content_hash) + self.transport.collections.mark_shared_with_any(prefix, tag, content_hash) def owner_share_collection( self, prefix: str, tag: str, content_hash: str, users: list[str] ) -> None: - connection = self.connection_for_send_message() - connection.owner_share_collection(prefix, tag, content_hash, users) + self.transport.collections.grant_read(prefix, tag, content_hash, users) def owner_upload_collection_files( self, @@ -370,32 +364,26 @@ def owner_upload_collection_files( name: self.peer_store.encrypt_if_needed(recipient_email, data) for name, data in files.items() } - connection = self.connection_for_send_message() - connection.owner_upload_collection_files(prefix, tag, content_hash, files) + self.transport.collections.upload(prefix, tag, content_hash, files) def owner_list_collections(self, prefix: str) -> list[str]: - connection = self.connection_for_send_message() - return connection.owner_list_collections(prefix) + return self.transport.collections.list_mine(prefix) def owner_list_all_collections_with_permissions( self, prefix: str ) -> list[FileCollection]: - connection = self.connection_for_send_message() - return connection.owner_list_all_collections_with_permissions(prefix) + return self.transport.collections.list_mine_with_permissions(prefix) def owner_delete_collection(self, prefix: str, tag: str) -> None: - connection = self.connection_for_send_message() - connection.owner_delete_collection(prefix, tag) + self.transport.collections.delete(prefix, tag) def watcher_list_collections(self, prefix: str) -> list[dict]: - connection = self.connection_for_receive_message() - return connection.watcher_list_collections(prefix) + return self.transport.collections.list_shared_with_me(prefix) def watcher_download_collection( self, prefix: str, tag: str, content_hash: str, owner_email: str ) -> dict[str, bytes]: - connection = self.connection_for_datasite_watcher() - files = connection.watcher_download_collection( + files = self.transport.collections.download( prefix, tag, content_hash, owner_email ) if self.peer_store and owner_email: @@ -417,35 +405,29 @@ def connection_for_version_read( def write_version_file(self, version_info: "VersionInfo") -> None: """Write version file to own SyftBox folder.""" - connection = self.connection_for_own_syftbox() - connection.write_version_file(version_info) + self.transport.peering.publish_version(version_info) def read_own_version_file(self) -> Optional["VersionInfo"]: """Read version file from own SyftBox folder.""" - connection = self.connection_for_own_syftbox() - return connection.read_own_version_file() + return self.transport.peering.read_own_version() def read_peer_version_file(self, peer_email: str) -> Optional["VersionInfo"]: """Read version file from a peer's SyftBox folder.""" - connection = self.connection_for_datasite_watcher() - return connection.read_peer_version_file(peer_email) + return self.transport.peering.read_peer_version(peer_email) def share_version_file_with_peer(self, peer_email: str) -> None: """Share version file with a peer so they can read it.""" - connection = self.connection_for_own_syftbox() - connection.share_version_file_with_peer(peer_email) + self.transport.peering.grant_version_read(peer_email) def watcher_get_collection_file_metadatas( self, prefix: str, tag: str, content_hash: str, owner_email: str ) -> List[dict]: - connection = self.connection_for_datasite_watcher() - return connection.watcher_get_collection_file_metadatas( + return self.transport.collections.list_files( prefix, tag, content_hash, owner_email ) def watcher_download_collection_file(self, file_id: str, owner_email: str) -> bytes: - connection = self.connection_for_datasite_watcher() - data = connection.watcher_download_collection_file(file_id) + data = self.transport.collections.fetch_file(file_id) data = self.peer_store.decrypt_dataset_if_needed(owner_email, data) return data @@ -469,8 +451,7 @@ def delete_multiple_files_by_ids( def owner_get_all_accepted_event_file_ids( self, since_timestamp: float | None = None ) -> List[str]: - connection = self.connection_for_eventlog() - return connection.owner_get_all_accepted_event_file_ids(since_timestamp) + return self.transport.owner_store.list_event_handles(since_timestamp) def gather_all_file_and_folder_ids(self) -> List[str]: connection = self.connection_for_own_syftbox() @@ -497,13 +478,11 @@ def upload_checkpoint(self, checkpoint: Checkpoint) -> str: """Serialize, encrypt, and upload a checkpoint.""" data = checkpoint.as_compressed_data() data = self.peer_store.encrypt_for_self_if_needed(data) - connection = self.connection_for_own_syftbox() - return connection.upload_raw_checkpoint(checkpoint.filename, data) + return self.transport.owner_store.put_checkpoint(checkpoint.filename, data) def get_latest_checkpoint(self) -> Checkpoint | None: """Download, decrypt, and deserialize the latest checkpoint.""" - connection = self.connection_for_own_syftbox() - raw = connection.download_raw_latest_checkpoint() + raw = self.transport.owner_store.latest_checkpoint() if raw is None: return None try: @@ -517,15 +496,13 @@ def get_events_count_since_checkpoint( self, checkpoint_timestamp: float | None ) -> int: """Count events created after the checkpoint timestamp.""" - connection = self.connection_for_eventlog() - return connection.get_events_count_since_checkpoint(checkpoint_timestamp) + return self.transport.owner_store.count_events_since(checkpoint_timestamp) def get_events_messages_since_timestamp( self, since_timestamp: float ) -> List[FileChangeEventsMessage]: """Download, decrypt, and deserialize events since a timestamp.""" - connection = self.connection_for_eventlog() - raw_list = connection.download_raw_events_since_timestamp(since_timestamp) + raw_list = self.transport.owner_store.fetch_events_since(since_timestamp) return [ FileChangeEventsMessage.from_compressed_data( self.peer_store.decrypt_and_verify_for_self_if_needed(raw) @@ -541,13 +518,11 @@ def upload_incremental_checkpoint(self, checkpoint: IncrementalCheckpoint) -> st """Serialize, encrypt, and upload an incremental checkpoint.""" data = checkpoint.as_compressed_data() data = self.peer_store.encrypt_for_self_if_needed(data) - connection = self.connection_for_own_syftbox() - return connection.upload_raw_incremental_checkpoint(checkpoint.filename, data) + return self.transport.owner_store.put_incremental(checkpoint.filename, data) def get_all_incremental_checkpoints(self) -> List[IncrementalCheckpoint]: """Download, decrypt, and deserialize all incremental checkpoints.""" - connection = self.connection_for_own_syftbox() - raw_list = connection.download_all_raw_incremental_checkpoints() + raw_list = self.transport.owner_store.all_incrementals() checkpoints = [] for raw in raw_list: try: @@ -561,18 +536,15 @@ def get_all_incremental_checkpoints(self) -> List[IncrementalCheckpoint]: def get_incremental_checkpoint_count(self) -> int: """Get the number of incremental checkpoints.""" - connection = self.connection_for_own_syftbox() - return connection.get_incremental_checkpoint_count() + return self.transport.owner_store.incremental_count() def get_next_incremental_sequence_number(self) -> int: """Get the next sequence number for incremental checkpoints.""" - connection = self.connection_for_own_syftbox() - return connection.get_next_incremental_sequence_number() + return self.transport.owner_store.next_incremental_sequence() def delete_all_incremental_checkpoints(self) -> None: """Delete all incremental checkpoints from the storage backend.""" - connection = self.connection_for_own_syftbox() - connection.delete_all_incremental_checkpoints() + self.transport.owner_store.delete_incrementals() # ========================================================================= # ROLLING STATE METHODS @@ -582,13 +554,13 @@ def upload_rolling_state(self, rolling_state: RollingState) -> str: """Serialize, encrypt, and upload rolling state.""" data = rolling_state.as_compressed_data() data = self.peer_store.encrypt_for_self_if_needed(data) - connection = self.connection_for_own_syftbox() - return connection.upload_raw_rolling_state(rolling_state.filename, data) + return self.transport.owner_store.put_rolling_state( + rolling_state.filename, data + ) def get_rolling_state(self) -> RollingState | None: """Download, decrypt, and deserialize rolling state.""" - connection = self.connection_for_own_syftbox() - raw = connection.download_raw_rolling_state() + raw = self.transport.owner_store.get_rolling_state() if raw is None: return None try: @@ -600,5 +572,4 @@ def get_rolling_state(self) -> RollingState | None: def delete_rolling_state(self) -> None: """Delete rolling state from the storage backend.""" - connection = self.connection_for_own_syftbox() - connection.delete_rolling_state() + self.transport.owner_store.delete_rolling_state() diff --git a/syft/sync/connections/drive/drive_connections.py b/syft/sync/connections/drive/drive_connections.py new file mode 100644 index 00000000000..ac521825b90 --- /dev/null +++ b/syft/sync/connections/drive/drive_connections.py @@ -0,0 +1,267 @@ +"""The transport jobs on Google Drive. + +Each class wraps a ``GDriveConnection`` and calls the method that does the job +today, so what is read from and written to Drive does not change. Where a +contract builds a method from simpler ones, the Drive class overrides it to +keep the single Drive call it makes today. The Drive code moves out of +``GDriveConnection`` into these classes, over a shared ``GDriveClient``, in a +later step. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from syft.sync.platforms.gdrive_files_platform import GdriveFilesPlatform +from syft.sync.transport.protocols import ( + Collections, + OwnerStore, + P2PConnection, + Peering, + Transport, + WatcherStore, +) +from syft.sync.transport.types import MessageKind, RawMessage, SendStatus + +if TYPE_CHECKING: + from syft.sync.connections.base_connection import FileCollection + from syft.sync.connections.drive.gdrive_transport import GDriveConnection + from syft.sync.version.version_info import VersionInfo + + +class DriveP2PConnection(P2PConnection): + """Messages through the shared inbox and outbox folders. + + A proposed change goes to the data owner's inbox and file events go to the + data scientist's outbox. ``msg_id`` is the file name. A written file is + ``DELIVERED``: it waits in the folder until the peer reads it. + + Only sending goes through this class so far. The router still reads and + archives messages with the Drive methods directly, because doing it through + ``receive`` and ``ack`` changes how they behave (one rule for both folders, + delete on ack, no read cursor); that lands with send-and-forget. + """ + + def __init__(self, connection: GDriveConnection): + self._conn = connection + + def send( + self, peer: str, msg_id: str, data: bytes, kind: MessageKind + ) -> SendStatus: + if kind is MessageKind.PROPOSED_CHANGES: + self._conn.watcher_send_raw_bytes_to_inbox(peer, msg_id, data) + else: + self._conn.owner_write_raw_bytes_to_outbox(peer, msg_id, data) + return SendStatus.DELIVERED + + def delivered(self) -> list[str]: + # Every Drive send is DELIVERED when it returns; none waits for a peer. + return [] + + def receive(self, max_messages: int | None = None) -> list[RawMessage]: + raise NotImplementedError( + "Messages are still read through the Drive methods; receiving " + "through P2PConnection lands with send-and-forget." + ) + + def ack(self, message: RawMessage) -> None: + raise NotImplementedError( + "Messages are still archived through the Drive methods; " + "acknowledging through P2PConnection lands with send-and-forget." + ) + + +class DriveOwnerStore(OwnerStore): + """The data owner's own folders in /SyftBox: event log, checkpoints, + rolling state.""" + + def __init__(self, connection: GDriveConnection): + self._conn = connection + + def append_event(self, name: str, data: bytes) -> str: + return self._conn.owner_write_raw_bytes_to_syftbox(name, data) + + def list_event_handles(self, since_timestamp: float | None = None) -> list[str]: + return self._conn.owner_get_all_accepted_event_file_ids(since_timestamp) + + def fetch(self, handle: str) -> bytes: + return self._conn.owner_download_raw_bytes_by_id(handle) + + def fetch_events_since(self, since_timestamp: float | None) -> list[bytes]: + if since_timestamp is None: + return self.fetch_all_events() + return self._conn.download_raw_events_since_timestamp(since_timestamp) + + def fetch_all_events(self) -> list[bytes]: + return self._conn.owner_download_all_raw_events_from_syftbox() + + def count_events_since(self, since_timestamp: float | None) -> int: + return self._conn.get_events_count_since_checkpoint(since_timestamp) + + def put_checkpoint(self, name: str, data: bytes) -> str: + return self._conn.upload_raw_checkpoint(name, data) + + def latest_checkpoint(self) -> bytes | None: + return self._conn.download_raw_latest_checkpoint() + + def put_incremental(self, name: str, data: bytes) -> str: + return self._conn.upload_raw_incremental_checkpoint(name, data) + + def all_incrementals(self) -> list[bytes]: + return self._conn.download_all_raw_incremental_checkpoints() + + def incremental_count(self) -> int: + return self._conn.get_incremental_checkpoint_count() + + def next_incremental_sequence(self) -> int: + return self._conn.get_next_incremental_sequence_number() + + def delete_incrementals(self) -> None: + self._conn.delete_all_incremental_checkpoints() + + def put_rolling_state(self, name: str, data: bytes) -> str: + return self._conn.upload_raw_rolling_state(name, data) + + def get_rolling_state(self) -> bytes | None: + return self._conn.download_raw_rolling_state() + + def delete_rolling_state(self) -> None: + self._conn.delete_rolling_state() + + +class DriveWatcherStore(WatcherStore): + """The data scientist's storage on Drive. Not built yet: today the data + scientist keeps what it received in a local cache.""" + + def __init__(self, connection: GDriveConnection): + self._conn = connection + + def put_checkpoint(self, peer: str, name: str, data: bytes) -> str: + raise NotImplementedError("Watcher storage on Drive is not built yet") + + def latest_checkpoint(self, peer: str) -> bytes | None: + raise NotImplementedError("Watcher storage on Drive is not built yet") + + def put_rolling_state(self, peer: str, name: str, data: bytes) -> str: + raise NotImplementedError("Watcher storage on Drive is not built yet") + + def get_rolling_state(self, peer: str) -> bytes | None: + raise NotImplementedError("Watcher storage on Drive is not built yet") + + def reset(self, peer: str) -> None: + raise NotImplementedError("Watcher storage on Drive is not built yet") + + +class DriveCollections(Collections): + """Collection folders in /SyftBox, shared by Drive permission.""" + + def __init__(self, connection: GDriveConnection): + self._conn = connection + + def create(self, prefix: str, tag: str, content_hash: str, owner_email: str) -> str: + return self._conn.owner_create_collection_folder( + prefix, tag, content_hash, owner_email + ) + + def mark_shared_with_any(self, prefix: str, tag: str, content_hash: str) -> None: + self._conn.owner_tag_collection_as_any(prefix, tag, content_hash) + + def grant_read( + self, prefix: str, tag: str, content_hash: str, users: list[str] + ) -> None: + self._conn.owner_share_collection(prefix, tag, content_hash, users) + + def upload( + self, prefix: str, tag: str, content_hash: str, files: dict[str, bytes] + ) -> None: + self._conn.owner_upload_collection_files(prefix, tag, content_hash, files) + + def list_mine_with_permissions(self, prefix: str) -> list[FileCollection]: + return self._conn.owner_list_all_collections_with_permissions(prefix) + + def list_mine(self, prefix: str) -> list[str]: + return self._conn.owner_list_collections(prefix) + + def delete(self, prefix: str, tag: str) -> None: + self._conn.owner_delete_collection(prefix, tag) + + def list_shared_with_me(self, prefix: str) -> list[dict]: + return self._conn.watcher_list_collections(prefix) + + def list_files( + self, prefix: str, tag: str, content_hash: str, owner_email: str + ) -> list[dict]: + return self._conn.watcher_get_collection_file_metadatas( + prefix, tag, content_hash, owner_email + ) + + def fetch_file(self, handle: str) -> bytes: + return self._conn.watcher_download_collection_file(handle) + + def download( + self, prefix: str, tag: str, content_hash: str, owner_email: str + ) -> dict[str, bytes]: + return self._conn.watcher_download_collection( + prefix, tag, content_hash, owner_email + ) + + +class DrivePeering(Peering): + """Identity from the Google account; channels, the peer list, key bundles + and version files in /SyftBox.""" + + def __init__(self, connection: GDriveConnection): + self._conn = connection + + def whoami(self) -> str: + return self._conn.get_authenticated_email() + + def open_channels(self, peer: str) -> None: + self._conn.add_peer(peer_email=peer) + + def incoming_requests(self) -> list[str]: + return self._conn.get_peer_requests() + + def read_peer_registry(self, force_download: bool = False) -> dict: + return self._conn._get_peers_json(force_download=force_download) + + def update_peer_registry( + self, peer: str, state: str, public_encryption_bundle: dict | None = None + ) -> None: + self._conn._update_peer_state(peer, state, public_encryption_bundle) + + def publish_key_bundle(self, peer: str, bundle_json: str) -> None: + self._conn.write_encryption_bundle(peer, bundle_json) + + def grant_key_bundle_read(self, peer: str) -> None: + self._conn.share_encryption_bundles_folder(peer) + + def read_own_key_bundle(self, peer: str) -> str | None: + return self._conn.read_own_encryption_bundle(peer) + + def read_peer_key_bundle(self, peer: str) -> str | None: + return self._conn.read_peer_encryption_bundle(peer) + + def publish_version(self, version_info: VersionInfo) -> None: + self._conn.write_version_file(version_info) + + def read_own_version(self) -> VersionInfo | None: + return self._conn.read_own_version_file() + + def read_peer_version(self, peer: str) -> VersionInfo | None: + return self._conn.read_peer_version_file(peer) + + def grant_version_read(self, peer: str) -> None: + self._conn.share_version_file_with_peer(peer) + + +def drive_transport(connection: GDriveConnection) -> Transport: + """The transport jobs over one Drive connection.""" + return Transport( + p2p=DriveP2PConnection(connection), + owner_store=DriveOwnerStore(connection), + watcher_store=DriveWatcherStore(connection), + collections=DriveCollections(connection), + peering=DrivePeering(connection), + platform=GdriveFilesPlatform(), + ) diff --git a/syft/sync/connections/drive/gdrive_transport.py b/syft/sync/connections/drive/gdrive_transport.py index 30b22fd878a..42a54bab529 100644 --- a/syft/sync/connections/drive/gdrive_transport.py +++ b/syft/sync/connections/drive/gdrive_transport.py @@ -38,6 +38,7 @@ from syft.sync.connections.drive.gdrive_utils import ( gather_all_file_and_folder_ids_recursive, ) +from syft.sync.connections.drive.drive_connections import drive_transport from syft.sync.environments.environment import Environment from syft.sync.events.file_change_event import ( FileChangeEventsMessage, @@ -48,6 +49,11 @@ MessageFileName, ProposedFileChangesMessage, ) +from syft.sync.transport.naming import ( # noqa: F401 (re-exported) + PEERS_META_KEY, + SYFT_PEERS_VERSION, + timestamp_from_message_name, +) from syft.sync.utils.syftbox_utils import check_env from syft.sync.version.version_info import _parse_semver from syft.version import SYFT_VERSION @@ -97,14 +103,6 @@ def build_drive_service( GDRIVE_P2P_FOLDER_DATASITE_PREFIX = "syft_datasite" SYFT_PEERS_FILE = "SYFT_peers.json" -# SYFT_peers.json is a flat map of peer email to entry, so a version at the top -# level would look like a peer email. The version goes under this reserved key. -# A client written before the key reads a peer state from that entry and fails. -# The key therefore never appears as a peer. -PEERS_META_KEY = "_meta" -# Shape of one entry in SYFT_peers.json. Raise this when an entry changes. A file -# with no reserved entry was written before the version, and is version 0. -SYFT_PEERS_VERSION = 1 SYFT_VERSION_FILE = "SYFT_version.json" @@ -441,6 +439,14 @@ def setup( self.get_personal_syftbox_folder_id() self._is_setup = True + def as_transport(self): + """This connection as the transport jobs. See ``drive_connections.py``.""" + return drive_transport(self) + + def clone_for_thread(self) -> "GDriveConnection": + """A connection safe to use from another thread: the Drive client is not.""" + return self.copy() + def copy(self) -> "GDriveConnection": # if is mock from syft.sync.connections.drive.mock_drive_service import ( @@ -762,7 +768,13 @@ def owner_get_all_accepted_event_file_ids( personal_syftbox_folder_id, since_timestamp=since_timestamp ) valid_fname_objs = self._filter_valid_file_metadatas(file_metadatas) - return [f["id"] for f in valid_fname_objs] + oldest_first = sorted( + valid_fname_objs, + key=lambda f: FileChangeEventsMessageFileName.from_string( + f["name"] + ).timestamp, + ) + return [f["id"] for f in oldest_first] def owner_download_all_raw_events_from_syftbox(self) -> list[bytes]: """Download all event files from /SyftBox/myemail as raw bytes.""" @@ -771,7 +783,7 @@ def owner_download_all_raw_events_from_syftbox(self) -> list[bytes]: valid_fname_objs = self._get_valid_events_from_file_metadatas(file_metadatas) result = [] - for fname_obj in valid_fname_objs: + for fname_obj in sorted(valid_fname_objs, key=lambda x: x.timestamp): gdrive_id = [ f for f in file_metadatas if f["name"] == fname_obj.as_string() ][0]["id"] @@ -965,35 +977,8 @@ def owner_get_archive_folder_id(self, sender_email: str) -> str: @staticmethod def _extract_timestamp_from_filename(filename: str) -> float | None: - """ - Extract timestamp from filename. - - Supports multiple filename formats: - - Event files: syfteventsmessagev3__.tar.gz - - Job files: msgv2__.tar.gz - - Args: - filename: The filename to parse - - Returns: - Timestamp as float, or None if can't parse - """ - try: - # Try event file format first - if filename.startswith("syfteventsmessagev3_"): - parts = filename.split("_") - if len(parts) >= 2: - return float(parts[1]) - - # Try job file format - if filename.startswith("msgv2_"): - parts = filename.split("_") - if len(parts) >= 2: - return float(parts[1]) - - return None - except (ValueError, IndexError): - return None + """The send time in a message file name. See ``naming.py``.""" + return timestamp_from_message_name(filename) def get_file_metadatas_from_folder( self, @@ -2232,7 +2217,7 @@ def download_raw_events_since_timestamp( valid_fname_objs = self._get_valid_events_from_file_metadatas(file_metadatas) result = [] - for fname_obj in valid_fname_objs: + for fname_obj in sorted(valid_fname_objs, key=lambda x: x.timestamp): if fname_obj.timestamp <= since_timestamp: continue @@ -2450,6 +2435,19 @@ def share_encryption_bundles_folder(self, peer_email: str) -> None: folder_id = self._get_or_create_encryption_bundles_folder_id() self.add_permission(folder_id, peer_email, write=False) + def read_own_encryption_bundle(self, peer_email: str) -> str | None: + """Read the bundle this client published for ``peer_email``, or None.""" + folder_id = self._get_or_create_encryption_bundles_folder_id() + filename = self._encryption_bundle_filename(self.email, peer_email) + query = f"name='{filename}' and '{folder_id}' in parents and trashed=false" + results = execute_with_retries( + self.drive_service.files().list(q=query, fields="files(id)") + ) + items = results.get("files", []) + if not items: + return None + return self.download_file(items[0]["id"]).decode("utf-8") + def read_peer_encryption_bundle(self, peer_email: str) -> str | None: """Read encryption bundle that peer wrote for us. diff --git a/syft/sync/version/peer_manager.py b/syft/sync/version/peer_manager.py index f14dbd1c9ca..07554d70bbe 100644 --- a/syft/sync/version/peer_manager.py +++ b/syft/sync/version/peer_manager.py @@ -823,14 +823,10 @@ def load_peers(self, force_download: bool = False): } for email in peer_request_emails: if email not in known_emails: - from syft.sync.platforms.gdrive_files_platform import ( - GdriveFilesPlatform, - ) - peers.append( Peer( email=email, - platforms=[GdriveFilesPlatform()], + platforms=[self.connection_router.transport.platform], state=PeerState.REQUESTED_BY_PEER, ) ) diff --git a/tests/unit/test_drive_connections.py b/tests/unit/test_drive_connections.py new file mode 100644 index 00000000000..b2cdd95d935 --- /dev/null +++ b/tests/unit/test_drive_connections.py @@ -0,0 +1,317 @@ +"""The transport jobs on Google Drive, and the router that calls them. + +The Drive classes wrap today's GDriveConnection methods, so these tests pin +what each job promises, on the mock Drive service the rest of the suite uses. +""" + +import inspect +import uuid +from unittest.mock import Mock + +import pytest + +import syft.sync.connections.connection_router as connection_router_module +from syft.sync.connections.connection_router import ConnectionRouter +from syft.sync.connections.drive.gdrive_transport import CollectionFolder +from syft.sync.connections.drive.mock_drive_service import pair_with_mock_service +from syft.sync.events.file_change_event import FileChangeEventsMessage +from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage +from syft.sync.peers.peer_store import PeerStore +from syft.sync.platforms.gdrive_files_platform import GdriveFilesPlatform +from syft.sync.syftbox_manager import SyftboxManager +from syft.sync.transport import ( + Collections, + MessageKind, + OwnerStore, + P2PConnection, + Peering, + RawMessage, + SendStatus, + Transport, + WatcherStore, +) +from syft.sync.version.version_info import VersionInfo + +DO = "do@example.com" +DS = "ds@example.com" +DATASETS = "syft_datasetcollection" + + +def _proposal_name(ts: float) -> str: + return f"msgv2_{ts}_{uuid.uuid4().hex}.tar.gz" + + +def _events_name(ts: float) -> str: + return f"syfteventsmessagev3_{ts}_{uuid.uuid4()}.tar.gz" + + +@pytest.fixture +def pair(): + """DO and DS connections over one mock Drive, with the DS's channels open.""" + do_conn, ds_conn = pair_with_mock_service(DO, DS) + ds_conn.as_transport().peering.open_channels(DO) + return do_conn, ds_conn + + +# -- the transport ---------------------------------------------------------------- + + +def test_drive_transport_does_every_job(): + do_conn, _ = pair_with_mock_service(DO, DS) + transport = do_conn.as_transport() + assert isinstance(transport, Transport) + assert isinstance(transport.p2p, P2PConnection) + assert isinstance(transport.owner_store, OwnerStore) + assert isinstance(transport.watcher_store, WatcherStore) + assert isinstance(transport.collections, Collections) + assert isinstance(transport.peering, Peering) + assert isinstance(transport.platform, GdriveFilesPlatform) + + +# -- P2PConnection ---------------------------------------------------------------- + + +def test_a_proposed_change_sent_through_p2p_reaches_the_owner_inbox(pair): + do_conn, ds_conn = pair + status = ds_conn.as_transport().p2p.send( + DO, _proposal_name(1.0), b"job", MessageKind.PROPOSED_CHANGES + ) + assert status is SendStatus.DELIVERED + data, _ = do_conn.owner_download_next_raw_proposed_message_from_inbox(DS) + assert data == b"job" + + +def test_file_events_sent_through_p2p_reach_the_scientist_outbox(pair): + do_conn, ds_conn = pair + name = _events_name(1.0) + status = do_conn.as_transport().p2p.send( + DS, name, b"results", MessageKind.FILE_EVENTS + ) + assert status is SendStatus.DELIVERED + listed = ds_conn.watcher_get_outbox_file_metadatas(DO, since_timestamp=None) + assert [m["file_name"] for m in listed] == [name] + + +def test_drive_never_has_sends_waiting_for_confirmation(pair): + do_conn, _ = pair + assert do_conn.as_transport().p2p.delivered() == [] + + +def test_receiving_through_p2p_is_not_built_yet(pair): + do_conn, _ = pair + p2p = do_conn.as_transport().p2p + message = RawMessage( + sender=DS, msg_id="m", kind=MessageKind.PROPOSED_CHANGES, data=b"" + ) + with pytest.raises(NotImplementedError): + p2p.receive() + with pytest.raises(NotImplementedError): + p2p.ack(message) + + +# -- OwnerStore ------------------------------------------------------------------- + + +def test_owner_event_log_is_read_oldest_first(pair): + do_conn, _ = pair + store = do_conn.as_transport().owner_store + # Written out of order: Drive lists by name, newest first. + second = store.append_event(_events_name(2.0), b"second") + first = store.append_event(_events_name(1.0), b"first") + + assert store.list_event_handles() == [first, second] + assert store.fetch_all_events() == [b"first", b"second"] + assert store.fetch_events_since(None) == [b"first", b"second"] + assert store.fetch_events_since(1.5) == [b"second"] + assert store.fetch(first) == b"first" + assert store.count_events_since(None) == 2 + assert store.count_events_since(1.5) == 1 + + +def test_owner_store_keeps_the_newest_full_checkpoint(pair): + do_conn, _ = pair + store = do_conn.as_transport().owner_store + store.put_checkpoint("checkpoint_1.0.tar.gz", b"one") + store.put_checkpoint("checkpoint_2.0.tar.gz", b"two") + assert store.latest_checkpoint() == b"two" + + +def test_owner_store_counts_and_deletes_incremental_checkpoints(pair): + do_conn, _ = pair + store = do_conn.as_transport().owner_store + store.put_incremental("incremental_checkpoint_1_1.0.tar.gz", b"inc") + assert store.incremental_count() == 1 + assert store.next_incremental_sequence() == 2 + assert store.all_incrementals() == [b"inc"] + store.delete_incrementals() + assert store.incremental_count() == 0 + + +def test_owner_store_overwrites_and_deletes_the_rolling_state(pair): + do_conn, _ = pair + store = do_conn.as_transport().owner_store + store.put_rolling_state("rolling_state_1.0.tar.gz", b"state 1") + store.put_rolling_state("rolling_state_2.0.tar.gz", b"state 2") + assert store.get_rolling_state() == b"state 2" + store.delete_rolling_state() + assert store.get_rolling_state() is None + + +# -- WatcherStore ----------------------------------------------------------------- + + +def test_watcher_storage_on_drive_is_not_built_yet(pair): + _, ds_conn = pair + with pytest.raises(NotImplementedError): + ds_conn.as_transport().watcher_store.latest_checkpoint(DO) + + +# -- Collections ------------------------------------------------------------------ + + +def test_a_published_collection_is_listed_and_downloaded_by_its_reader(pair): + do_conn, ds_conn = pair + owner, reader = do_conn.as_transport(), ds_conn.as_transport() + files = {"mock.csv": b"a,b\n1,2\n"} + content_hash = CollectionFolder.compute_hash(files) + + owner.collections.create(DATASETS, "census", content_hash, DO) + owner.collections.upload(DATASETS, "census", content_hash, files) + owner.collections.grant_read(DATASETS, "census", content_hash, [DS]) + + assert owner.collections.list_mine(DATASETS) == ["census"] + mine = owner.collections.list_mine_with_permissions(DATASETS) + assert [c.tag for c in mine] == ["census"] + + shared = reader.collections.list_shared_with_me(DATASETS) + assert [(c["owner_email"], c["tag"]) for c in shared] == [(DO, "census")] + assert reader.collections.download(DATASETS, "census", content_hash, DO) == files + [listed] = reader.collections.list_files(DATASETS, "census", content_hash, DO) + assert reader.collections.fetch_file(listed["file_id"]) == files["mock.csv"] + + owner.collections.delete(DATASETS, "census") + assert owner.collections.list_mine(DATASETS) == [] + + +# -- Peering ---------------------------------------------------------------------- + + +def test_peering_finds_the_peer_that_opened_channels(pair): + do_conn, _ = pair + peering = do_conn.as_transport().peering + assert peering.whoami() == DO + assert DS in peering.incoming_requests() + + +def test_peering_keeps_the_peer_registry(pair): + do_conn, _ = pair + peering = do_conn.as_transport().peering + peering.update_peer_registry(DS, "accepted") + assert peering.read_peer_registry(force_download=True)[DS]["state"] == "accepted" + + +def test_peering_publishes_a_key_bundle_both_sides_can_read(pair): + do_conn, ds_conn = pair + owner, scientist = do_conn.as_transport(), ds_conn.as_transport() + bundle = '{"bundle": "do-public-key"}' + + owner.peering.publish_key_bundle(DS, bundle) + owner.peering.grant_key_bundle_read(DS) + + assert scientist.peering.read_peer_key_bundle(DO) == bundle + assert owner.peering.read_own_key_bundle(DS) == bundle + assert owner.peering.read_own_key_bundle("someone-else@example.com") is None + + +def test_peering_publishes_a_version_the_peer_can_read(pair): + do_conn, ds_conn = pair + owner, scientist = do_conn.as_transport(), ds_conn.as_transport() + version = VersionInfo.current() + + owner.peering.publish_version(version) + owner.peering.grant_version_read(DS) + + assert owner.peering.read_own_version() == version + assert scientist.peering.read_peer_version(DO) == version + + +# -- ConnectionRouter ------------------------------------------------------------- + + +def _router_with_mock_p2p(): + do_conn, _ = pair_with_mock_service(DO, DS) + router = ConnectionRouter(connections=[do_conn], peer_store=PeerStore(email=DO)) + drive = do_conn.as_transport() + p2p = Mock(spec=P2PConnection) + router._transport = drive.model_copy(update={"p2p": p2p}) + router._transport_source = do_conn + return router, p2p + + +def test_the_router_sends_file_events_through_p2p(): + router, p2p = _router_with_mock_p2p() + message = FileChangeEventsMessage(events=[]) + router.owner_write_event_messages_to_outbox(DS, message) + + peer, msg_id, _, kind = p2p.send.call_args.args + assert (peer, msg_id, kind) == ( + DS, + message.message_filepath.as_string(), + MessageKind.FILE_EVENTS, + ) + + +def test_the_router_sends_proposed_changes_through_p2p(): + router, p2p = _router_with_mock_p2p() + message = ProposedFileChangesMessage(sender_email=DO, proposed_file_changes=[]) + router.watcher_send_proposed_file_changes_message(DS, message) + + peer, msg_id, _, kind = p2p.send.call_args.args + assert (peer, msg_id, kind) == ( + DS, + message.message_filename.as_string(), + MessageKind.PROPOSED_CHANGES, + ) + + +def test_the_router_reuses_its_transport_until_its_connection_changes(): + do_conn, other_conn = pair_with_mock_service(DO, DS) + router = ConnectionRouter(connections=[do_conn], peer_store=PeerStore(email=DO)) + first = router.transport + assert router.transport is first + + router.connections[0] = other_conn + assert router.transport is not first + + +def test_the_router_holds_no_drive_specific_code(): + source = inspect.getsource(connection_router_module).lower() + assert "gdrive" not in source + assert "drive_service" not in source + + +# -- restoring the owner's state -------------------------------------------------- + + +def test_restoring_from_the_event_log_keeps_the_newest_version(): + """A fresh client that restores from the event log alone replays the + messages oldest first, so a file changed twice ends at its last version.""" + _, do = SyftboxManager.pair_with_mock_drive_service_connection( + use_in_memory_cache=True + ) + files = do.datasite_owner_syncer.event_cache.file_connection + files.write_file("public/report.txt", "version 1") + do.sync() + files.write_file("public/report.txt", "version 2") + do.sync() + + router = do._connection_router + router.delete_rolling_state() + router.delete_all_incremental_checkpoints() + + fresh = do._copy() + fresh.datasite_owner_syncer.pull_initial_state() + restored = fresh.datasite_owner_syncer.event_cache.file_connection.read_file( + "public/report.txt" + ) + assert restored == "version 2" diff --git a/tests/unit/test_peers_json_version.py b/tests/unit/test_peers_json_version.py index 6f9cae2af59..17178da8c18 100644 --- a/tests/unit/test_peers_json_version.py +++ b/tests/unit/test_peers_json_version.py @@ -33,7 +33,7 @@ def _conn(peers_data): def _router(conn): router = Mock() - router.connection_for_send_message = Mock(return_value=conn) + router.transport = conn.as_transport() from syft.sync.connections.connection_router import ConnectionRouter return ConnectionRouter.get_all_peers_from_json.__get__(router, ConnectionRouter) From aec5f30f0e6b7ad7d1e5d4f8549b4fd354106787 Mon Sep 17 00:00:00 2001 From: rasswanth-s <43314053+rasswanth-s@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:38:47 +0530 Subject: [PATCH 2/2] refactor(transport): address review on the Drive connections --- syft/sync/connections/drive/drive_connections.py | 7 +++---- syft/sync/sync/datasite_owner_syncer.py | 9 ++------- tests/unit/test_drive_connections.py | 15 +++++++++++++++ 3 files changed, 20 insertions(+), 11 deletions(-) diff --git a/syft/sync/connections/drive/drive_connections.py b/syft/sync/connections/drive/drive_connections.py index ac521825b90..d440e3814eb 100644 --- a/syft/sync/connections/drive/drive_connections.py +++ b/syft/sync/connections/drive/drive_connections.py @@ -50,8 +50,10 @@ def send( ) -> SendStatus: if kind is MessageKind.PROPOSED_CHANGES: self._conn.watcher_send_raw_bytes_to_inbox(peer, msg_id, data) - else: + elif kind is MessageKind.FILE_EVENTS: self._conn.owner_write_raw_bytes_to_outbox(peer, msg_id, data) + else: + raise ValueError(f"Drive has no folder for messages of kind {kind!r}") return SendStatus.DELIVERED def delivered(self) -> list[str]: @@ -95,9 +97,6 @@ def fetch_events_since(self, since_timestamp: float | None) -> list[bytes]: def fetch_all_events(self) -> list[bytes]: return self._conn.owner_download_all_raw_events_from_syftbox() - def count_events_since(self, since_timestamp: float | None) -> int: - return self._conn.get_events_count_since_checkpoint(since_timestamp) - def put_checkpoint(self, name: str, data: bytes) -> str: return self._conn.upload_raw_checkpoint(name, data) diff --git a/syft/sync/sync/datasite_owner_syncer.py b/syft/sync/sync/datasite_owner_syncer.py index 4ca8204a878..208e7b428b8 100644 --- a/syft/sync/sync/datasite_owner_syncer.py +++ b/syft/sync/sync/datasite_owner_syncer.py @@ -225,13 +225,8 @@ def sync(self, peer_emails: list[str], recompute_hashes: bool = True): def download_events_message_by_id_with_connection( self, events_message_id: str ) -> FileChangeEventsMessage: - # we need a new connection object because gdrive connections are not thread safe - connection = self.connection_router.connection_for_eventlog(create_new=True) - raw = connection.owner_download_raw_bytes_by_id(events_message_id) - raw = self.connection_router.peer_store.decrypt_and_verify_for_self_if_needed( - raw - ) - return FileChangeEventsMessage.from_compressed_data(raw) + """One event-log message, on its own connection (safe to run in parallel).""" + return self.connection_router.download_events_message_by_id(events_message_id) def get_all_accepted_events_messages( self, since_timestamp: float | None = None diff --git a/tests/unit/test_drive_connections.py b/tests/unit/test_drive_connections.py index b2cdd95d935..8073ca15b7c 100644 --- a/tests/unit/test_drive_connections.py +++ b/tests/unit/test_drive_connections.py @@ -92,6 +92,12 @@ def test_file_events_sent_through_p2p_reach_the_scientist_outbox(pair): assert [m["file_name"] for m in listed] == [name] +def test_an_unknown_message_kind_is_refused(pair): + do_conn, _ = pair + with pytest.raises(ValueError): + do_conn.as_transport().p2p.send(DS, _events_name(1.0), b"x", "unknown-kind") + + def test_drive_never_has_sends_waiting_for_confirmation(pair): do_conn, _ = pair assert do_conn.as_transport().p2p.delivered() == [] @@ -128,6 +134,15 @@ def test_owner_event_log_is_read_oldest_first(pair): assert store.count_events_since(1.5) == 1 +def test_owner_event_count_matches_the_listed_handles(pair): + do_conn, _ = pair + store = do_conn.as_transport().owner_store + store.append_event(_events_name(1.0), b"event") + store.append_event("syfteventsmessagev3_not-a-valid-name.tar.gz", b"junk") + + assert store.count_events_since(None) == len(store.list_event_handles()) == 1 + + def test_owner_store_keeps_the_newest_full_checkpoint(pair): do_conn, _ = pair store = do_conn.as_transport().owner_store