diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index 05e245f710b..10cd91b08ab 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -15,7 +15,7 @@ from syft_rds import SyftRDSClient, SyftRDSClientConfig from syft_rds.waiting import JobCheck, JobStatusArg, wait_for_job -from syft.sync.transport.protocols import Transport +from syft.sync.connections.protocols import Connections from syft.sync.encryption_policy import EncryptionPolicy from syft.sync.peers.peer import Peer from syft.sync.peers.peer_list import PeerList @@ -783,7 +783,7 @@ def from_config( cls, config: SyftRDSClientConfig, data_owners: list[str] | None = None, - transport: Transport | None = None, + connections: Connections | None = None, ) -> "SyftEnclaveClient": """Build a SyftEnclaveClient from an RDS config with a wrapped job_client. @@ -791,10 +791,10 @@ def from_config( ``peer_manager_config.encryption_policy``; ``SyftboxManager.from_config`` (invoked inside ``SyftRDSClient.from_config``) loads/generates this datasite's keys from its per-datasite key file, so no extra step is needed. - ``transport`` replaces the one ``config.sync`` describes; tests pass - one over a mock Drive. + ``connections`` replace the ones ``config.sync`` describes; tests pass + them over a mock Drive. """ - rds = SyftRDSClient.from_config(config, transport) + rds = SyftRDSClient.from_config(config, connections) rds.job_client = EnclaveJobClient(rds.job_client) sync_engine = rds.sync_engine diff --git a/packages/syft-enclave/src/syft_enclaves/utils.py b/packages/syft-enclave/src/syft_enclaves/utils.py index 6403696865f..84281b55bbd 100644 --- a/packages/syft-enclave/src/syft_enclaves/utils.py +++ b/packages/syft-enclave/src/syft_enclaves/utils.py @@ -5,7 +5,7 @@ ) from syft.sync.connections.drive.drive_cached_client import DriveCachedClient from syft.sync.syftbox_manager import SyftboxManager -from syft.sync.transport.protocols import Transport +from syft.sync.connections.protocols import Connections def create_configs( @@ -57,15 +57,15 @@ def create_clients(configs) -> tuple: backing_store = MockDriveBackingStore() return tuple( SyftEnclaveClient.from_config( - c, transport=_mock_transport(backing_store, c.sync.email) + c, connections=_mock_connections(backing_store, c.sync.email) ) for c in configs ) -def _mock_transport(backing_store: MockDriveBackingStore, email: str) -> Transport: +def _mock_connections(backing_store: MockDriveBackingStore, email: str) -> Connections: service = MockDriveService(backing_store, email) - return SyftboxManager._drive_transport( + return SyftboxManager._drive_connections( DriveCachedClient.from_service(email, service) ) diff --git a/packages/syft-rds/src/syft_rds/client.py b/packages/syft-rds/src/syft_rds/client.py index f34a6b41c5c..cb112eb5011 100644 --- a/packages/syft-rds/src/syft_rds/client.py +++ b/packages/syft-rds/src/syft_rds/client.py @@ -13,7 +13,7 @@ from pydantic import BaseModel, ConfigDict from syft_migration import ProtocolSchema -from syft.sync.transport.protocols import Transport +from syft.sync.connections.protocols import Connections from syft.sync.syftbox_manager import SyftboxManager from syft.sync.peers.peer import Peer from syft.sync.utils.waiting import DEFAULT_POLL_INTERVAL, DEFAULT_WAIT_TIMEOUT @@ -152,11 +152,11 @@ def model_post_init(self, __context: Any) -> None: def from_config( cls, config: "SyftRDSClientConfig", - transport: Transport | None = None, + connections: Connections | None = None, ) -> "SyftRDSClient": - """``transport`` replaces the one ``config.sync`` describes; tests pass - one over a mock Drive (see ``SyftboxManager.from_config``).""" - sync_engine = SyftboxManager.from_config(config.sync, transport) + """``connections`` replace the ones ``config.sync`` describes; tests + pass them over a mock Drive (see ``SyftboxManager.from_config``).""" + sync_engine = SyftboxManager.from_config(config.sync, connections) # The job client selects a job protocol version per peer from the # schema the peer advertises, as the dataset manager does for layouts. job_client = JobClient.from_config( diff --git a/packages/syft-rds/tests/test_dataset_upload_private.py b/packages/syft-rds/tests/test_dataset_upload_private.py index a015c9bab6d..785fc930e48 100644 --- a/packages/syft-rds/tests/test_dataset_upload_private.py +++ b/packages/syft-rds/tests/test_dataset_upload_private.py @@ -144,7 +144,7 @@ def test_ds_cannot_find_private_folders_via_gdrive_query(self): # Get the DS's raw GDrive connection and search for private folders ds_cached_client = drive_cached_client_of( - ds_manager.sync_engine._connection_router.transport + ds_manager.sync_engine._connection_router.connections ) results = ( ds_cached_client.client.drive_service.files() @@ -158,7 +158,7 @@ def test_ds_cannot_find_private_folders_via_gdrive_query(self): # DO should find it with the same query do_cached_client = drive_cached_client_of( - do_manager.sync_engine._connection_router.transport + do_manager.sync_engine._connection_router.connections ) do_results = ( do_cached_client.client.drive_service.files() diff --git a/packages/syft-rds/tests/test_sync_manager.py b/packages/syft-rds/tests/test_sync_manager.py index df9abb69af0..1faa2819bd3 100644 --- a/packages/syft-rds/tests/test_sync_manager.py +++ b/packages/syft-rds/tests/test_sync_manager.py @@ -234,7 +234,7 @@ def test_dataset_only_mock_data_uploaded(): ds_manager.sync() files = drive_cached_client_of( - ds_manager.sync_engine._connection_router.transport + ds_manager.sync_engine._connection_router.connections ).client.drive_service._backing_store.files list(files.keys()) file_objs = list(files.values()) @@ -285,7 +285,7 @@ def test_jobs(): # a short cut, but we could do this based on timing eventually (if there are items in the # queue for longer than a certain time we start pushing) do_cached_client = drive_cached_client_of( - do_manager.sync_engine._connection_router.transport + do_manager.sync_engine._connection_router.connections ) inbox_folder_id = do_cached_client.get_own_datasite_inbox_id(ds_manager.email) inbox_file_metadatas = do_cached_client.get_file_metadatas_from_folder( @@ -1043,7 +1043,7 @@ def test_ds_dataset_cache_aware_sync(): # Get the mock backing store and directories for creating second pair mock_backing_store = drive_cached_client_of( - ds_manager.sync_engine._connection_router.transport + ds_manager.sync_engine._connection_router.connections ).client.drive_service._backing_store ds_folder = ds_manager.syftbox_folder do_folder = do_manager.syftbox_folder @@ -1062,10 +1062,10 @@ def test_ds_dataset_cache_aware_sync(): # Replace mock backing store to share dataset collections drive_cached_client_of( - ds_manager2.sync_engine._connection_router.transport + ds_manager2.sync_engine._connection_router.connections ).client.drive_service._backing_store = mock_backing_store drive_cached_client_of( - do_manager2.sync_engine._connection_router.transport + do_manager2.sync_engine._connection_router.connections ).client.drive_service._backing_store = mock_backing_store # Load peers (already approved in shared backing store) @@ -1143,7 +1143,7 @@ def test_do_dataset_cache_aware_sync(): # Get the mock backing store and directories for creating second pair mock_backing_store = drive_cached_client_of( - ds_manager.sync_engine._connection_router.transport + ds_manager.sync_engine._connection_router.connections ).client.drive_service._backing_store ds_folder = ds_manager.syftbox_folder do_folder = do_manager.syftbox_folder @@ -1162,10 +1162,10 @@ def test_do_dataset_cache_aware_sync(): # Replace mock backing store to share dataset collections drive_cached_client_of( - ds_manager2.sync_engine._connection_router.transport + ds_manager2.sync_engine._connection_router.connections ).client.drive_service._backing_store = mock_backing_store drive_cached_client_of( - do_manager2.sync_engine._connection_router.transport + do_manager2.sync_engine._connection_router.connections ).client.drive_service._backing_store = mock_backing_store # Load peers (already approved in shared backing store) @@ -1210,7 +1210,7 @@ def test_in_memory_connection_load_state(): # Get the backing store that will persist across "restarts" backing_store = drive_cached_client_of( - do_manager1.sync_engine._connection_router.transport + do_manager1.sync_engine._connection_router.connections ).client.drive_service._backing_store ds_folder = ds_manager1.syftbox_folder do_folder = do_manager1.syftbox_folder @@ -1277,10 +1277,10 @@ def test_in_memory_connection_load_state(): ds_cached_client2 = DriveCachedClient.from_service(ds_email, service_ds) do_manager2 = SyftRDSClient.from_config( - do_config2, SyftboxManager._drive_transport(do_cached_client2) + do_config2, SyftboxManager._drive_connections(do_cached_client2) ) ds_manager2 = SyftRDSClient.from_config( - ds_config2, SyftboxManager._drive_transport(ds_cached_client2) + ds_config2, SyftboxManager._drive_connections(ds_cached_client2) ) # Load peers diff --git a/packages/syft-rds/tests/test_version_mismatch_flow.py b/packages/syft-rds/tests/test_version_mismatch_flow.py index 7b4728ac12d..5d74221fb18 100644 --- a/packages/syft-rds/tests/test_version_mismatch_flow.py +++ b/packages/syft-rds/tests/test_version_mismatch_flow.py @@ -7,7 +7,7 @@ from unittest.mock import patch -from syft.sync.connections.drive.drive_peering import DrivePeering +from syft.sync.connections.drive.drive_peering import DrivePeeringConnection from syft.sync.connections.drive.gdrive_client import GOOGLE_FOLDER_MIME_TYPE from syft.sync.connections.drive.drive_cached_client import ( GDRIVE_P2P_FOLDER_DATASITE_PREFIX, @@ -52,7 +52,7 @@ def _find_versioned_p2p_folders(cached_client, peer_email, version): def _get_backing_store(manager): """Extract the mock backing store from a manager's connection.""" cached_client = drive_cached_client_of( - manager.peer_manager.connection_router.transport + manager.peer_manager.connection_router.connections ) return cached_client.client.drive_service._backing_store @@ -76,7 +76,7 @@ def _reinitialize_manager( mock_service = MockDriveService(backing_store, email) cached_client = DriveCachedClient.from_service(email, mock_service) manager = SyftRDSClient.from_config( - config, SyftboxManager._drive_transport(cached_client) + config, SyftboxManager._drive_connections(cached_client) ) if has_ds_role: @@ -104,7 +104,7 @@ def _simulate_continue_on_mismatch(manager, backing_store): cached_client = DriveCachedClient.from_service(email, mock_service) def read_remote(e, t): - return DrivePeering(cached_client).read_own_peer_info().version_info + return DrivePeeringConnection(cached_client).read_own_peer_info().version_info with ( patch( @@ -180,7 +180,7 @@ def test_version_mismatch_continues_and_repairs(): # -- Step 4: Record P2P folders and personal folder id before upgrade -- do_cached_client = drive_cached_client_of( - do_manager.peer_manager.connection_router.transport + do_manager.peer_manager.connection_router.connections ) do_p2p_current = _find_versioned_p2p_folders( do_cached_client, ds_manager.email, SYFT_VERSION @@ -225,7 +225,7 @@ def test_version_mismatch_continues_and_repairs(): # Personal folder is adopted (same Drive id, new name), not recreated. do_cached_client_new = drive_cached_client_of( - do_manager.peer_manager.connection_router.transport + do_manager.peer_manager.connection_router.connections ) new_personal_name = f"{NEW_VERSION}#{do_email}" new_personal_id = do_cached_client_new.client.find_folder_by_name( @@ -263,7 +263,7 @@ def test_version_mismatch_continues_and_repairs(): ) ds_cached_client_new = drive_cached_client_of( - ds_manager.peer_manager.connection_router.transport + ds_manager.peer_manager.connection_router.connections ) ds_personal_name = f"{NEW_VERSION}#{ds_email}" ds_personal_id = ds_cached_client_new.client.find_folder_by_name( @@ -348,7 +348,7 @@ def test_login_writes_the_remote_version_file_too(): use_in_memory_cache=False, check_versions=True, ) - peering = do_manager.peer_manager.connection_router.transport.peering + peering = do_manager.peer_manager.connection_router.connections.peering_con assert peering.read_own_peer_info().version_info.syft_client_version == SYFT_VERSION with ( @@ -356,7 +356,7 @@ def test_login_writes_the_remote_version_file_too(): patch("syft.sync.version.version_info.SYFT_VERSION", NEW_VERSION), ): upgraded = _upgraded_manager(do_manager) - new_peering = upgraded.peer_manager.connection_router.transport.peering + new_peering = upgraded.peer_manager.connection_router.connections.peering_con # Still the pre-upgrade version: nothing has refreshed it yet. assert ( new_peering.read_own_peer_info().version_info.syft_client_version @@ -407,7 +407,7 @@ def test_the_mismatch_prompt_does_not_return_after_a_login(): handle_potential_version_mismatches_on_login, ) - peering = do_manager.peer_manager.connection_router.transport.peering + peering = do_manager.peer_manager.connection_router.connections.peering_con with patch( "syft.sync.login_utils._read_remote_version", side_effect=lambda e, t: peering.read_own_peer_info().version_info, diff --git a/syft/gdrive_utils.py b/syft/gdrive_utils.py index 0139a2c63c9..047607df1d7 100644 --- a/syft/gdrive_utils.py +++ b/syft/gdrive_utils.py @@ -314,14 +314,14 @@ def delete_remote_syftbox( verbose: If True (default), print deletion progress. """ from syft.sync.connections.drive.drive_cached_client import DriveCachedClient - from syft.sync.connections.drive.drive_owner_store import DriveOwnerStore + from syft.sync.connections.drive.drive_owner_store import DriveOwnerStoreConnection email = _resolve_email(email) token_path = _resolve_token_path(token_path) cached_client = DriveCachedClient.from_token_path( email=email, token_path=token_path ) - store = DriveOwnerStore(cached_client) + store = DriveOwnerStoreConnection(cached_client) # Gather file IDs via folder hierarchy folder_file_ids = set(store.gather_all_file_and_folder_ids()) diff --git a/syft/sync/connections/config.py b/syft/sync/connections/config.py new file mode 100644 index 00000000000..9a92ec1345d --- /dev/null +++ b/syft/sync/connections/config.py @@ -0,0 +1,31 @@ +"""The login each connection of the router uses, as plain config.""" + +from typing import Self + +from pydantic import BaseModel + +from syft.sync.connections.base_connection import ConnectionConfig +from syft.sync.connections.protocols import CONNECTION_FIELDS + + +class ConnectionRouterConfig(BaseModel): + """The login each connection of the router uses. + ``SyftboxManager.from_config`` builds the ``Connections`` from it; only one + Google Drive login for every connection is built so far.""" + + p2p_con: ConnectionConfig + owner_store_con: ConnectionConfig + watcher_store_con: ConnectionConfig + collections_con: ConnectionConfig + peering_con: ConnectionConfig + + @classmethod + def one_connection(cls, connection_config: ConnectionConfig) -> Self: + """Every component over the same connection.""" + return cls(**{name: connection_config for name in CONNECTION_FIELDS}) + + def connection_configs(self) -> list[ConnectionConfig]: + return [getattr(self, name) for name in CONNECTION_FIELDS] + + def uses_one_connection(self) -> bool: + return all(config == self.p2p_con for config in self.connection_configs()) diff --git a/syft/sync/connections/connection_router.py b/syft/sync/connections/connection_router.py index 8a77f684c82..a9fafbc0383 100644 --- a/syft/sync/connections/connection_router.py +++ b/syft/sync/connections/connection_router.py @@ -23,9 +23,9 @@ from syft.sync.peers.exceptions import SyftPeerNotReadyError from syft.sync.peers.peer import Peer, PeerState from syft.sync.peers.peer_store import PeerStore -from syft.sync.transport.naming import PEERS_META_KEY -from syft.sync.transport.protocols import Transport -from syft.sync.transport.types import MessageKind, RawMessage, SendStatus +from syft.sync.connections.naming import PEERS_META_KEY +from syft.sync.connections.protocols import Connections +from syft.sync.connections.types import MessageKind, RawMessage, SendStatus from syft.sync.utils.print_utils import ( print_peer_added_to_platform, print_peer_adding_to_platform, @@ -40,17 +40,17 @@ class ConnectionRouter(BaseModel): """The one router of a manager. ``SyftboxManager.from_config`` builds it - and hands it to every part, so they share one transport, and through it - one connection, and one peer store.""" + and hands it to every part, so they share one set of connections and one + peer store.""" # Built once with the router and never replaced. A test may swap in a - # component that does not use Google Drive, such as a fake p2p. - transport: Transport + # connection that does not use Google Drive, such as a fake p2p. + connections: Connections peer_store: PeerStore def get_authenticated_email(self) -> str: - return self.transport.peering.whoami() + return self.connections.peering_con.whoami() # ========================================================================= # MESSAGE SEND/RECEIVE (with encryption) @@ -100,7 +100,9 @@ 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.transport.p2p.send(recipient, filename, data, MessageKind.PROPOSED_CHANGES) + self.connections.p2p_con.send( + recipient, filename, data, MessageKind.PROPOSED_CHANGES + ) def owner_receive_proposed_messages( self, sender_email: str @@ -117,7 +119,7 @@ def owner_receive_proposed_messages( SyftPeerNotReadyError: when the peer is not set up yet. Every message stays unprocessed. """ - received = self.transport.p2p.receive( + received = self.connections.p2p_con.receive( sender_email, MessageKind.PROPOSED_CHANGES ) readable = [] @@ -140,11 +142,11 @@ def _read_proposed_message( raise except Exception as e: warnings.warn(f"Archived message {raw.msg_id} unread: {e!r}") - self.transport.p2p.mark_processed(raw) + self.connections.p2p_con.mark_processed(raw) return None def owner_mark_all_processed(self, messages: list[RawMessage]) -> None: - self.transport.p2p.mark_all_processed(messages) + self.connections.p2p_con.mark_all_processed(messages) def owner_write_event_messages_to_outbox( self, recipient_email: str, events_message: FileChangeEventsMessage @@ -153,7 +155,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() - return self.transport.p2p.send( + return self.connections.p2p_con.send( recipient_email, fname, data, MessageKind.FILE_EVENTS ) @@ -161,7 +163,7 @@ def watcher_get_events_messages( self, peer_email: str, since_timestamp: float | None ) -> List[FileChangeEventsMessage]: """The outbox messages of ``peer_email``, without the skipped ones.""" - raw_list = self.transport.p2p.outbox_messages(peer_email, since_timestamp) + raw_list = self.connections.p2p_con.outbox_messages(peer_email, since_timestamp) messages = [ self.watcher_read_events_message(peer_email, raw) for raw in raw_list ] @@ -188,18 +190,20 @@ def watcher_read_events_message( def watcher_get_outbox_file_metadatas( self, peer_email: str, since_timestamp: float | None ) -> List[dict]: - return self.transport.p2p.outbox_message_metadatas(peer_email, since_timestamp) + return self.connections.p2p_con.outbox_message_metadatas( + peer_email, since_timestamp + ) def download_outbox_message(self, file_id: str) -> bytes: """Download one outbox message by its handle (thread-safe).""" - return self.transport.p2p.clone_for_thread().fetch_message(file_id) + return self.connections.p2p_con.clone_for_thread().fetch_message(file_id) def owner_get_outbox_file_metadatas(self, recipient_email: str) -> List[dict]: """The messages in my outbox to ``recipient_email``, without downloading.""" - return self.transport.p2p.own_outbox_message_metadatas(recipient_email) + return self.connections.p2p_con.own_outbox_message_metadatas(recipient_email) def owner_delete_outbox_messages(self, file_ids: List[str]) -> None: - self.transport.p2p.delete_messages(file_ids) + self.connections.p2p_con.delete_messages(file_ids) # ========================================================================= # EVENT LOG (self-encrypted at rest when encryption enabled) @@ -211,13 +215,15 @@ 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.transport.owner_store.append_event(filename, data) + self.connections.owner_store_con.append_event(filename, data) def download_events_message_by_id( self, events_message_id: str ) -> FileChangeEventsMessage: """Download and decrypt a single events message (thread-safe).""" - raw = self.transport.owner_store.clone_for_thread().fetch(events_message_id) + raw = self.connections.owner_store_con.clone_for_thread().fetch( + events_message_id + ) raw = self.peer_store.decrypt_and_verify_for_self_if_needed(raw) return FileChangeEventsMessage.from_compressed_data(raw) @@ -226,12 +232,12 @@ def download_events_message_by_id( # ========================================================================= def add_peer(self, peer_email: str, verbose: bool = True) -> Peer: - platform = self.transport.platform + platform = self.connections.platform if verbose: print_peer_adding_to_platform(peer_email, platform.module_path) - self.transport.peering.open_channels(peer_email) + self.connections.peering_con.open_channels(peer_email) if verbose: print_peer_added_to_platform(peer_email, platform.module_path) @@ -239,7 +245,7 @@ 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.""" - peers_data = self.transport.peering.read_peer_registry( + peers_data = self.connections.peering_con.read_peer_registry( force_download=force_download ) peers = [] @@ -260,7 +266,7 @@ def get_all_peers_from_json(self, force_download: bool = False) -> List[Peer]: continue peer = Peer( email=email, - platforms=[self.transport.platform], + platforms=[self.connections.platform], state=state, public_encryption_bundle=data.get("public_encryption_bundle"), ) @@ -269,11 +275,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""" - peer_emails = self.transport.peering.incoming_requests() + peer_emails = self.connections.peering_con.incoming_requests() return [ Peer( email=peer_email, - platforms=[self.transport.platform], + platforms=[self.connections.platform], state=PeerState.REQUESTED_BY_PEER, ) for peer_email in peer_emails @@ -286,7 +292,7 @@ def update_peer_state( public_encryption_bundle: dict | None = None, ): """Update peer state in storage""" - self.transport.peering.update_peer_registry( + self.connections.peering_con.update_peer_registry( peer_email, state, public_encryption_bundle ) @@ -296,15 +302,15 @@ def update_peer_state( def write_encryption_bundle(self, peer_email: str, bundle_json: str) -> None: """Write own encryption bundle for a peer.""" - self.transport.peering.publish_key_bundle(peer_email, bundle_json) + self.connections.peering_con.publish_key_bundle(peer_email, bundle_json) def share_encryption_bundles_folder(self, peer_email: str) -> None: """Share the bundles folder with a peer.""" - self.transport.peering.grant_key_bundle_read(peer_email) + self.connections.peering_con.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.""" - return self.transport.peering.read_peer_key_bundle(peer_email) + return self.connections.peering_con.read_peer_key_bundle(peer_email) # ========================================================================= # GENERIC PREFIX-PARAMETERIZED COLLECTION METHODS (with encryption) @@ -313,17 +319,19 @@ 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: - return self.transport.collections.create(prefix, tag, content_hash, owner_email) + return self.connections.collections_con.create( + prefix, tag, content_hash, owner_email + ) def owner_tag_collection_as_any( self, prefix: str, tag: str, content_hash: str ) -> None: - self.transport.collections.mark_shared_with_any(prefix, tag, content_hash) + self.connections.collections_con.mark_shared_with_any(prefix, tag, content_hash) def owner_share_collection( self, prefix: str, tag: str, content_hash: str, users: list[str] ) -> None: - self.transport.collections.grant_read(prefix, tag, content_hash, users) + self.connections.collections_con.grant_read(prefix, tag, content_hash, users) def owner_upload_collection_files( self, @@ -339,26 +347,26 @@ def owner_upload_collection_files( name: self.peer_store.encrypt_if_needed(recipient_email, data) for name, data in files.items() } - self.transport.collections.upload(prefix, tag, content_hash, files) + self.connections.collections_con.upload(prefix, tag, content_hash, files) def owner_list_collections(self, prefix: str) -> list[str]: - return self.transport.collections.list_mine(prefix) + return self.connections.collections_con.list_mine(prefix) def owner_list_all_collections_with_permissions( self, prefix: str ) -> list[FileCollection]: - return self.transport.collections.list_mine_with_permissions(prefix) + return self.connections.collections_con.list_mine_with_permissions(prefix) def owner_delete_collection(self, prefix: str, tag: str) -> None: - self.transport.collections.delete(prefix, tag) + self.connections.collections_con.delete(prefix, tag) def watcher_list_collections(self, prefix: str) -> list[dict]: - return self.transport.collections.list_shared_with_me(prefix) + return self.connections.collections_con.list_shared_with_me(prefix) def watcher_download_collection( self, prefix: str, tag: str, content_hash: str, owner_email: str ) -> dict[str, bytes]: - files = self.transport.collections.download( + files = self.connections.collections_con.download( prefix, tag, content_hash, owner_email ) if self.peer_store and owner_email: @@ -370,31 +378,31 @@ def watcher_download_collection( def write_peer_info_file(self, peer_info: "PeerInfo") -> None: """Write the peer info file to own SyftBox folder.""" - self.transport.peering.publish_peer_info(peer_info) + self.connections.peering_con.publish_peer_info(peer_info) def read_own_peer_info_file(self) -> Optional["PeerInfo"]: """Read the peer info file from own SyftBox folder.""" - return self.transport.peering.read_own_peer_info() + return self.connections.peering_con.read_own_peer_info() def read_peer_info_file(self, peer_email: str) -> Optional["PeerInfo"]: """Read the peer info file from a peer's SyftBox folder.""" - return self.transport.peering.read_peer_info(peer_email) + return self.connections.peering_con.read_peer_info(peer_email) def share_peer_info_file_with_peer( self, peer_email: str, peer_info: "PeerInfo" ) -> None: """Share the peer info file with a peer so they can read it.""" - self.transport.peering.grant_peer_info_read(peer_email, peer_info) + self.connections.peering_con.grant_peer_info_read(peer_email, peer_info) def watcher_get_collection_file_metadatas( self, prefix: str, tag: str, content_hash: str, owner_email: str ) -> List[dict]: - return self.transport.collections.list_files( + return self.connections.collections_con.list_files( prefix, tag, content_hash, owner_email ) def watcher_download_collection_file(self, file_id: str, owner_email: str) -> bytes: - data = self.transport.collections.fetch_file(file_id) + data = self.connections.collections_con.fetch_file(file_id) data = self.peer_store.decrypt_dataset_if_needed(owner_email, data) return data @@ -405,19 +413,19 @@ def watcher_download_collection_file(self, file_id: str, owner_email: str) -> by def owner_get_all_accepted_event_file_ids( self, since_timestamp: float | None = None ) -> List[str]: - return self.transport.owner_store.list_event_handles(since_timestamp) + return self.connections.owner_store_con.list_event_handles(since_timestamp) def delete_all(self) -> int: """Delete everything this account stores; return how many files and folders went.""" - return self.transport.owner_store.delete_all() + return self.connections.owner_store_con.delete_all() def reset_caches(self): - for component in self.transport.components(): - component.reset_caches() + for connection in self.connections.all(): + connection.reset_caches() def delete_file_by_id(self, file_id: str): - self.transport.collections.delete_by_handle(file_id) + self.connections.collections_con.delete_by_handle(file_id) # ========================================================================= # CHECKPOINT METHODS @@ -427,11 +435,13 @@ 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) - return self.transport.owner_store.put_checkpoint(checkpoint.filename, data) + return self.connections.owner_store_con.put_checkpoint( + checkpoint.filename, data + ) def get_latest_checkpoint(self) -> Checkpoint | None: """Download, decrypt, and deserialize the latest checkpoint.""" - raw = self.transport.owner_store.latest_checkpoint() + raw = self.connections.owner_store_con.latest_checkpoint() if raw is None: return None try: @@ -445,13 +455,13 @@ def get_events_count_since_checkpoint( self, checkpoint_timestamp: float | None ) -> int: """Count events created after the checkpoint timestamp.""" - return self.transport.owner_store.count_events_since(checkpoint_timestamp) + return self.connections.owner_store_con.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.""" - raw_list = self.transport.owner_store.fetch_events_since(since_timestamp) + raw_list = self.connections.owner_store_con.fetch_events_since(since_timestamp) return [ FileChangeEventsMessage.from_compressed_data( self.peer_store.decrypt_and_verify_for_self_if_needed(raw) @@ -467,11 +477,13 @@ 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) - return self.transport.owner_store.put_incremental(checkpoint.filename, data) + return self.connections.owner_store_con.put_incremental( + checkpoint.filename, data + ) def get_all_incremental_checkpoints(self) -> List[IncrementalCheckpoint]: """Download, decrypt, and deserialize all incremental checkpoints.""" - raw_list = self.transport.owner_store.all_incrementals() + raw_list = self.connections.owner_store_con.all_incrementals() checkpoints = [] for raw in raw_list: try: @@ -485,15 +497,15 @@ def get_all_incremental_checkpoints(self) -> List[IncrementalCheckpoint]: def get_incremental_checkpoint_count(self) -> int: """Get the number of incremental checkpoints.""" - return self.transport.owner_store.incremental_count() + return self.connections.owner_store_con.incremental_count() def get_next_incremental_sequence_number(self) -> int: """Get the next sequence number for incremental checkpoints.""" - return self.transport.owner_store.next_incremental_sequence() + return self.connections.owner_store_con.next_incremental_sequence() def delete_all_incremental_checkpoints(self) -> None: """Delete all incremental checkpoints from the storage backend.""" - self.transport.owner_store.delete_incrementals() + self.connections.owner_store_con.delete_incrementals() # ========================================================================= # ROLLING STATE METHODS @@ -503,13 +515,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) - return self.transport.owner_store.put_rolling_state( + return self.connections.owner_store_con.put_rolling_state( rolling_state.filename, data ) def get_rolling_state(self) -> RollingState | None: """Download, decrypt, and deserialize rolling state.""" - raw = self.transport.owner_store.get_rolling_state() + raw = self.connections.owner_store_con.get_rolling_state() if raw is None: return None try: @@ -521,4 +533,4 @@ def get_rolling_state(self) -> RollingState | None: def delete_rolling_state(self) -> None: """Delete rolling state from the storage backend.""" - self.transport.owner_store.delete_rolling_state() + self.connections.owner_store_con.delete_rolling_state() diff --git a/syft/sync/connections/drive/drive_cached_client.py b/syft/sync/connections/drive/drive_cached_client.py index af1e291d31f..f0183d2bd88 100644 --- a/syft/sync/connections/drive/drive_cached_client.py +++ b/syft/sync/connections/drive/drive_cached_client.py @@ -22,7 +22,7 @@ ) from syft.sync.connections.drive.gdrive_retry import execute_with_retries from syft.sync.events.file_change_event import FileChangeEventsMessageFileName -from syft.sync.transport.naming import timestamp_from_message_name +from syft.sync.connections.naming import 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 @@ -192,7 +192,7 @@ class DriveCachedClient: It logs in, finds and creates syft's folders, follows the version rules for a versioned folder name, and caches the folder ids so a lookup costs no - Drive call the second time. The transport components share one. + Drive call the second time. The router's connections share one. """ client: GDriveClient diff --git a/syft/sync/connections/drive/drive_collections.py b/syft/sync/connections/drive/drive_collections.py index 2919b2f00ef..be7cc01f875 100644 --- a/syft/sync/connections/drive/drive_collections.py +++ b/syft/sync/connections/drive/drive_collections.py @@ -14,7 +14,7 @@ from syft.sync.connections.drive.gdrive_client import GOOGLE_FOLDER_MIME_TYPE from syft.sync.connections.drive.drive_cached_client import DriveCachedClient from syft.sync.connections.drive.gdrive_retry import execute_with_retries -from syft.sync.transport.protocols import Collections +from syft.sync.connections.protocols import CollectionsConnection def collection_name_query(prefix: str) -> str: @@ -85,7 +85,7 @@ def compute_hash(files: dict[str, bytes]) -> str: return compute_file_hashes(files) -class DriveCollections(Collections): +class DriveCollectionsConnection(CollectionsConnection): """Collection folders in /SyftBox, shared by Drive permission.""" def __init__(self, cached_client: DriveCachedClient): @@ -93,10 +93,10 @@ def __init__(self, cached_client: DriveCachedClient): # folder name -> id of my collection folder self.collection_folder_id_cache: dict[str, str] = {} - def clone_for_thread(self) -> "DriveCollections": + def clone_for_thread(self) -> "DriveCollectionsConnection": """Collections over a new Drive client, for use from another thread, with these caches.""" - clone = DriveCollections(self._cached_client.clone_for_thread()) + clone = DriveCollectionsConnection(self._cached_client.clone_for_thread()) clone.collection_folder_id_cache = dict(self.collection_folder_id_cache) return clone diff --git a/syft/sync/connections/drive/drive_connections.py b/syft/sync/connections/drive/drive_connections.py index 93ec9b1051e..12b785b96e8 100644 --- a/syft/sync/connections/drive/drive_connections.py +++ b/syft/sync/connections/drive/drive_connections.py @@ -1,16 +1,16 @@ -"""The transport components on Google Drive. +"""The router's connections on Google Drive. -Each component has its own module: ``drive_p2p``, ``drive_owner_store``, +Each connection has its own module: ``drive_p2p``, ``drive_owner_store``, ``drive_collections`` and ``drive_peering``. They share one ``DriveCachedClient``, -which holds the Drive client. ``SyftboxManager._drive_transport`` builds them. +which holds the Drive client. ``SyftboxManager._drive_connections`` builds them. """ -from syft.sync.connections.drive.drive_owner_store import DriveOwnerStore +from syft.sync.connections.drive.drive_owner_store import DriveOwnerStoreConnection from syft.sync.connections.drive.drive_cached_client import DriveCachedClient -from syft.sync.transport.protocols import Transport, WatcherStore +from syft.sync.connections.protocols import Connections, WatcherStoreConnection -class DriveWatcherStore(WatcherStore): +class DriveWatcherStoreConnection(WatcherStoreConnection): """The DS's storage on Drive. Not built yet: today the DS keeps what it received in a local cache.""" @@ -30,11 +30,11 @@ def reset(self, peer: str) -> None: raise NotImplementedError("Watcher storage on Drive is not built yet") -def drive_cached_client_of(transport: Transport) -> DriveCachedClient: - """The ``DriveCachedClient`` under ``transport``, for tests and tools that +def drive_cached_client_of(connections: Connections) -> DriveCachedClient: + """The ``DriveCachedClient`` under ``connections``, for tests and tools that need Drive itself. The sync code reaches Drive only through the components. - Raises if the transport is not built on Drive.""" - store = transport.owner_store - if not isinstance(store, DriveOwnerStore): - raise TypeError("This transport is not built on Google Drive") + Raises if the connections are not built on Drive.""" + store = connections.owner_store_con + if not isinstance(store, DriveOwnerStoreConnection): + raise TypeError("These connections are not built on Google Drive") return store._cached_client diff --git a/syft/sync/connections/drive/drive_owner_store.py b/syft/sync/connections/drive/drive_owner_store.py index b0e20170951..8f8c779e006 100644 --- a/syft/sync/connections/drive/drive_owner_store.py +++ b/syft/sync/connections/drive/drive_owner_store.py @@ -31,7 +31,7 @@ gather_all_file_and_folder_ids_recursive, ) from syft.sync.events.file_change_event import FileChangeEventsMessageFileName -from syft.sync.transport.protocols import OwnerStore +from syft.sync.connections.protocols import OwnerStoreConnection from syft.version import SYFT_VERSION @@ -51,7 +51,7 @@ def _filter_valid_file_metadatas( return res -class DriveOwnerStore(OwnerStore): +class DriveOwnerStoreConnection(OwnerStoreConnection): """The DO's own folders in /SyftBox: event log, checkpoints, rolling state.""" def __init__(self, cached_client: DriveCachedClient): @@ -62,10 +62,10 @@ def __init__(self, cached_client: DriveCachedClient): # fname -> gdrive id self.personal_syftbox_event_id_cache: dict[str, str] = {} - def clone_for_thread(self) -> "DriveOwnerStore": + def clone_for_thread(self) -> "DriveOwnerStoreConnection": """A store over a new Drive client, for use from another thread, with this store's caches.""" - clone = DriveOwnerStore(self._cached_client.clone_for_thread()) + clone = DriveOwnerStoreConnection(self._cached_client.clone_for_thread()) clone._rolling_state_folder_id = self._rolling_state_folder_id clone._rolling_state_file_id = self._rolling_state_file_id clone.personal_syftbox_event_id_cache = dict( diff --git a/syft/sync/connections/drive/drive_p2p.py b/syft/sync/connections/drive/drive_p2p.py index 9132e94cbbc..a62c8c1da61 100644 --- a/syft/sync/connections/drive/drive_p2p.py +++ b/syft/sync/connections/drive/drive_p2p.py @@ -15,9 +15,9 @@ raise_unavailable, ) from syft.sync.messages.proposed_filechange import FileNameParseError, MessageFileName -from syft.sync.transport.exceptions import TransportUnavailableError -from syft.sync.transport.protocols import P2PConnection -from syft.sync.transport.types import MessageKind, RawMessage, SendStatus +from syft.sync.connections.exceptions import ConnectionUnavailableError +from syft.sync.connections.protocols import P2PConnection +from syft.sync.connections.types import MessageKind, RawMessage, SendStatus # Threads that download the messages of one receive at the same time. DOWNLOAD_WORKERS = 10 @@ -28,13 +28,13 @@ def _download_or_skip(client: GDriveClient, listed: dict) -> bytes | None: file cannot be downloaded, for example because it was deleted. Raises: - TransportUnavailableError: when Drive cannot be reached. + ConnectionUnavailableError: when Drive cannot be reached. """ name = listed["file_name"] try: with raise_unavailable(f"Drive is unreachable; inbox message {name} waits"): return client.download_file(listed["file_id"]) - except TransportUnavailableError: + except ConnectionUnavailableError: raise except Exception as e: warnings.warn(f"Skipped inbox message {name}: download failed: {e!r}") @@ -108,7 +108,7 @@ def receive( not hold back the others. Raises: - TransportUnavailableError: when Drive cannot be reached. Every + ConnectionUnavailableError: when Drive cannot be reached. Every message stays in the inbox. """ _require_inbox_kind(kind) diff --git a/syft/sync/connections/drive/drive_peering.py b/syft/sync/connections/drive/drive_peering.py index 7884de0f838..0848bcf66bf 100644 --- a/syft/sync/connections/drive/drive_peering.py +++ b/syft/sync/connections/drive/drive_peering.py @@ -22,8 +22,8 @@ execute_with_retries, raise_unavailable, ) -from syft.sync.transport.naming import PEERS_META_KEY, SYFT_PEERS_VERSION -from syft.sync.transport.protocols import Peering +from syft.sync.connections.naming import PEERS_META_KEY, SYFT_PEERS_VERSION +from syft.sync.connections.protocols import PeeringConnection if TYPE_CHECKING: from syft.sync.version.peer_info import PeerInfo @@ -43,7 +43,7 @@ def as_string(self) -> str: return f"syft_encryption_bundles#{self.email}" -class DrivePeering(Peering): +class DrivePeeringConnection(PeeringConnection): """Identity from the Google account; channels, the peer list, key bundles and version files in /SyftBox.""" @@ -55,10 +55,10 @@ def __init__(self, cached_client: DriveCachedClient): # plus "_meta" -> {"version"}. None until the first read. self._peers_json_cache: dict[str, dict[str, Any]] | None = None - def clone_for_thread(self) -> "DrivePeering": + def clone_for_thread(self) -> "DrivePeeringConnection": """Peering over a new Drive client, for use from another thread, with these caches.""" - clone = DrivePeering(self._cached_client.clone_for_thread()) + clone = DrivePeeringConnection(self._cached_client.clone_for_thread()) clone._encryption_bundles_folder_id = self._encryption_bundles_folder_id clone._peers_json_cache = ( dict(self._peers_json_cache) if self._peers_json_cache is not None else None @@ -117,7 +117,7 @@ def _download_peers_json(self) -> dict[str, dict[str, Any]]: dict}``, and ``_meta`` to ``{"version": int}``. Raises: - TransportUnavailableError: when Drive cannot be reached. + ConnectionUnavailableError: when Drive cannot be reached. ValueError: when the file is not valid JSON. """ with raise_unavailable("Could not read the peers file from Drive."): @@ -295,7 +295,7 @@ def read_own_peer_info(self) -> Optional["PeerInfo"]: Returns None when there is no file. Raises: - TransportUnavailableError: when Drive cannot be reached. + ConnectionUnavailableError: when Drive cannot be reached. ValueError: when the file is not a peer info file this client reads. """ return self._read_peer_info_file( @@ -308,7 +308,7 @@ def read_peer_info(self, peer: str) -> Optional["PeerInfo"]: Returns None when the peer has no file that this account may read. Raises: - TransportUnavailableError: when Drive cannot be reached. + ConnectionUnavailableError: when Drive cannot be reached. ValueError: when the file is not a peer info file this client reads. """ return self._read_peer_info_file( diff --git a/syft/sync/connections/drive/gdrive_client.py b/syft/sync/connections/drive/gdrive_client.py index 46db30bd4aa..9d9663ccda7 100644 --- a/syft/sync/connections/drive/gdrive_client.py +++ b/syft/sync/connections/drive/gdrive_client.py @@ -1,4 +1,4 @@ -"""The Google Drive API client that the Drive transport jobs share. +"""The Google Drive API client that the Drive connections share. One client per account. It holds the Drive service, which is not safe to use from more than one thread. diff --git a/syft/sync/connections/drive/gdrive_retry.py b/syft/sync/connections/drive/gdrive_retry.py index e861994eca3..ebb7ed83392 100644 --- a/syft/sync/connections/drive/gdrive_retry.py +++ b/syft/sync/connections/drive/gdrive_retry.py @@ -14,7 +14,7 @@ from googleapiclient.errors import HttpError from googleapiclient.http import MediaIoBaseDownload -from syft.sync.transport.exceptions import TransportUnavailableError +from syft.sync.connections.exceptions import ConnectionUnavailableError logger = logging.getLogger(__name__) @@ -74,7 +74,7 @@ def is_retryable_error(error: Exception) -> bool: @contextmanager def raise_unavailable(message: str) -> Iterator[None]: - """Raise TransportUnavailableError(message) when Drive cannot be reached. + """Raise ConnectionUnavailableError(message) when Drive cannot be reached. Drive cannot be reached when the error is in UNREACHABLE_ERRORS, or when a transient error outlasts the retries. Any other error, such as a 403 or a @@ -84,7 +84,7 @@ def raise_unavailable(message: str) -> Iterator[None]: yield except (HttpError, *RETRYABLE_TRANSPORT_ERRORS, *UNREACHABLE_ERRORS) as e: if isinstance(e, UNREACHABLE_ERRORS) or is_retryable_error(e): - raise TransportUnavailableError(message) from e + raise ConnectionUnavailableError(message) from e raise diff --git a/syft/sync/connections/exceptions.py b/syft/sync/connections/exceptions.py new file mode 100644 index 00000000000..d2cb0becab8 --- /dev/null +++ b/syft/sync/connections/exceptions.py @@ -0,0 +1,9 @@ +"""Errors raised by the connections of every backend.""" + + +class SyftConnectionError(Exception): + """Base exception for connection errors.""" + + +class ConnectionUnavailableError(SyftConnectionError): + """The backend could not be reached.""" diff --git a/syft/sync/transport/naming.py b/syft/sync/connections/naming.py similarity index 100% rename from syft/sync/transport/naming.py rename to syft/sync/connections/naming.py diff --git a/syft/sync/transport/protocols.py b/syft/sync/connections/protocols.py similarity index 87% rename from syft/sync/transport/protocols.py rename to syft/sync/connections/protocols.py index fa67ae4602d..1de8510c994 100644 --- a/syft/sync/transport/protocols.py +++ b/syft/sync/connections/protocols.py @@ -1,10 +1,11 @@ -"""The jobs a transport backend does, one abstract class each. +"""The connections a backend provides, one abstract class for each job. -``P2PConnection`` carries messages between peers, ``OwnerStore`` and -``WatcherStore`` hold each role's own storage, ``Collections`` shares datasets, -and ``Peering`` finds and trusts peers. A backend implements the classes it -can. The sync layer above never sees backend ids: it passes back the messages -and handles a backend returns. +``P2PConnection`` carries messages between peers, ``OwnerStoreConnection`` and +``WatcherStoreConnection`` hold each role's own storage, +``CollectionsConnection`` shares datasets, and ``PeeringConnection`` finds and +trusts peers. A backend implements the classes it can. The sync layer above +never sees backend ids: it passes back the messages and handles a backend +returns. Each class keeps its abstract methods to the few a backend must write. Methods that only combine those are ordinary methods on the base class, which a @@ -19,15 +20,15 @@ from pydantic import BaseModel, ConfigDict from syft.sync.platforms.base_platform import BasePlatform -from syft.sync.transport.types import MessageKind, RawMessage, SendStatus +from syft.sync.connections.types import MessageKind, RawMessage, SendStatus if TYPE_CHECKING: from syft.sync.connections.base_connection import FileCollection from syft.sync.version.peer_info import PeerInfo -class TransportComponent: - """What every component of a transport does besides its own work.""" +class RouterConnection: + """What every connection of the router does besides its own work.""" def clone_for_thread(self) -> Self: """A copy that another thread may use. This one, unless its @@ -38,7 +39,7 @@ def reset_caches(self) -> None: """Forget what was cached from the connection.""" -class P2PConnection(TransportComponent, ABC): +class P2PConnection(RouterConnection, ABC): """Messages between peers, kept until the recipient has processed them. Sending and receiving each have their own confirmation: @@ -91,7 +92,7 @@ def mark_all_processed(self, messages: list[RawMessage]) -> None: self.mark_processed(message) -class OwnerStore(TransportComponent, ABC): +class OwnerStoreConnection(RouterConnection, ABC): """The data owner's own storage: event log, checkpoints, rolling state. Only the owner writes and reads it, and it is never shared. @@ -172,7 +173,7 @@ def delete_all(self) -> int: how many files and folders went.""" -class WatcherStore(TransportComponent, ABC): +class WatcherStoreConnection(RouterConnection, ABC): """The data scientist's memory of what each owner sent it. One area per owner, in the scientist's own storage, never shared, so it @@ -203,7 +204,7 @@ def reset(self, peer: str) -> None: everything again.""" -class Collections(TransportComponent, ABC): +class CollectionsConnection(RouterConnection, ABC): """Sets of files a data owner shares with named readers (datasets).""" @abstractmethod @@ -266,7 +267,7 @@ def download( return {f["file_name"]: self.fetch_file(f["file_id"]) for f in files} -class Peering(TransportComponent, ABC): +class PeeringConnection(RouterConnection, ABC): """Finding and trusting a peer: identity, channels, peer list, published documents.""" @@ -320,7 +321,7 @@ def read_own_peer_info(self) -> PeerInfo | None: """My published version document, or None. Raises: - TransportUnavailableError: when the backend cannot be reached. + ConnectionUnavailableError: when the backend cannot be reached. ValueError: when the document is not one this client reads. """ @@ -329,7 +330,7 @@ def read_peer_info(self, peer: str) -> PeerInfo | None: """``peer``'s published version document, or None. Raises: - TransportUnavailableError: when the backend cannot be reached. + ConnectionUnavailableError: when the backend cannot be reached. ValueError: when the document is not one this client reads. """ @@ -341,30 +342,30 @@ def grant_peer_info_read(self, peer: str, peer_info: PeerInfo) -> None: """ -# The components of a transport, as named on ``Transport``. -TRANSPORT_COMPONENTS = ( - "p2p", - "owner_store", - "watcher_store", - "collections", - "peering", +# The fields of ``Connections``, one per connection. +CONNECTION_FIELDS = ( + "p2p_con", + "owner_store_con", + "watcher_store_con", + "collections_con", + "peering_con", ) -class Transport(BaseModel): - """One implementation of each job. The parts may come from different - backends, though a peer's ``p2p`` and ``peering`` go together.""" +class Connections(BaseModel): + """One connection for each job. The connections may come from different + backends, though a peer's ``p2p_con`` and ``peering_con`` go together.""" model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) - p2p: P2PConnection - owner_store: OwnerStore - watcher_store: WatcherStore - collections: Collections - peering: Peering + p2p_con: P2PConnection + owner_store_con: OwnerStoreConnection + watcher_store_con: WatcherStoreConnection + collections_con: CollectionsConnection + peering_con: PeeringConnection # Recorded on each Peer so a peer list says which backend reaches it. platform: BasePlatform - def components(self) -> list[TransportComponent]: - """The components, in ``TRANSPORT_COMPONENTS`` order.""" - return [getattr(self, name) for name in TRANSPORT_COMPONENTS] + def all(self) -> list[RouterConnection]: + """The connections, in ``CONNECTION_FIELDS`` order.""" + return [getattr(self, name) for name in CONNECTION_FIELDS] diff --git a/syft/sync/transport/types.py b/syft/sync/connections/types.py similarity index 95% rename from syft/sync/transport/types.py rename to syft/sync/connections/types.py index a3f17526867..89b92d3f656 100644 --- a/syft/sync/transport/types.py +++ b/syft/sync/connections/types.py @@ -1,4 +1,4 @@ -"""Types shared by every transport backend.""" +"""Types shared by the connections of every backend.""" from enum import Enum diff --git a/syft/sync/login_utils.py b/syft/sync/login_utils.py index e3f5d028e81..b1de8753e54 100644 --- a/syft/sync/login_utils.py +++ b/syft/sync/login_utils.py @@ -20,12 +20,12 @@ def _read_remote_version( ) -> Optional[VersionInfo]: """Read version file from GDrive without a full SyftboxManager.""" from syft.sync.connections.drive.drive_cached_client import DriveCachedClient - from syft.sync.connections.drive.drive_peering import DrivePeering + from syft.sync.connections.drive.drive_peering import DrivePeeringConnection cached_client = DriveCachedClient.from_token_path( email=email, token_path=token_path ) - peer_info = DrivePeering(cached_client).read_own_peer_info() + peer_info = DrivePeeringConnection(cached_client).read_own_peer_info() return peer_info.version_info if peer_info else None diff --git a/syft/sync/messages/raw_message.py b/syft/sync/messages/raw_message.py index 6ce9a329196..4597cb2bff0 100644 --- a/syft/sync/messages/raw_message.py +++ b/syft/sync/messages/raw_message.py @@ -1,5 +1,5 @@ -"""``RawMessage`` lives with the other transport types; this path keeps working.""" +"""``RawMessage`` lives with the other connection types; this path keeps working.""" -from syft.sync.transport.types import RawMessage +from syft.sync.connections.types import RawMessage __all__ = ["RawMessage"] diff --git a/syft/sync/syftbox_manager.py b/syft/sync/syftbox_manager.py index 60f4847eeb4..a0322bbc11b 100644 --- a/syft/sync/syftbox_manager.py +++ b/syft/sync/syftbox_manager.py @@ -14,14 +14,14 @@ from syft.sync.connections.connection_router import ConnectionRouter from syft.sync.connections.drive import mock_drive_service from syft.sync.connections.drive.drive_cached_client import DriveCachedClient -from syft.sync.connections.drive.drive_collections import DriveCollections +from syft.sync.connections.drive.drive_collections import DriveCollectionsConnection from syft.sync.connections.drive.drive_connections import ( - DriveWatcherStore, + DriveWatcherStoreConnection, drive_cached_client_of, ) -from syft.sync.connections.drive.drive_owner_store import DriveOwnerStore +from syft.sync.connections.drive.drive_owner_store import DriveOwnerStoreConnection from syft.sync.connections.drive.drive_p2p import DriveP2PConnection -from syft.sync.connections.drive.drive_peering import DrivePeering +from syft.sync.connections.drive.drive_peering import DrivePeeringConnection from syft.sync.connections.drive.grdrive_config import GdriveConnectionConfig from syft.sync.events.file_change_event import ( FileChangeEvent, @@ -60,9 +60,9 @@ random_syftbox_folder_for_testing, ) from syft.sync.peers.peer_store import PeerStore -from syft.sync.transport.config import TransportConfig +from syft.sync.connections.config import ConnectionRouterConfig from syft.sync.platforms.gdrive_files_platform import GdriveFilesPlatform -from syft.sync.transport.protocols import Transport +from syft.sync.connections.protocols import Connections from syft.sync.version.peer_manager import ( PeerManager, PeerManagerConfig, @@ -131,11 +131,11 @@ class SyftboxManagerConfig(BaseModel): # log, rolling state, checkpoints. Off for datasites that can never # restore (see syft-enclave's EnclaveSettings). persist_owner_state: bool = True - # The connection each transport component uses. Only the manager reads - # it: it builds one transport from it and shares that with every part. - # None when the caller passes a transport to SyftboxManager.from_config + # The login each connection of the router uses. Only the manager reads + # it: it builds the connections from it and shares them with every part. + # None when the caller passes connections to SyftboxManager.from_config # instead, as tests do with a mock Drive. - transport_config: TransportConfig | None = None + connection_router_config: ConnectionRouterConfig | None = None datasite_owner_syncer_config: DatasiteOwnerSyncerConfig peer_manager_config: PeerManagerConfig @@ -170,7 +170,7 @@ def for_colab( syftbox_folder, email, collection_specs ) collection_variants = default_collection_variants(collection_specs) - transport_config = TransportConfig.one_connection( + connection_router_config = ConnectionRouterConfig.one_connection( GdriveConnectionConfig(email=email, token_path=None) ) datasite_owner_syncer_config = DatasiteOwnerSyncerConfig( @@ -209,7 +209,7 @@ def for_colab( return cls( email=email, syftbox_folder=syftbox_folder, - transport_config=transport_config, + connection_router_config=connection_router_config, has_ds_role=has_ds_role, has_do_role=has_do_role, use_in_memory_cache=False, @@ -249,7 +249,7 @@ def for_jupyter( ) collection_variants = default_collection_variants(collection_specs) - transport_config = TransportConfig.one_connection( + connection_router_config = ConnectionRouterConfig.one_connection( GdriveConnectionConfig(email=email, token_path=token_path) ) datasite_owner_syncer_config = DatasiteOwnerSyncerConfig( @@ -288,7 +288,7 @@ def for_jupyter( return cls( email=email, syftbox_folder=syftbox_folder, - transport_config=transport_config, + connection_router_config=connection_router_config, has_ds_role=has_ds_role, has_do_role=has_do_role, use_in_memory_cache=False, @@ -400,7 +400,7 @@ def for_google_drive_testing_connection( syftbox_folder, email, collection_specs ) collection_variants = default_collection_variants(collection_specs) - transport_config = TransportConfig.one_connection( + connection_router_config = ConnectionRouterConfig.one_connection( GdriveConnectionConfig(email=email, token_path=token_path) ) datasite_owner_syncer_config = DatasiteOwnerSyncerConfig( @@ -438,7 +438,7 @@ def for_google_drive_testing_connection( return cls( email=email, syftbox_folder=syftbox_folder, - transport_config=transport_config, + connection_router_config=connection_router_config, write_files=write_files, persist_owner_state=persist_owner_state, datasite_owner_syncer_config=datasite_owner_syncer_config, @@ -529,7 +529,7 @@ def peers(self) -> PeerList: def from_config( cls, config: SyftboxManagerConfig, - transport: Transport | None = None, + connections: Connections | None = None, ): """Build a manager and its parts from ``config``. @@ -537,17 +537,19 @@ def from_config( objects or state. A live object that several parts share is built here, once, and passed to each part's ``from_config`` next to its config: the peer store and the connection router that holds it, and through - the router one transport, and its connection, for the whole manager. + the router one set of connections for the whole manager. - ``transport`` replaces the one ``config`` describes. + ``connections`` replace the ones ``config`` describes. Tests pass one over a mock Drive, some with a fake p2p. """ peer_store = PeerManager.peer_store_from_config( config.peer_manager_config, email=config.email ) - if transport is None: - transport = cls._transport_from_config(config) - connection_router = ConnectionRouter(transport=transport, peer_store=peer_store) + if connections is None: + connections = cls._connections_from_config(config) + connection_router = ConnectionRouter( + connections=connections, peer_store=peer_store + ) file_writer = FileWriter( base_path=config.syftbox_folder, write_files=config.write_files ) @@ -587,34 +589,33 @@ def from_config( return manager_res @staticmethod - def _transport_from_config(config: SyftboxManagerConfig) -> Transport: - transport_config = config.transport_config - if transport_config is None: + def _connections_from_config(config: SyftboxManagerConfig) -> Connections: + connection_router_config = config.connection_router_config + if connection_router_config is None: raise ValueError( - "The config names no transport: set transport_config, or pass a " - "transport to SyftboxManager.from_config" + "The config names no connections: set connection_router_config, " + "or pass connections to SyftboxManager.from_config" ) - drive_config = transport_config.p2p + drive_config = connection_router_config.p2p_con if not ( isinstance(drive_config, GdriveConnectionConfig) - and transport_config.uses_one_connection() + and connection_router_config.uses_one_connection() ): raise NotImplementedError( - "Only one Google Drive connection for every transport component " - "is built so far" + "Only one Google Drive login for every connection is built so far" ) cached_client = DriveCachedClient.from_config(drive_config) - return SyftboxManager._drive_transport(cached_client) + return SyftboxManager._drive_connections(cached_client) @staticmethod - def _drive_transport(cached_client: DriveCachedClient) -> Transport: - """Every transport component over one Google Drive account.""" - return Transport( - p2p=DriveP2PConnection(cached_client), - owner_store=DriveOwnerStore(cached_client), - watcher_store=DriveWatcherStore(), - collections=DriveCollections(cached_client), - peering=DrivePeering(cached_client), + def _drive_connections(cached_client: DriveCachedClient) -> Connections: + """Every connection of the router over one Google Drive account.""" + return Connections( + p2p_con=DriveP2PConnection(cached_client), + owner_store_con=DriveOwnerStoreConnection(cached_client), + watcher_store_con=DriveWatcherStoreConnection(), + collections_con=DriveCollectionsConnection(cached_client), + peering_con=DrivePeeringConnection(cached_client), platform=GdriveFilesPlatform(), ) @@ -833,10 +834,10 @@ def pair_with_mock_drive_service_connection( do_config.email, ds_config.email ) do_manager = cls.from_config( - do_config, SyftboxManager._drive_transport(do_cached_client) + do_config, SyftboxManager._drive_connections(do_cached_client) ) ds_manager = cls.from_config( - ds_config, SyftboxManager._drive_transport(ds_cached_client) + ds_config, SyftboxManager._drive_connections(ds_cached_client) ) # Set up callbacks for DS -> DO communication @@ -1326,12 +1327,12 @@ def _copy(self): new_config = deepcopy(self.config) drive_service = drive_cached_client_of( - self._connection_router.transport + self._connection_router.connections ).client.drive_service if isinstance(drive_service, mock_drive_service.MockDriveService): # A new connection to the same backing store cached_client = DriveCachedClient.from_service(self.email, drive_service) return SyftboxManager.from_config( - new_config, SyftboxManager._drive_transport(cached_client) + new_config, SyftboxManager._drive_connections(cached_client) ) return SyftboxManager.from_config(new_config) diff --git a/syft/sync/sync/datasite_owner_syncer.py b/syft/sync/sync/datasite_owner_syncer.py index 43a38558f81..9e4df6bfa91 100644 --- a/syft/sync/sync/datasite_owner_syncer.py +++ b/syft/sync/sync/datasite_owner_syncer.py @@ -32,7 +32,7 @@ ) from syft.sync.messages.proposed_filechange import ProposedFileChangesMessage from syft.sync.peers.exceptions import SyftPeerNotReadyError -from syft.sync.transport.types import RawMessage +from syft.sync.connections.types import RawMessage from syft.sync.sync.caches.datasite_owner_cache import ( DataSiteOwnerEventCache, DataSiteOwnerEventCacheConfig, @@ -656,7 +656,9 @@ def _get_file_metadatas_with_new_connection( self, collection: FileCollection, spec: CollectionSyncSpec ) -> list: """Get file metadatas for a collection using a new connection for thread safety.""" - collections = self.connection_router.transport.collections.clone_for_thread() + collections = ( + self.connection_router.connections.collections_con.clone_for_thread() + ) return collections.list_files( spec.wire_prefix(collection.variant), tag=collection.tag, @@ -666,7 +668,9 @@ def _get_file_metadatas_with_new_connection( def _download_file_with_new_connection(self, file_id: str) -> bytes: """Download a file using a new connection for thread safety.""" - collections = self.connection_router.transport.collections.clone_for_thread() + collections = ( + self.connection_router.connections.collections_con.clone_for_thread() + ) return collections.fetch_file(file_id) def _create_current_filestate_event(self, path: str) -> FileChangeEvent: @@ -788,7 +792,7 @@ def _apply_proposed_message( ) -> bool: """Apply one proposal; False, with a warning, when applying it failed. Either way the caller archives it, so it does not hold back the next.""" - # The sender is the peer whose inbox held the message. The transport + # The sender is the peer whose inbox held the message. The p2p connection # read the message from that peer's folder and checked the signature # against that peer's key. The body is sender-supplied, so a claim # there that disagrees is a spoof attempt. diff --git a/syft/sync/sync/datasite_watcher_syncer.py b/syft/sync/sync/datasite_watcher_syncer.py index 9f217f98867..7771c88a1ee 100644 --- a/syft/sync/sync/datasite_watcher_syncer.py +++ b/syft/sync/sync/datasite_watcher_syncer.py @@ -187,7 +187,9 @@ def download_collection_file_with_new_connection( self, file_id: str, owner_email: str ) -> bytes: """Download collection file using a new connection (thread-safe).""" - collections = self.connection_router.transport.collections.clone_for_thread() + collections = ( + self.connection_router.connections.collections_con.clone_for_thread() + ) data = collections.fetch_file(file_id) data = self.connection_router.peer_store.decrypt_dataset_if_needed( owner_email, data diff --git a/syft/sync/transport/__init__.py b/syft/sync/transport/__init__.py deleted file mode 100644 index fcc47e061c5..00000000000 --- a/syft/sync/transport/__init__.py +++ /dev/null @@ -1,26 +0,0 @@ -"""Backend-neutral transport contracts for the syft sync engine.""" - -from syft.sync.transport.protocols import ( - Collections, - OwnerStore, - P2PConnection, - Peering, - Transport, - WatcherStore, -) -from syft.sync.transport.exceptions import TransportError, TransportUnavailableError -from syft.sync.transport.types import MessageKind, RawMessage, SendStatus - -__all__ = [ - "Collections", - "MessageKind", - "OwnerStore", - "P2PConnection", - "Peering", - "RawMessage", - "SendStatus", - "Transport", - "TransportError", - "TransportUnavailableError", - "WatcherStore", -] diff --git a/syft/sync/transport/config.py b/syft/sync/transport/config.py deleted file mode 100644 index 01c943062db..00000000000 --- a/syft/sync/transport/config.py +++ /dev/null @@ -1,31 +0,0 @@ -"""The connection each transport component uses, as plain config.""" - -from typing import Self - -from pydantic import BaseModel - -from syft.sync.connections.base_connection import ConnectionConfig -from syft.sync.transport.protocols import TRANSPORT_COMPONENTS - - -class TransportConfig(BaseModel): - """The connection each transport component uses. - ``SyftboxManager.from_config`` builds the transport from it; only one - Google Drive connection for every component is built so far.""" - - p2p: ConnectionConfig - owner_store: ConnectionConfig - watcher_store: ConnectionConfig - collections: ConnectionConfig - peering: ConnectionConfig - - @classmethod - def one_connection(cls, connection_config: ConnectionConfig) -> Self: - """Every component over the same connection.""" - return cls(**{name: connection_config for name in TRANSPORT_COMPONENTS}) - - def connection_configs(self) -> list[ConnectionConfig]: - return [getattr(self, name) for name in TRANSPORT_COMPONENTS] - - def uses_one_connection(self) -> bool: - return all(config == self.p2p for config in self.connection_configs()) diff --git a/syft/sync/transport/exceptions.py b/syft/sync/transport/exceptions.py deleted file mode 100644 index 9b28a1dcce5..00000000000 --- a/syft/sync/transport/exceptions.py +++ /dev/null @@ -1,9 +0,0 @@ -"""Errors raised by every transport backend.""" - - -class TransportError(Exception): - """Base exception for transport errors.""" - - -class TransportUnavailableError(TransportError): - """The backend could not be reached.""" diff --git a/syft/sync/version/peer_manager.py b/syft/sync/version/peer_manager.py index 4ff7c370007..fa396eabbf1 100644 --- a/syft/sync/version/peer_manager.py +++ b/syft/sync/version/peer_manager.py @@ -20,7 +20,7 @@ datasite_settings_path, resolve_encryption_policy, ) -from syft.sync.transport.exceptions import TransportUnavailableError +from syft.sync.connections.exceptions import ConnectionUnavailableError from syft.sync.utils.print_utils import ( print_peer_already_connected, print_peer_connection_established, @@ -277,13 +277,13 @@ def _load_single_peer_info(self, peer_email: str) -> tuple[str, Optional[PeerInf A peer whose file cannot be read gets None, so it does not stop the sync. Raises: - TransportUnavailableError: when the backend cannot be reached. + ConnectionUnavailableError: when the backend cannot be reached. """ # A copy per thread: the connection may not be thread-safe - peering = self.connection_router.transport.peering.clone_for_thread() + peering = self.connection_router.connections.peering_con.clone_for_thread() try: return (peer_email, peering.read_peer_info(peer_email)) - except TransportUnavailableError: + except ConnectionUnavailableError: raise except Exception as e: if not self.suppress_version_warnings: @@ -594,7 +594,7 @@ def load_peers(self, force_download: bool = False): peers.append( Peer( email=email, - platforms=[self.connection_router.transport.platform], + platforms=[self.connection_router.connections.platform], state=PeerState.REQUESTED_BY_PEER, ) ) @@ -650,7 +650,7 @@ def _check_published_bundles(self, peer_emails: list[str]) -> None: def _read_own_published_bundle(self, peer_email: str) -> dict | None: """The bundle this client published for ``peer_email``, or None (thread-safe).""" - peering = self.connection_router.transport.peering.clone_for_thread() + peering = self.connection_router.connections.peering_con.clone_for_thread() bundle_json = peering.read_own_key_bundle(peer_email) if bundle_json is None: return None diff --git a/tests/integration/with_unit_coverage/test_sync_manager.py b/tests/integration/with_unit_coverage/test_sync_manager.py index 9bf2baccd0e..d7fd19eaa7e 100644 --- a/tests/integration/with_unit_coverage/test_sync_manager.py +++ b/tests/integration/with_unit_coverage/test_sync_manager.py @@ -552,7 +552,7 @@ def test_version_upgrade_breaks_communication(): ) # Phase 2: Simulate DS "upgrading" to a new incompatible version - ds_peering = ds_manager._connection_router.transport.peering + ds_peering = ds_manager._connection_router.connections.peering_con current = VersionInfo.current() new_version = VersionInfo( diff --git a/tests/migrations/p2p/test_message_protocol_downgrade.py b/tests/migrations/p2p/test_message_protocol_downgrade.py index ce810d0bcfe..91457516c47 100644 --- a/tests/migrations/p2p/test_message_protocol_downgrade.py +++ b/tests/migrations/p2p/test_message_protocol_downgrade.py @@ -24,7 +24,7 @@ from syft.sync.events.file_change_event import FileChangeEventsMessageV1 from syft.sync.messages.proposed_filechange import ProposedFileChangesMessageV1 from syft.sync.syftbox_manager import SyftboxManager -from syft.sync.transport import MessageKind +from syft.sync.connections.types import MessageKind from syft.sync.utils.syftbox_utils import uncompress_data from syft_migration import MigrationError, ProtocolSchema @@ -120,7 +120,7 @@ def _wire_version(manager, peer_email: str, raw: bytes) -> str: def _raw_proposal_version(do_manager, ds_email: str) -> str: """The version field of the next proposal blob in the DO's inbox, unparsed.""" - [message] = do_manager._connection_router.transport.p2p.receive( + [message] = do_manager._connection_router.connections.p2p_con.receive( ds_email, MessageKind.PROPOSED_CHANGES, max_messages=1 ) return _wire_version(do_manager, ds_email, message.data) @@ -128,7 +128,7 @@ def _raw_proposal_version(do_manager, ds_email: str) -> str: def _raw_outbox_versions(ds_manager, do_email: str) -> list[str]: """The version fields of the DO's outbox blobs for us, unparsed.""" - p2p = ds_manager._connection_router.transport.p2p + p2p = ds_manager._connection_router.connections.p2p_con raw_list = p2p.outbox_messages(do_email, None) return [_wire_version(ds_manager, do_email, raw) for raw in raw_list] diff --git a/tests/unit/test_checkpoints.py b/tests/unit/test_checkpoints.py index ede30eddc23..8a6bbd84eec 100644 --- a/tests/unit/test_checkpoints.py +++ b/tests/unit/test_checkpoints.py @@ -13,7 +13,7 @@ import pytest from googleapiclient.errors import HttpError -from syft.sync.connections.drive.drive_owner_store import DriveOwnerStore +from syft.sync.connections.drive.drive_owner_store import DriveOwnerStoreConnection from syft.sync.checkpoints.checkpoint import ( CHECKPOINT_FILENAME_PREFIX, INCREMENTAL_CHECKPOINT_PREFIX, @@ -753,7 +753,7 @@ def test_owner_state_is_written_by_default(): def test_latest_checkpoint_read_retries_drive_error(monkeypatch): do_cached_client, _ = pair_with_mock_service("do@example.com", "ds@example.com") - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.put_checkpoint("checkpoint_1.0.tar.gz", b"full") fail_next_listing(monkeypatch, CHECKPOINT_FILENAME_PREFIX) @@ -762,7 +762,7 @@ def test_latest_checkpoint_read_retries_drive_error(monkeypatch): def test_incremental_count_retries_drive_error(monkeypatch): do_cached_client, _ = pair_with_mock_service("do@example.com", "ds@example.com") - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.put_incremental("incremental_checkpoint_1_1.0.tar.gz", b"inc") fail_next_listing(monkeypatch, INCREMENTAL_CHECKPOINT_PREFIX) @@ -771,7 +771,7 @@ def test_incremental_count_retries_drive_error(monkeypatch): def test_latest_checkpoint_download_error_raises(monkeypatch): do_cached_client, _ = pair_with_mock_service("do@example.com", "ds@example.com") - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.put_checkpoint("checkpoint_1.0.tar.gz", b"full") def forbidden(self, file_id: str) -> bytes: diff --git a/tests/unit/test_transport_contracts.py b/tests/unit/test_connection_contracts.py similarity index 64% rename from tests/unit/test_transport_contracts.py rename to tests/unit/test_connection_contracts.py index 467fb3aed46..691673589f3 100644 --- a/tests/unit/test_transport_contracts.py +++ b/tests/unit/test_connection_contracts.py @@ -1,4 +1,4 @@ -"""The transport contracts: the abstract jobs, their helper methods, the +"""The connection contracts: the abstract jobs, their helper methods, the message types, the naming rules and the DeliveryTracker state.""" from types import SimpleNamespace @@ -13,19 +13,24 @@ DeliveryState, DeliveryStateStore, ) -from syft.sync.transport import ( - Collections, - MessageKind, - OwnerStore, +from syft.sync.connections.protocols import ( + CollectionsConnection, + Connections, + OwnerStoreConnection, P2PConnection, - Peering, - RawMessage, - Transport, - WatcherStore, + PeeringConnection, + WatcherStoreConnection, ) -from syft.sync.transport.naming import timestamp_from_message_name +from syft.sync.connections.types import MessageKind, RawMessage +from syft.sync.connections.naming import timestamp_from_message_name -JOBS = [P2PConnection, OwnerStore, WatcherStore, Collections, Peering] +JOBS = [ + P2PConnection, + OwnerStoreConnection, + WatcherStoreConnection, + CollectionsConnection, + PeeringConnection, +] @pytest.mark.parametrize("job", JOBS) @@ -35,58 +40,63 @@ def test_a_job_cannot_be_used_without_a_backend(job): def test_owner_store_helpers_are_built_from_list_and_fetch(): - store = Mock(spec=OwnerStore) + store = Mock(spec=OwnerStoreConnection) store.list_event_handles.return_value = ["h1", "h2"] store.fetch.side_effect = lambda handle: f"bytes of {handle}".encode() - assert OwnerStore.fetch_events_since(store, 5.0) == [b"bytes of h1", b"bytes of h2"] + assert OwnerStoreConnection.fetch_events_since(store, 5.0) == [ + b"bytes of h1", + b"bytes of h2", + ] store.list_event_handles.assert_called_with(5.0) - assert OwnerStore.count_events_since(store, 5.0) == 2 + assert OwnerStoreConnection.count_events_since(store, 5.0) == 2 def test_fetching_all_events_lists_from_the_beginning(): - store = Mock(spec=OwnerStore) + store = Mock(spec=OwnerStoreConnection) store.fetch_events_since.return_value = [b"all"] - assert OwnerStore.fetch_all_events(store) == [b"all"] + assert OwnerStoreConnection.fetch_all_events(store) == [b"all"] store.fetch_events_since.assert_called_once_with(None) def test_list_mine_returns_each_tag_once_in_order(): - collections = Mock(spec=Collections) + collections = Mock(spec=CollectionsConnection) collections.list_mine_with_permissions.return_value = [ SimpleNamespace(tag="census"), SimpleNamespace(tag="census"), # the same dataset in a second layout SimpleNamespace(tag="sales"), ] - assert Collections.list_mine(collections, "syft_datasetcollection") == [ + assert CollectionsConnection.list_mine(collections, "syft_datasetcollection") == [ "census", "sales", ] def test_download_fetches_every_listed_file_by_name(): - collections = Mock(spec=Collections) + collections = Mock(spec=CollectionsConnection) collections.list_files.return_value = [ {"file_id": "id-1", "file_name": "mock.csv"}, {"file_id": "id-2", "file_name": "readme.md"}, ] collections.fetch_file.side_effect = lambda handle: handle.encode() - files = Collections.download(collections, "prefix", "census", "abc", "do@x.org") + files = CollectionsConnection.download( + collections, "prefix", "census", "abc", "do@x.org" + ) assert files == {"mock.csv": b"id-1", "readme.md": b"id-2"} -def test_a_transport_bundles_one_of_each_job_and_cannot_be_changed(): - transport = Transport( - p2p=Mock(spec=P2PConnection), - owner_store=Mock(spec=OwnerStore), - watcher_store=Mock(spec=WatcherStore), - collections=Mock(spec=Collections), - peering=Mock(spec=Peering), +def test_connections_hold_one_of_each_job_and_cannot_be_changed(): + connections = Connections( + p2p_con=Mock(spec=P2PConnection), + owner_store_con=Mock(spec=OwnerStoreConnection), + watcher_store_con=Mock(spec=WatcherStoreConnection), + collections_con=Mock(spec=CollectionsConnection), + peering_con=Mock(spec=PeeringConnection), platform=GdriveFilesPlatform(), ) with pytest.raises(ValidationError): - transport.p2p = Mock(spec=P2PConnection) + connections.p2p_con = Mock(spec=P2PConnection) def test_a_received_message_cannot_be_changed(): diff --git a/tests/unit/test_dataset_collection_listing.py b/tests/unit/test_dataset_collection_listing.py index 5aa34119d11..b09655dc68f 100644 --- a/tests/unit/test_dataset_collection_listing.py +++ b/tests/unit/test_dataset_collection_listing.py @@ -1,4 +1,4 @@ -"""Collections.list_mine_with_permissions on Drive skips only bad names. +"""CollectionsConnection.list_mine_with_permissions on Drive skips only bad names. The Drive query matches a name prefix, so another tool can return a folder that this client cannot parse. The listing skips that folder. Every other failure is a @@ -10,7 +10,7 @@ import pytest from syft_datasets.dataset_manager import DATASET_COLLECTION_PREFIX -from syft.sync.connections.drive.drive_collections import DriveCollections +from syft.sync.connections.drive.drive_collections import DriveCollectionsConnection from syft.sync.connections.drive.gdrive_client import GDriveClient from syft.sync.connections.drive.drive_cached_client import DriveCachedClient @@ -24,7 +24,7 @@ def _collections(files): cached_client.client.drive_service.files().list().execute.return_value = { "files": files } - return DriveCollections(cached_client) + return DriveCollectionsConnection(cached_client) def test_a_valid_collection_is_returned(): diff --git a/tests/unit/test_delete_syftbox.py b/tests/unit/test_delete_syftbox.py index f61bd8d1ccc..cbf17094953 100644 --- a/tests/unit/test_delete_syftbox.py +++ b/tests/unit/test_delete_syftbox.py @@ -205,11 +205,11 @@ def test_syftbox_folder_not_found(self): def _cached_client(manager: SyftboxManager): - return drive_cached_client_of(manager._connection_router.transport) + return drive_cached_client_of(manager._connection_router.connections) def _owner_store(manager: SyftboxManager): - return manager._connection_router.transport.owner_store + return manager._connection_router.connections.owner_store_con def _archive_one_proposal(): diff --git a/tests/unit/test_drive_connections.py b/tests/unit/test_drive_connections.py index 68e3da70632..1e5d5bad6d8 100644 --- a/tests/unit/test_drive_connections.py +++ b/tests/unit/test_drive_connections.py @@ -1,4 +1,4 @@ -"""The transport jobs on Google Drive, and the router that calls them. +"""The connections on Google Drive, and the router that calls them. These tests pin what each job promises, on the mock Drive service the rest of the suite uses. @@ -20,12 +20,12 @@ from syft.sync.connections.drive.drive_cached_client import DriveCachedClient from syft.sync.connections.drive.drive_collections import ( CollectionFolder, - DriveCollections, + DriveCollectionsConnection, ) -from syft.sync.connections.drive.drive_connections import DriveWatcherStore -from syft.sync.connections.drive.drive_owner_store import DriveOwnerStore +from syft.sync.connections.drive.drive_connections import DriveWatcherStoreConnection +from syft.sync.connections.drive.drive_owner_store import DriveOwnerStoreConnection from syft.sync.connections.drive.drive_p2p import DriveP2PConnection -from syft.sync.connections.drive.drive_peering import DrivePeering +from syft.sync.connections.drive.drive_peering import DrivePeeringConnection from syft.sync.connections.drive.gdrive_client import GDriveClient from syft.sync.connections.drive.mock_drive_service import ( MockDeleteRequest, @@ -40,18 +40,16 @@ from syft.sync.platforms.gdrive_files_platform import GdriveFilesPlatform from syft.sync.syftbox_manager import SyftboxManager from syft.sync.sync import datasite_owner_syncer -from syft.sync.transport import ( - Collections, - MessageKind, - OwnerStore, +from syft.sync.connections.exceptions import ConnectionUnavailableError +from syft.sync.connections.protocols import ( + CollectionsConnection, + Connections, + OwnerStoreConnection, P2PConnection, - Peering, - RawMessage, - SendStatus, - Transport, - TransportUnavailableError, - WatcherStore, + PeeringConnection, + WatcherStoreConnection, ) +from syft.sync.connections.types import MessageKind, RawMessage, SendStatus from syft.sync.version.peer_info import PeerInfo from syft.sync.version.version_info import VersionInfo from tests.unit.test_sync_manager import _store_knowing @@ -77,23 +75,23 @@ def _events_name(ts: float) -> str: def pair(): """DO and DS cached clients over one mock Drive, with the DS's channels open.""" do_cached_client, ds_cached_client = pair_with_mock_service(DO, DS) - DrivePeering(ds_cached_client).open_channels(DO) + DrivePeeringConnection(ds_cached_client).open_channels(DO) return do_cached_client, ds_cached_client -# -- the transport ---------------------------------------------------------------- +# -- the connections -------------------------------------------------------------- -def test_drive_transport_does_every_job(): +def test_drive_connections_do_every_job(): do_cached_client, _ = pair_with_mock_service(DO, DS) - transport = SyftboxManager._drive_transport(do_cached_client) - 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) + connections = SyftboxManager._drive_connections(do_cached_client) + assert isinstance(connections, Connections) + assert isinstance(connections.p2p_con, P2PConnection) + assert isinstance(connections.owner_store_con, OwnerStoreConnection) + assert isinstance(connections.watcher_store_con, WatcherStoreConnection) + assert isinstance(connections.collections_con, CollectionsConnection) + assert isinstance(connections.peering_con, PeeringConnection) + assert isinstance(connections.platform, GdriveFilesPlatform) def test_login_on_new_account_searches_each_folder_once(monkeypatch): @@ -293,7 +291,7 @@ def test_receiving_inbox_while_drive_is_down_raises(pair, monkeypatch): do_cached_client, ds_cached_client = pair _send_proposals(ds_cached_client, 2) monkeypatch.setattr(GDriveClient, "download_file", _drive_down) - with pytest.raises(TransportUnavailableError): + with pytest.raises(ConnectionUnavailableError): DriveP2PConnection(do_cached_client).receive(DS, MessageKind.PROPOSED_CHANGES) @@ -359,12 +357,12 @@ def test_receiving_file_events_through_p2p_is_not_built_yet(pair): p2p.mark_processed(message) -# -- OwnerStore ------------------------------------------------------------------- +# -- OwnerStoreConnection --------------------------------------------------------- def test_owner_event_log_is_read_oldest_first(pair): do_cached_client, _ = pair - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) # 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") @@ -380,7 +378,7 @@ def test_owner_event_log_is_read_oldest_first(pair): def test_owner_event_count_matches_the_listed_handles(pair): do_cached_client, _ = pair - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.append_event(_events_name(1.0), b"event") store.append_event("syfteventsmessagev3_not-a-valid-name.tar.gz", b"junk") @@ -389,17 +387,17 @@ def test_owner_event_count_matches_the_listed_handles(pair): def test_owner_event_log_read_while_drive_is_down_raises(pair, monkeypatch): do_cached_client, _ = pair - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.append_event(_events_name(2.0), b"event") monkeypatch.setattr(GDriveClient, "download_file", _drive_down) - with pytest.raises(TransportUnavailableError): + with pytest.raises(ConnectionUnavailableError): store.fetch_events_since(1.0) def test_owner_event_log_download_error_raises(pair, monkeypatch): do_cached_client, _ = pair - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.append_event(_events_name(2.0), b"event") def forbidden(self, file_id: str) -> bytes: @@ -412,7 +410,7 @@ def forbidden(self, file_id: str) -> bytes: def test_owner_store_keeps_the_newest_full_checkpoint(pair): do_cached_client, _ = pair - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) 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" @@ -420,7 +418,7 @@ def test_owner_store_keeps_the_newest_full_checkpoint(pair): def test_owner_store_counts_and_deletes_incremental_checkpoints(pair): do_cached_client, _ = pair - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.put_incremental("incremental_checkpoint_1_1.0.tar.gz", b"inc") assert store.incremental_count() == 1 assert store.next_incremental_sequence() == 2 @@ -431,7 +429,7 @@ def test_owner_store_counts_and_deletes_incremental_checkpoints(pair): def test_owner_store_overwrites_and_deletes_the_rolling_state(pair): do_cached_client, _ = pair - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) 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" @@ -441,7 +439,7 @@ def test_owner_store_overwrites_and_deletes_the_rolling_state(pair): def test_owner_store_files_carry_the_syft_tag(pair): do_cached_client, _ = pair - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.put_checkpoint("checkpoint_1.0.tar.gz", b"full") store.put_incremental("incremental_checkpoint_1_1.0.tar.gz", b"inc") store.put_rolling_state("rolling_state_1.0.tar.gz", b"state") @@ -456,23 +454,23 @@ def test_owner_store_files_carry_the_syft_tag(pair): assert untagged == [] -# -- WatcherStore ----------------------------------------------------------------- +# -- WatcherStoreConnection ------------------------------------------------------- def test_watcher_storage_on_drive_is_not_built_yet(pair): _, ds_cached_client = pair with pytest.raises(NotImplementedError): - DriveWatcherStore().latest_checkpoint(DO) + DriveWatcherStoreConnection().latest_checkpoint(DO) -# -- Collections ------------------------------------------------------------------ +# -- CollectionsConnection -------------------------------------------------------- def test_a_published_collection_is_listed_and_downloaded_by_its_reader(pair): do_cached_client, ds_cached_client = pair owner, reader = ( - DriveCollections(do_cached_client), - DriveCollections(ds_cached_client), + DriveCollectionsConnection(do_cached_client), + DriveCollectionsConnection(ds_cached_client), ) files = {"mock.csv": b"a,b\n1,2\n"} content_hash = CollectionFolder.compute_hash(files) @@ -497,7 +495,7 @@ def test_a_published_collection_is_listed_and_downloaded_by_its_reader(pair): def test_a_failed_collection_delete_raises(pair, monkeypatch): do_cached_client, _ = pair - owner = DriveCollections(do_cached_client) + owner = DriveCollectionsConnection(do_cached_client) owner.create(DATASETS, "census", "hash", DO) def forbidden(self) -> None: @@ -511,31 +509,31 @@ def forbidden(self) -> None: assert owner.list_mine(DATASETS) == ["census"] -# -- Peering ---------------------------------------------------------------------- +# -- PeeringConnection ------------------------------------------------------------ def test_peering_finds_the_peer_that_opened_channels(pair): do_cached_client, _ = pair - peering = DrivePeering(do_cached_client) + peering = DrivePeeringConnection(do_cached_client) assert peering.whoami() == DO assert DS in peering.incoming_requests() def test_peering_keeps_the_peer_registry(pair): do_cached_client, _ = pair - peering = DrivePeering(do_cached_client) + peering = DrivePeeringConnection(do_cached_client) peering.update_peer_registry(DS, "accepted") assert peering.read_peer_registry(force_download=True)[DS]["state"] == "accepted" def test_peer_registry_update_while_drive_is_down_keeps_other_peers(pair, monkeypatch): do_cached_client, _ = pair - peering = DrivePeering(do_cached_client) + peering = DrivePeeringConnection(do_cached_client) peering.update_peer_registry(DS, "accepted") peering._peers_json_cache = None # a new process has no cache monkeypatch.setattr(GDriveClient, "download_file", _drive_down) - with pytest.raises(TransportUnavailableError): + with pytest.raises(ConnectionUnavailableError): peering.update_peer_registry("other@example.com", "accepted") monkeypatch.undo() @@ -543,14 +541,17 @@ def test_peer_registry_update_while_drive_is_down_keeps_other_peers(pair, monkey def test_reading_own_key_bundle_creates_no_folder(pair): - owner = DrivePeering(pair[0]) + owner = DrivePeeringConnection(pair[0]) assert owner.read_own_key_bundle(DS) is None assert owner._find_encryption_bundles_folder_id() is None def test_peering_publishes_a_key_bundle_both_sides_can_read(pair): do_cached_client, ds_cached_client = pair - owner, scientist = DrivePeering(do_cached_client), DrivePeering(ds_cached_client) + owner, scientist = ( + DrivePeeringConnection(do_cached_client), + DrivePeeringConnection(ds_cached_client), + ) bundle = '{"bundle": "do-public-key"}' owner.publish_key_bundle(DS, bundle) @@ -563,7 +564,10 @@ def test_peering_publishes_a_key_bundle_both_sides_can_read(pair): def test_peering_publishes_peer_info_the_peer_can_read(pair): do_cached_client, ds_cached_client = pair - owner, scientist = DrivePeering(do_cached_client), DrivePeering(ds_cached_client) + owner, scientist = ( + DrivePeeringConnection(do_cached_client), + DrivePeeringConnection(ds_cached_client), + ) info = PeerInfo( version_info=VersionInfo.current(), encryption_policy=EncryptionPolicy.REQUIRED, @@ -582,10 +586,12 @@ def test_peering_publishes_peer_info_the_peer_can_read(pair): def _router_with_mock_p2p(): do_cached_client, _ = pair_with_mock_service(DO, DS) p2p = Mock(spec=DriveP2PConnection) - transport = SyftboxManager._drive_transport(do_cached_client).model_copy( - update={"p2p": p2p} + connections = SyftboxManager._drive_connections(do_cached_client).model_copy( + update={"p2p_con": p2p} + ) + router = ConnectionRouter( + connections=connections, peer_store=_store_knowing(DO, DS) ) - router = ConnectionRouter(transport=transport, peer_store=_store_knowing(DO, DS)) return router, p2p diff --git a/tests/unit/test_encryption.py b/tests/unit/test_encryption.py index 5621bb15f22..c35aef902dd 100644 --- a/tests/unit/test_encryption.py +++ b/tests/unit/test_encryption.py @@ -624,7 +624,7 @@ def test_tampered_inbox_message_is_archived_with_warning(): ds_manager._send_file_change(file_path, "will be tampered") # Tamper with raw bytes in the mock backing store - do_cached_client = drive_cached_client_of(do_manager._connection_router.transport) + do_cached_client = drive_cached_client_of(do_manager._connection_router.connections) inbox_id = do_cached_client.get_own_datasite_inbox_id(ds_manager.email) files = do_cached_client.get_file_metadatas_from_folder(inbox_id) assert len(files) > 0 diff --git a/tests/unit/test_encryption_policy.py b/tests/unit/test_encryption_policy.py index 63f963b0e56..b79065e92ee 100644 --- a/tests/unit/test_encryption_policy.py +++ b/tests/unit/test_encryption_policy.py @@ -9,7 +9,7 @@ import pytest from pydantic import ValidationError -from syft.sync.connections.drive.drive_peering import DrivePeering +from syft.sync.connections.drive.drive_peering import DrivePeeringConnection from syft.sync.connections.connection_router import ConnectionRouter from syft.sync.connections.drive import mock_drive_service from syft.sync.encryption_policy import ( @@ -29,7 +29,7 @@ ) from syft.sync.syftbox_manager import SyftboxManager from syft.sync.sync.datasite_owner_syncer import OutgoingMessage -from syft.sync.transport import MessageKind +from syft.sync.connections.types import MessageKind from syft.sync.version.peer_info import PeerInfo from syft.sync.version.peer_manager import PeerManager, PeerManagerConfig from syft.sync.version.version_info import VersionInfo @@ -76,7 +76,7 @@ def _keyed_pair( def _raw_inbox_message(do_manager, sender_email: str) -> bytes: router = do_manager.datasite_owner_syncer.connection_router - [message] = router.transport.p2p.receive( + [message] = router.connections.p2p_con.receive( sender_email, MessageKind.PROPOSED_CHANGES, max_messages=1 ) return message.data @@ -173,11 +173,13 @@ def test_missing_version_file_created_with_own_policy(): a_cached_client, b_cached_client = mock_drive_service.pair_with_mock_service( "a@example.com", "b@example.com" ) - DrivePeering(a_cached_client).grant_peer_info_read( + DrivePeeringConnection(a_cached_client).grant_peer_info_read( "b@example.com", _peer_info(REQUIRED) ) assert ( - DrivePeering(b_cached_client).read_peer_info("a@example.com").encryption_policy + DrivePeeringConnection(b_cached_client) + .read_peer_info("a@example.com") + .encryption_policy == REQUIRED ) diff --git a/tests/unit/test_gdrive_pagination.py b/tests/unit/test_gdrive_pagination.py index 3355b013004..9869960989e 100644 --- a/tests/unit/test_gdrive_pagination.py +++ b/tests/unit/test_gdrive_pagination.py @@ -1,9 +1,9 @@ from unittest.mock import Mock, patch from uuid import uuid4 -from syft.sync.connections.drive.drive_collections import DriveCollections +from syft.sync.connections.drive.drive_collections import DriveCollectionsConnection from syft.sync.connections.drive.drive_p2p import DriveP2PConnection -from syft.sync.connections.drive.drive_peering import DrivePeering +from syft.sync.connections.drive.drive_peering import DrivePeeringConnection from syft.sync.connections.drive.gdrive_client import GDriveClient from syft.sync.connections.drive.drive_cached_client import DriveCachedClient from syft.sync.connections.drive.mock_drive_service import ( @@ -266,8 +266,8 @@ def test_collection_listings_read_every_page(): "do@example.com", "ds@example.com" ) owner, reader = ( - DriveCollections(do_cached_client), - DriveCollections(ds_cached_client), + DriveCollectionsConnection(do_cached_client), + DriveCollectionsConnection(ds_cached_client), ) tags = [f"tag{i}" for i in range(FILE_COUNT)] for tag in tags: @@ -285,7 +285,7 @@ def test_peer_requests_read_every_page(): peer_emails = [f"peer{i}@example.com" for i in range(FILE_COUNT)] do_cached_client, *peers = _mock_cached_clients("do@example.com", *peer_emails) for peer in peers: - DrivePeering(peer).open_channels("do@example.com") + DrivePeeringConnection(peer).open_channels("do@example.com") - requests = DrivePeering(do_cached_client).incoming_requests() + requests = DrivePeeringConnection(do_cached_client).incoming_requests() assert sorted(requests) == sorted(peer_emails) diff --git a/tests/unit/test_login_email_check.py b/tests/unit/test_login_email_check.py index 8cbb29f71ac..f9d9c93d4d4 100644 --- a/tests/unit/test_login_email_check.py +++ b/tests/unit/test_login_email_check.py @@ -19,7 +19,7 @@ def test_verify_token_matches_email_passes_when_matching(): def _drive_service(manager): return drive_cached_client_of( - manager._connection_router.transport + manager._connection_router.connections ).client.drive_service diff --git a/tests/unit/test_outbox_compaction.py b/tests/unit/test_outbox_compaction.py index acebbac9f4b..c0ef9dddfbe 100644 --- a/tests/unit/test_outbox_compaction.py +++ b/tests/unit/test_outbox_compaction.py @@ -21,7 +21,7 @@ def test_compact_outboxes_if_needed_collapses_to_single_message(): ds_manager.email, msg ) - do_cached_client = drive_cached_client_of(do_manager._connection_router.transport) + do_cached_client = drive_cached_client_of(do_manager._connection_router.connections) outbox_id = do_cached_client.get_own_datasite_outbox_id(ds_manager.email) pre = do_cached_client.get_file_metadatas_from_folder( outbox_id, since_timestamp=None diff --git a/tests/unit/test_owner_inbox.py b/tests/unit/test_owner_inbox.py index d6c88bfd346..ba0cc6c6aca 100644 --- a/tests/unit/test_owner_inbox.py +++ b/tests/unit/test_owner_inbox.py @@ -9,7 +9,7 @@ from syft.sync.syftbox_manager import SyftboxManager from syft.sync.sync.datasite_owner_syncer import DatasiteOwnerSyncer -from syft.sync.transport import MessageKind +from syft.sync.connections.types import MessageKind from tests.unit.test_sender_binding import send_spoofed_message, sync_do, written_paths @@ -25,7 +25,7 @@ def _send_proposal(ds_manager: SyftboxManager, do_email: str, path: str) -> None def _inbox(do_manager: SyftboxManager, ds_email: str) -> list: - p2p = do_manager.datasite_owner_syncer.connection_router.transport.p2p + p2p = do_manager.datasite_owner_syncer.connection_router.connections.p2p_con return p2p.receive(ds_email, MessageKind.PROPOSED_CHANGES) @@ -54,7 +54,7 @@ def fail_on_first(self, sender_email, message): def test_unreadable_proposal_is_archived_and_next_is_applied(): ds_manager, do_manager = _pair_with_write_access() older = f"msgv2_{time.time() - 60}_{uuid.uuid4().hex}.tar.gz" - ds_p2p = ds_manager.datasite_watcher_syncer.connection_router.transport.p2p + ds_p2p = ds_manager.datasite_watcher_syncer.connection_router.connections.p2p_con ds_p2p.send(do_manager.email, older, b"garbage", MessageKind.PROPOSED_CHANGES) _send_proposal(ds_manager, do_manager.email, "after_garbage.txt") diff --git a/tests/unit/test_peer_request_owner.py b/tests/unit/test_peer_request_owner.py index c4b5c60ceb0..94f24af748a 100644 --- a/tests/unit/test_peer_request_owner.py +++ b/tests/unit/test_peer_request_owner.py @@ -16,7 +16,7 @@ MockPermission, ) from syft.sync.syftbox_manager import SyftboxManager -from syft.sync.transport.naming import PEERS_META_KEY +from syft.sync.connections.naming import PEERS_META_KEY from syft.sync.version.peer_manager import PeerState from syft.version import SYFT_VERSION from syft.sync.connections.drive.drive_connections import drive_cached_client_of @@ -48,7 +48,7 @@ def share_request_folder( ) -> None: """Put a request folder for `victim` in Drive and share it with `victim`.""" backing_store = drive_cached_client_of( - victim._connection_router.transport + victim._connection_router.connections ).client.drive_service._backing_store owners = [{"emailAddress": owner_email}] if owner_email else [] folder = MockDriveFile( diff --git a/tests/unit/test_peers_json_version.py b/tests/unit/test_peers_json_version.py index b887422f206..52c46a67130 100644 --- a/tests/unit/test_peers_json_version.py +++ b/tests/unit/test_peers_json_version.py @@ -16,25 +16,25 @@ from syft.sync.connections.drive.gdrive_client import GDriveClient from syft.sync.connections.drive.drive_cached_client import DriveCachedClient -from syft.sync.connections.drive.drive_peering import DrivePeering +from syft.sync.connections.drive.drive_peering import DrivePeeringConnection from syft.sync.peers.peer import PeerState from syft.sync.platforms.gdrive_files_platform import GdriveFilesPlatform -from syft.sync.transport.naming import PEERS_META_KEY, SYFT_PEERS_VERSION +from syft.sync.connections.naming import PEERS_META_KEY, SYFT_PEERS_VERSION PEER = "bob@example.com" def _peering(peers_data): cached_client = DriveCachedClient(GDriveClient(Mock(), Mock), "alice@example.com") - peering = DrivePeering(cached_client) + peering = DrivePeeringConnection(cached_client) peering._peers_json_cache = dict(peers_data) return peering def _router(peering): router = Mock() - router.transport.peering = peering - router.transport.platform = GdriveFilesPlatform() + router.connections.peering_con = peering + router.connections.platform = GdriveFilesPlatform() from syft.sync.connections.connection_router import ConnectionRouter return ConnectionRouter.get_all_peers_from_json.__get__(router, ConnectionRouter) @@ -43,7 +43,9 @@ def _router(peering): def test_a_write_stamps_the_reserved_entry(): peering = _peering({PEER: {"state": "accepted"}}) with ( - patch.object(DrivePeering, "_get_peers_file_id", return_value="file-id"), + patch.object( + DrivePeeringConnection, "_get_peers_file_id", return_value="file-id" + ), patch.object( DriveCachedClient, "get_syftbox_folder_id", return_value="folder-id" ), diff --git a/tests/unit/test_rolling_state.py b/tests/unit/test_rolling_state.py index b3723dedf20..797845ef1d3 100644 --- a/tests/unit/test_rolling_state.py +++ b/tests/unit/test_rolling_state.py @@ -11,7 +11,7 @@ import pytest from googleapiclient.errors import HttpError -from syft.sync.connections.drive.drive_owner_store import DriveOwnerStore +from syft.sync.connections.drive.drive_owner_store import DriveOwnerStoreConnection from syft.sync.checkpoints.rolling_state import ( ROLLING_STATE_FILENAME_PREFIX, RollingState, @@ -24,7 +24,7 @@ ) from syft.sync.syftbox_manager import SyftboxManager from syft.sync.sync.constants import CACHE_DIR, ROLLING_STATE_FILENAME -from syft.sync.transport import TransportUnavailableError +from syft.sync.connections.exceptions import ConnectionUnavailableError from tests.unit.utils import fail_next_listing, get_mock_event @@ -289,7 +289,7 @@ def test_no_rolling_state_when_owner_state_disabled(): def test_rolling_state_read_retries_drive_error(monkeypatch): do_cached_client, _ = pair_with_mock_service("do@example.com", "ds@example.com") - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.put_rolling_state("rolling_state_1.0.tar.gz", b"state") fail_next_listing(monkeypatch, ROLLING_STATE_FILENAME_PREFIX) @@ -298,33 +298,33 @@ def test_rolling_state_read_retries_drive_error(monkeypatch): def test_rolling_state_read_while_drive_is_down_raises(monkeypatch): do_cached_client, _ = pair_with_mock_service("do@example.com", "ds@example.com") - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.put_rolling_state("rolling_state_1.0.tar.gz", b"state") def drive_down(self, file_id: str) -> bytes: raise ConnectionResetError("Drive is down") monkeypatch.setattr(GDriveClient, "download_file", drive_down) - with pytest.raises(TransportUnavailableError): + with pytest.raises(ConnectionUnavailableError): store.get_rolling_state() def test_rolling_state_listing_while_drive_is_down_raises(monkeypatch): do_cached_client, _ = pair_with_mock_service("do@example.com", "ds@example.com") - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.put_rolling_state("rolling_state_1.0.tar.gz", b"state") def drive_down(self) -> dict: raise ConnectionRefusedError("Drive is down") monkeypatch.setattr(MockListRequest, "execute", drive_down) - with pytest.raises(TransportUnavailableError): + with pytest.raises(ConnectionUnavailableError): store.get_rolling_state() def test_rolling_state_download_error_raises(monkeypatch): do_cached_client, _ = pair_with_mock_service("do@example.com", "ds@example.com") - store = DriveOwnerStore(do_cached_client) + store = DriveOwnerStoreConnection(do_cached_client) store.put_rolling_state("rolling_state_1.0.tar.gz", b"state") def forbidden(self, file_id: str) -> bytes: diff --git a/tests/unit/test_sender_binding.py b/tests/unit/test_sender_binding.py index 87929d02ed1..14ebf43d739 100644 --- a/tests/unit/test_sender_binding.py +++ b/tests/unit/test_sender_binding.py @@ -82,7 +82,7 @@ def test_spoofed_sender_does_not_name_an_archive_folder(): sync_do(do_manager) do_cached_client = drive_cached_client_of( - do_manager.datasite_owner_syncer.connection_router.transport + do_manager.datasite_owner_syncer.connection_router.connections ) folder_names = [ f.name diff --git a/tests/unit/test_shared_router.py b/tests/unit/test_shared_router.py index add45eb5bf8..6402c8f00fe 100644 --- a/tests/unit/test_shared_router.py +++ b/tests/unit/test_shared_router.py @@ -1,5 +1,5 @@ """A manager builds one connection router and every part shares it, so the -whole manager has one transport, one connection and one peer store.""" +whole manager has one set of connections and one peer store.""" from unittest.mock import Mock @@ -7,15 +7,15 @@ from syft.sync.connections.base_connection import ConnectionConfig from syft.sync.connections.drive.drive_connections import ( - DriveWatcherStore, + DriveWatcherStoreConnection, drive_cached_client_of, ) from syft.sync.connections.drive.drive_cached_client import DriveCachedClient from syft.sync.connections.drive.grdrive_config import GdriveConnectionConfig from syft.sync.connections.drive.mock_drive_service import pair_with_mock_service from syft.sync.syftbox_manager import SyftboxManager, SyftboxManagerConfig -from syft.sync.transport.config import TransportConfig -from syft.sync.transport.protocols import OwnerStore +from syft.sync.connections.config import ConnectionRouterConfig +from syft.sync.connections.protocols import OwnerStoreConnection def _config_with_both_roles() -> SyftboxManagerConfig: @@ -35,11 +35,11 @@ def _mock_cached_client(email: str) -> DriveCachedClient: def test_every_part_of_manager_shares_one_router(): config = _config_with_both_roles() - transport = SyftboxManager._drive_transport(_mock_cached_client(config.email)) - manager = SyftboxManager.from_config(config, transport) + connections = SyftboxManager._drive_connections(_mock_cached_client(config.email)) + manager = SyftboxManager.from_config(config, connections) router = manager._connection_router - assert router.transport is transport + assert router.connections is connections assert manager.peer_manager.connection_router is router assert manager.datasite_owner_syncer.connection_router is router assert manager.datasite_watcher_syncer.connection_router is router @@ -49,13 +49,13 @@ def test_every_part_of_manager_shares_one_router(): assert manager._peer_store is router.peer_store -def _drive_cached_clients(transport) -> list[DriveCachedClient]: - """The cached client of each component that uses Drive. The watcher store does not - use it yet.""" +def _drive_cached_clients(connections) -> list[DriveCachedClient]: + """The cached client of each connection that uses Drive. The watcher store + does not use it yet.""" return [ - component._cached_client - for component in transport.components() - if not isinstance(component, DriveWatcherStore) + connection._cached_client + for connection in connections.all() + if not isinstance(connection, DriveWatcherStoreConnection) ] @@ -68,35 +68,39 @@ def build_mock_cached_client(connection_config): monkeypatch.setattr(DriveCachedClient, "from_config", build_mock_cached_client) config = _config_with_both_roles() - config.transport_config = TransportConfig.one_connection( + config.connection_router_config = ConnectionRouterConfig.one_connection( _drive_config(config.email) ) manager = SyftboxManager.from_config(config) [built_client] = built - for cached_client in _drive_cached_clients(manager._connection_router.transport): + for cached_client in _drive_cached_clients(manager._connection_router.connections): assert cached_client is built_client -def test_every_drive_component_shares_one_cached_client(): +def test_every_drive_connection_shares_one_cached_client(): _, do_manager = SyftboxManager.pair_with_mock_drive_service_connection() - transport = do_manager._connection_router.transport - cached_client = drive_cached_client_of(transport) + connections = do_manager._connection_router.connections + cached_client = drive_cached_client_of(connections) - for component_cached_client in _drive_cached_clients(transport): - assert component_cached_client is cached_client + for connection_cached_client in _drive_cached_clients(connections): + assert connection_cached_client is cached_client -def test_drive_cached_client_of_transport_not_on_drive_raises(): - transport = SyftboxManager._drive_transport(_mock_cached_client("do@example.com")) - transport = transport.model_copy(update={"owner_store": Mock(spec=OwnerStore)}) +def test_drive_cached_client_of_connections_not_on_drive_raises(): + connections = SyftboxManager._drive_connections( + _mock_cached_client("do@example.com") + ) + connections = connections.model_copy( + update={"owner_store_con": Mock(spec=OwnerStoreConnection)} + ) with pytest.raises(TypeError, match="not built on Google Drive"): - drive_cached_client_of(transport) + drive_cached_client_of(connections) -def test_manager_without_transport_config_fails_at_init(): - with pytest.raises(ValueError, match="names no transport"): +def test_manager_without_connections_config_fails_at_init(): + with pytest.raises(ValueError, match="names no connections"): SyftboxManager.from_config(_config_with_both_roles()) @@ -105,11 +109,11 @@ def test_manager_without_transport_config_fails_at_init(): [ConnectionConfig(), _drive_config("other@example.com")], ids=["not_drive", "other_drive_login"], ) -def test_transport_other_than_one_drive_connection_is_not_built_yet(p2p_config): +def test_connections_other_than_one_drive_connection_are_not_built_yet(p2p_config): config = _config_with_both_roles() - config.transport_config = TransportConfig.one_connection( + config.connection_router_config = ConnectionRouterConfig.one_connection( _drive_config(config.email) - ).model_copy(update={"p2p": p2p_config}) + ).model_copy(update={"p2p_con": p2p_config}) - with pytest.raises(NotImplementedError, match="one Google Drive connection"): + with pytest.raises(NotImplementedError, match="one Google Drive login"): SyftboxManager.from_config(config) diff --git a/tests/unit/test_sync_manager.py b/tests/unit/test_sync_manager.py index 07afb3083c0..9554379c1f5 100644 --- a/tests/unit/test_sync_manager.py +++ b/tests/unit/test_sync_manager.py @@ -3,7 +3,7 @@ import pytest import yaml -from syft.sync.connections.drive.drive_peering import DrivePeering +from syft.sync.connections.drive.drive_peering import DrivePeeringConnection from syft.sync.connections.connection_router import ConnectionRouter from syft.sync.connections.drive.drive_cached_client import DriveCachedClient from syft.sync.encryption_policy import EncryptionPolicy @@ -150,7 +150,7 @@ def test_sync_existing_datasite_state_do(): ds_manager, do_manager = SyftboxManager.pair_with_mock_drive_service_connection( use_in_memory_cache=False ) - ds_p2p = ds_manager._connection_router.transport.p2p + ds_p2p = ds_manager._connection_router.connections.p2p_con events_messages = get_mock_events_messages(2) for message in events_messages: @@ -489,13 +489,13 @@ def _peer_router(ds_manager, do_manager, email: str) -> ConnectionRouter: cached_client = DriveCachedClient.from_service( email, drive_cached_client_of( - ds_manager._connection_router.transport + ds_manager._connection_router.connections ).client.drive_service, ) - DrivePeering(cached_client).open_channels(do_manager.email) + DrivePeeringConnection(cached_client).open_channels(do_manager.email) _add_known_peer(do_manager, email) return ConnectionRouter( - transport=SyftboxManager._drive_transport(cached_client), + connections=SyftboxManager._drive_connections(cached_client), peer_store=_store_knowing(email, do_manager.email), ) diff --git a/tests/unit/test_version_negotiation.py b/tests/unit/test_version_negotiation.py index d1d0d19f0b9..99ce800ec20 100644 --- a/tests/unit/test_version_negotiation.py +++ b/tests/unit/test_version_negotiation.py @@ -7,7 +7,7 @@ from googleapiclient.errors import HttpError from syft.sync.connections.drive.gdrive_client import GDriveClient from syft.sync.syftbox_manager import SyftboxManager -from syft.sync.transport import TransportUnavailableError +from syft.sync.connections.exceptions import ConnectionUnavailableError from syft.sync.version.exceptions import ( VersionMismatchError, ) @@ -241,7 +241,7 @@ def test_load_peer_versions_parallel(self): ) backing_store = drive_cached_client_of( - ds_manager._connection_router.transport + ds_manager._connection_router.connections ).client.drive_service._backing_store third_peer_email = "third_peer@test.com" @@ -286,7 +286,7 @@ def test_drive_error_during_peer_version_read_raises(self, monkeypatch): ) monkeypatch.setattr(GDriveClient, "download_file", _raises(DRIVE_DOWN)) - with pytest.raises(TransportUnavailableError): + with pytest.raises(ConnectionUnavailableError): ds_manager.peer_manager.load_peer_infos_parallel([do_manager.email]) def test_drive_error_during_peer_version_read_aborts_sync(self, monkeypatch): @@ -295,14 +295,14 @@ def test_drive_error_during_peer_version_read_aborts_sync(self, monkeypatch): ) monkeypatch.setattr(GDriveClient, "download_file", _raises(DRIVE_DOWN)) - with pytest.raises(TransportUnavailableError): + with pytest.raises(ConnectionUnavailableError): do_manager.sync() def test_drive_error_during_own_version_read_raises(self, monkeypatch): _, do_manager = SyftboxManager.pair_with_mock_drive_service_connection() monkeypatch.setattr(GDriveClient, "download_file", _raises(DRIVE_DOWN)) - with pytest.raises(TransportUnavailableError): + with pytest.raises(ConnectionUnavailableError): do_manager.peer_manager.connection_router.read_own_peer_info_file() def test_offline_during_peer_version_read_raises_unavailable(self, monkeypatch): @@ -312,7 +312,7 @@ def test_offline_during_peer_version_read_raises_unavailable(self, monkeypatch): no_dns = httplib2.ServerNotFoundError("no DNS") monkeypatch.setattr(GDriveClient, "download_file", _raises(no_dns)) - with pytest.raises(TransportUnavailableError): + with pytest.raises(ConnectionUnavailableError): ds_manager.peer_manager.load_peer_infos_parallel([do_manager.email]) def test_drive_refusal_during_peer_version_read_skips_peer(self, monkeypatch): diff --git a/tests/unit/test_versioned_folder_adopt.py b/tests/unit/test_versioned_folder_adopt.py index fbb2cad084e..55bdb1ff170 100644 --- a/tests/unit/test_versioned_folder_adopt.py +++ b/tests/unit/test_versioned_folder_adopt.py @@ -158,12 +158,12 @@ def _checkpoints_owner_with_version(checkpoint_version: int): sync_automatically=False, ) router = do_manager._connection_router - cached_client = drive_cached_client_of(router.transport) + cached_client = drive_cached_client_of(router.connections) router.upload_checkpoint( Checkpoint(email=do_manager.email, version=checkpoint_version) ) - folder_id = router.transport.owner_store._get_checkpoints_folder_id() + folder_id = router.connections.owner_store_con._get_checkpoints_folder_id() older_name = f"{do_manager.email}-0.0.1-checkpoints" execute_with_retries( cached_client.client.drive_service.files().update( @@ -193,7 +193,7 @@ def test_an_adopted_folder_serves_a_checkpoint_from_an_earlier_client(): assert checkpoint is not None, "the adopted folder must still serve its files" assert checkpoint.version == 0 assert _checkpoints_folder_names(cached_client) == [ - router.transport.owner_store._get_checkpoints_folder_name() + router.connections.owner_store_con._get_checkpoints_folder_name() ], "the read must go through the adopt, which renames the folder" assert older_name not in _checkpoints_folder_names(cached_client) diff --git a/tests/unit/utils.py b/tests/unit/utils.py index 798fcc31f0f..a9db8c26e29 100644 --- a/tests/unit/utils.py +++ b/tests/unit/utils.py @@ -251,6 +251,6 @@ def router_with_mock_connection(email: str) -> ConnectionRouter: ``SyftboxManager.from_config`` builds.""" cached_client, _ = pair_with_mock_service(email, "peer@example.com") return ConnectionRouter( - transport=SyftboxManager._drive_transport(cached_client), + connections=SyftboxManager._drive_connections(cached_client), peer_store=PeerStore(email=email), )