diff --git a/src/sync.py b/src/sync.py index 24eb0ea96..8453d54f2 100644 --- a/src/sync.py +++ b/src/sync.py @@ -2,6 +2,7 @@ __author__ = "Mandar Patil " import datetime +import json import os from time import sleep @@ -438,6 +439,134 @@ def _log_retry_time(sleep_for: int): LOGGER.info(f"Retrying login at {next_sync} ...") +def _check_webaccess_state(api) -> dict: + """ + Check iCloud web access state to determine if Advanced Data Protection (ADP) is enabled. + + Calls the requestWebAccessState endpoint to discover whether the account has ADP enabled + and whether device consent for PCS (Private Cloud Storage) cookies has been granted. + + Args: + api: Authenticated ICloudPyService instance + + Returns: + dict: Web access state data, or empty dict on failure. + Key 'isDeviceConsentedForPCS' is present only when ADP is enabled. + """ + try: + response = api.session.post( + f"{api.setup_endpoint}/requestWebAccessState", + params=api.params, + ) + return response.json() + except Exception as error: + LOGGER.debug(f"Could not check web access state: {error}") + return {} + + +def _request_pcs_cookies(api) -> bool: + """ + Request PCS (Private Cloud Storage) cookies for ADP-protected iCloud services. + + Must be called after device consent for PCS has been granted. The cookies are + automatically persisted by the icloudpy session. + + Args: + api: Authenticated ICloudPyService instance with ADP consent granted + + Returns: + bool: True if PCS cookies were obtained successfully, False otherwise + """ + try: + response = api.session.post( + f"{api.setup_endpoint}/requestPCS", + params=api.params, + data=json.dumps({"appName": "iclouddrive", "derivedFromUserAction": False}), + ) + data = response.json() + if data.get("status") == "success": + LOGGER.debug("PCS cookies obtained successfully.") + return True + LOGGER.debug(f"PCS cookie request not yet ready: {data.get('message', 'unknown')}") + return False + except Exception as error: + LOGGER.debug(f"Failed to request PCS cookies: {error}") + return False + + +def _handle_pcs_required(config, api, username: str, sync_state: SyncState) -> bool: + """ + Handle PCS (Private Cloud Storage) cookie acquisition for ADP-enabled accounts. + + Apple's Advanced Data Protection (ADP) requires PCS cookies in addition to the + standard authentication token. This function checks if PCS is required and either + requests consent (sending a notification to the user's iPhone) or acquires the + cookies if consent is already granted. + + When ADP is enabled and consent is not yet granted, a notification is sent to the + user's iPhone. The user must approve the request on their device; sync will retry + on the next cycle. + + Args: + config: Configuration dictionary + api: Authenticated ICloudPyService instance + username: iCloud username + sync_state: Current sync state (used for notification rate limiting) + + Returns: + bool: True if sync should proceed (no ADP, or PCS cookies obtained), + False if PCS consent is still pending or the PCS state could not be + determined (sync will retry next cycle) + """ + pcs_state = _check_webaccess_state(api) + + if not pcs_state: + LOGGER.error( + "Unable to determine PCS web access state. " + "Will retry on next sync cycle before proceeding with sync.", + ) + return False + + if "isDeviceConsentedForPCS" not in pcs_state: + # ADP is not enabled for this account — no PCS handling needed + return True + + if pcs_state.get("isDeviceConsentedForPCS"): + # ADP is enabled and device consent is already granted — acquire PCS cookies + LOGGER.debug("ADP is enabled and device consent is granted. Requesting PCS cookies...") + if _request_pcs_cookies(api): + return True + LOGGER.warning("PCS cookies not yet ready. Will retry on next sync cycle.") + return False + + # ADP is enabled but device consent has not been granted yet + LOGGER.warning( + "Advanced Data Protection (ADP) is enabled. " + "Please approve the PCS consent request on your iPhone to allow iCloud access.", + ) + try: + response = api.session.post( + f"{api.setup_endpoint}/enableDeviceConsentForPCS", + params=api.params, + ) + data = response.json() + if data.get("isDeviceConsentNotificationSent"): + LOGGER.info("PCS consent notification sent to your iPhone. Please approve on your device.") + else: + LOGGER.error("Failed to send PCS consent notification to your iPhone.") + except Exception as error: + LOGGER.error(f"Failed to request PCS consent: {error}") + + server_region = config_parser.get_region(config=config) + sync_state.last_send = notify.send( + config=config, + username=username, + last_send=sync_state.last_send, + region=server_region, + ) + return False + + def _calculate_next_sync_schedule(config, sync_state: SyncState): """ Calculate next sync schedule and update sync state. @@ -559,49 +688,51 @@ def sync(): api = _authenticate_and_get_api(config, username) if not api.requires_2sa: - # Create summary for this sync cycle - summary = SyncSummary() - - # Perform syncs and collect statistics - drive_stats = _perform_drive_sync(config, api, sync_state, drive_sync_interval) - photos_stats = _perform_photos_sync(config, api, sync_state, photos_sync_interval) - - # Populate summary with statistics - summary.drive_stats = drive_stats - summary.photo_stats = photos_stats - summary.sync_end_time = datetime.datetime.now() - - # Send usage statistics (anonymized summary data) - try: - _send_usage_statistics(config, summary) - except Exception as e: - LOGGER.debug(f"Failed to send usage statistics: {e!s}") - - # Send sync summary notification if configured - # Only send notification when both enabled services have synced in this cycle - # Gracefully handle notification failures to not break sync - has_drive_config = config and "drive" in config - has_photos_config = config and "photos" in config - - should_send_notification = False - if has_drive_config and has_photos_config: - # Both services configured - send notification only when both have synced - should_send_notification = drive_stats is not None and photos_stats is not None - elif has_drive_config and not has_photos_config: - # Only drive configured - send when drive synced - should_send_notification = drive_stats is not None - elif has_photos_config and not has_drive_config: - # Only photos configured - send when photos synced - should_send_notification = photos_stats is not None - - if should_send_notification: + if _handle_pcs_required(config, api, username, sync_state): + # Create summary for this sync cycle + summary = SyncSummary() + + # Perform syncs and collect statistics + drive_stats = _perform_drive_sync(config, api, sync_state, drive_sync_interval) + photos_stats = _perform_photos_sync(config, api, sync_state, photos_sync_interval) + + # Populate summary with statistics + summary.drive_stats = drive_stats + summary.photo_stats = photos_stats + summary.sync_end_time = datetime.datetime.now() + + # Send usage statistics (anonymized summary data) try: - notify.send_sync_summary(config=config, summary=summary) + _send_usage_statistics(config, summary) except Exception as e: - LOGGER.debug(f"Failed to send sync summary notification: {e!s}") - - if not _check_services_configured(config): - LOGGER.warning("Nothing to sync. Please add drive: and/or photos: section in config.yaml file.") + LOGGER.debug(f"Failed to send usage statistics: {e!s}") + + # Send sync summary notification if configured + # Only send notification when both enabled services have synced in this cycle + # Gracefully handle notification failures to not break sync + has_drive_config = config and "drive" in config + has_photos_config = config and "photos" in config + + should_send_notification = False + if has_drive_config and has_photos_config: + # Both services configured - send notification only when both have synced + should_send_notification = drive_stats is not None and photos_stats is not None + elif has_drive_config and not has_photos_config: + # Only drive configured - send when drive synced + should_send_notification = drive_stats is not None + elif has_photos_config and not has_drive_config: + # Only photos configured - send when photos synced + should_send_notification = photos_stats is not None + + if should_send_notification: + try: + notify.send_sync_summary(config=config, summary=summary) + except Exception as e: + LOGGER.debug(f"Failed to send sync summary notification: {e!s}") + + if not _check_services_configured(config): + LOGGER.warning("Nothing to sync. Please add drive: and/or photos: section in config.yaml file.") + # else: PCS not ready; fall through to sleep/exit logic below else: if not _handle_2fa_required(config, username, sync_state): break diff --git a/tests/data/__init__.py b/tests/data/__init__.py index 33dc386b2..59a6548d1 100644 --- a/tests/data/__init__.py +++ b/tests/data/__init__.py @@ -43,7 +43,16 @@ AUTHENTICATED_USER = PRIMARY_EMAIL REQUIRES_2FA_TOKEN = "requires_2fa_token" REQUIRES_2FA_USER = "requires_2fa_user" -VALID_USERS = [AUTHENTICATED_USER, REQUIRES_2FA_USER, APPLE_ID_EMAIL, ICLOUD_ID_EMAIL] +REQUIRES_PCS_CONSENT_USER = "requires_pcs_consent_user" +AUTHENTICATED_WITH_ADP_USER = "authenticated_with_adp_user" +VALID_USERS = [ + AUTHENTICATED_USER, + REQUIRES_2FA_USER, + APPLE_ID_EMAIL, + ICLOUD_ID_EMAIL, + REQUIRES_PCS_CONSENT_USER, + AUTHENTICATED_WITH_ADP_USER, +] VALID_PASSWORD = "valid_password" VALID_COOKIE = "valid_cookie" VALID_TOKEN = "valid_token" @@ -478,6 +487,49 @@ VERIFICATION_CODE_OK = {"success": True} VERIFICATION_CODE_KO = {"success": False} +# PCS (Private Cloud Storage) mock data for Advanced Data Protection (ADP) testing +PCS_WEBACCESS_STATE_NO_ADP = { + "dsid": PERSON_ID, + "isWebAccessAllowed": True, +} +PCS_WEBACCESS_STATE_ADP_NO_CONSENT = { + "dsid": PERSON_ID, + "isWebAccessAllowed": True, + "isDeviceConsentedForPCS": False, + "isICDRSDisabled": True, + "deviceConsentForPCSExpiry": 0, +} +PCS_WEBACCESS_STATE_ADP_CONSENTED = { + "dsid": PERSON_ID, + "isWebAccessAllowed": True, + "isDeviceConsentedForPCS": True, + "isICDRSDisabled": True, + "deviceConsentForPCSExpiry": 9999999999999, +} +PCS_CONSENT_NOTIFICATION_SENT = { + "isDeviceConsentNotificationSent": True, + "isWebAccessAllowed": True, + "isDeviceConsentedForPCS": False, + "isICDRSDisabled": True, + "deviceConsentForPCSExpiry": 0, +} +PCS_REQUEST_SUCCESS = { + "isWebAccessAllowed": True, + "isDeviceConsentedForPCS": True, + "isICDRSDisabled": True, + "message": "Cookies attached.", + "deviceConsentForPCSExpiry": 9999999999999, + "status": "success", +} +PCS_REQUEST_PENDING = { + "isWebAccessAllowed": True, + "isDeviceConsentedForPCS": True, + "isICDRSDisabled": True, + "message": "Cookies not available yet on server.", + "deviceConsentForPCSExpiry": 9999999999999, + "status": "pending", +} + # Data # Re-generated device : # id = rawDeviceModel + prsId (if not None) @@ -3721,7 +3773,7 @@ def request(self, method, url, **kwargs): """Override request method.""" params = kwargs.get("params") headers = kwargs.get("headers") - data = json.loads(kwargs.get("data", "{}")) + data = json.loads(kwargs.get("data") or "{}") # Login if self.service.setup_endpoint in url: @@ -3752,6 +3804,20 @@ def request(self, method, url, **kwargs): return ResponseMock(LOGIN_WORKING) self._raise_error(None, "Session expired") + if "requestWebAccessState" in url and method == "POST": + account_name = self.service.user.get("accountName", "") + if account_name == REQUIRES_PCS_CONSENT_USER: + return ResponseMock(PCS_WEBACCESS_STATE_ADP_NO_CONSENT) + if account_name == AUTHENTICATED_WITH_ADP_USER: + return ResponseMock(PCS_WEBACCESS_STATE_ADP_CONSENTED) + return ResponseMock(PCS_WEBACCESS_STATE_NO_ADP) + + if "enableDeviceConsentForPCS" in url and method == "POST": + return ResponseMock(PCS_CONSENT_NOTIFICATION_SENT) + + if "requestPCS" in url and method == "POST": + return ResponseMock(PCS_REQUEST_SUCCESS) + if self.service.auth_endpoint in url: if "signin" in url and method == "POST": if data.get("accountName") not in VALID_USERS: diff --git a/tests/test_sync.py b/tests/test_sync.py index fa2f6b32e..23157b184 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -5,6 +5,7 @@ import os import shutil import unittest +import unittest.mock from copy import deepcopy from io import StringIO from types import SimpleNamespace @@ -560,6 +561,7 @@ def raise_on_getsize(_path): @patch("src.sync.notify.send_sync_summary", side_effect=RuntimeError("notify failure")) @patch("src.sync._perform_photos_sync") @patch("src.sync._perform_drive_sync", return_value=DriveStats(files_downloaded=1)) + @patch("src.sync._handle_pcs_required", return_value=True) @patch("src.sync._authenticate_and_get_api", return_value=SimpleNamespace(requires_2sa=False)) @patch("src.sync.read_config") @patch("requests.post", side_effect=tests.mocked_usage_post) @@ -568,6 +570,7 @@ def test_sync_summary_notification_failure_logged( mock_usage_post, mock_read_config, _mock_auth, + _mock_pcs, _mock_drive_sync, mock_photo_sync, mock_notify, @@ -821,6 +824,7 @@ def test_sync_mixed_intervals_should_not_exit( @patch("src.sync.notify.send_sync_summary") @patch("src.sync._perform_photos_sync", return_value=None) @patch("src.sync._perform_drive_sync", return_value=DriveStats(files_downloaded=1)) + @patch("src.sync._handle_pcs_required", return_value=True) @patch("src.sync._authenticate_and_get_api", return_value=SimpleNamespace(requires_2sa=False)) @patch("src.sync.read_config") @patch("requests.post", side_effect=tests.mocked_usage_post) @@ -829,6 +833,7 @@ def test_sync_notification_not_sent_when_both_services_not_synced( mock_usage_post, mock_read_config, _mock_auth, + _mock_pcs, _mock_drive_sync, _mock_photo_sync, mock_notify, @@ -1003,3 +1008,297 @@ def test_calculate_next_sync_schedule_equal_small_timers(self): self.assertEqual(sleep_for, 0, "Should have 0 sleep for small equal timers") self.assertTrue(sync_state.enable_sync_drive, "Drive sync should be enabled") self.assertFalse(sync_state.enable_sync_photos, "Photos sync should be disabled") + + # ---- PCS / Advanced Data Protection (ADP) tests ---- + + def test_check_webaccess_state_no_adp(self): + """Test _check_webaccess_state returns state without ADP keys for regular accounts.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + api_mock.session.post.return_value.json.return_value = data.PCS_WEBACCESS_STATE_NO_ADP + + result = sync._check_webaccess_state(api_mock) # noqa: SLF001 + + self.assertEqual(result, data.PCS_WEBACCESS_STATE_NO_ADP) + self.assertNotIn("isDeviceConsentedForPCS", result) + + def test_check_webaccess_state_adp_no_consent(self): + """Test _check_webaccess_state returns state with ADP consent=False when ADP enabled.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + api_mock.session.post.return_value.json.return_value = data.PCS_WEBACCESS_STATE_ADP_NO_CONSENT + + result = sync._check_webaccess_state(api_mock) # noqa: SLF001 + + self.assertIn("isDeviceConsentedForPCS", result) + self.assertFalse(result["isDeviceConsentedForPCS"]) + + def test_check_webaccess_state_adp_consented(self): + """Test _check_webaccess_state returns state with ADP consent=True when consent granted.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + api_mock.session.post.return_value.json.return_value = data.PCS_WEBACCESS_STATE_ADP_CONSENTED + + result = sync._check_webaccess_state(api_mock) # noqa: SLF001 + + self.assertIn("isDeviceConsentedForPCS", result) + self.assertTrue(result["isDeviceConsentedForPCS"]) + + def test_check_webaccess_state_exception_returns_empty(self): + """Test _check_webaccess_state returns empty dict on exception.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + api_mock.session.post.side_effect = Exception("Network error") + + result = sync._check_webaccess_state(api_mock) # noqa: SLF001 + + self.assertEqual(result, {}) + + def test_request_pcs_cookies_success(self): + """Test _request_pcs_cookies returns True on success.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + api_mock.session.post.return_value.json.return_value = data.PCS_REQUEST_SUCCESS + + result = sync._request_pcs_cookies(api_mock) # noqa: SLF001 + + self.assertTrue(result) + + def test_request_pcs_cookies_pending(self): + """Test _request_pcs_cookies returns False when cookies are not yet ready.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + api_mock.session.post.return_value.json.return_value = data.PCS_REQUEST_PENDING + + result = sync._request_pcs_cookies(api_mock) # noqa: SLF001 + + self.assertFalse(result) + + def test_request_pcs_cookies_exception(self): + """Test _request_pcs_cookies returns False on exception.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + api_mock.session.post.side_effect = Exception("Network error") + + result = sync._request_pcs_cookies(api_mock) # noqa: SLF001 + + self.assertFalse(result) + + def test_handle_pcs_required_no_adp(self): + """Test _handle_pcs_required returns True when ADP is not enabled.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + api_mock.session.post.return_value.json.return_value = data.PCS_WEBACCESS_STATE_NO_ADP + config = self.config.copy() + sync_state = sync.SyncState() + + result = sync._handle_pcs_required(config, api_mock, data.AUTHENTICATED_USER, sync_state) # noqa: SLF001 + + self.assertTrue(result) + + def test_handle_pcs_required_webaccess_state_failed(self): + """Test _handle_pcs_required returns False when webaccess state cannot be determined.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + api_mock.session.post.side_effect = Exception("Network error") + config = self.config.copy() + sync_state = sync.SyncState() + + with self.assertLogs() as captured: + result = sync._handle_pcs_required(config, api_mock, data.AUTHENTICATED_USER, sync_state) # noqa: SLF001 + + self.assertFalse(result) + self.assertTrue(any("Unable to determine PCS web access state" in msg for msg in captured.output)) + + def test_handle_pcs_required_adp_no_consent(self): + """Test _handle_pcs_required sends consent notification and returns False when consent not granted.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + + def side_effect_post(url, **kwargs): + response_mock = unittest.mock.MagicMock() + if "requestWebAccessState" in url: + response_mock.json.return_value = data.PCS_WEBACCESS_STATE_ADP_NO_CONSENT + elif "enableDeviceConsentForPCS" in url: + response_mock.json.return_value = data.PCS_CONSENT_NOTIFICATION_SENT + return response_mock + + api_mock.session.post.side_effect = side_effect_post + config = self.config.copy() + sync_state = sync.SyncState() + + with self.assertLogs() as captured: + result = sync._handle_pcs_required(config, api_mock, data.AUTHENTICATED_USER, sync_state) # noqa: SLF001 + + self.assertFalse(result) + self.assertTrue(any("Advanced Data Protection" in msg for msg in captured.output)) + + def test_handle_pcs_required_adp_no_consent_notification_not_sent(self): + """Test _handle_pcs_required logs error when notification fails to send.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + + def side_effect_post(url, **kwargs): + response_mock = unittest.mock.MagicMock() + if "requestWebAccessState" in url: + response_mock.json.return_value = data.PCS_WEBACCESS_STATE_ADP_NO_CONSENT + elif "enableDeviceConsentForPCS" in url: + response_mock.json.return_value = {"isDeviceConsentNotificationSent": False} + return response_mock + + api_mock.session.post.side_effect = side_effect_post + config = self.config.copy() + sync_state = sync.SyncState() + + with self.assertLogs() as captured: + result = sync._handle_pcs_required(config, api_mock, data.AUTHENTICATED_USER, sync_state) # noqa: SLF001 + + self.assertFalse(result) + self.assertTrue(any("Failed to send PCS consent notification" in msg for msg in captured.output)) + + def test_handle_pcs_required_adp_enable_consent_exception(self): + """Test _handle_pcs_required handles exception from enableDeviceConsentForPCS.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + + def side_effect_post(url, **kwargs): + response_mock = unittest.mock.MagicMock() + if "requestWebAccessState" in url: + response_mock.json.return_value = data.PCS_WEBACCESS_STATE_ADP_NO_CONSENT + return response_mock + connection_error = Exception("Connection error") + raise connection_error + + api_mock.session.post.side_effect = side_effect_post + config = self.config.copy() + sync_state = sync.SyncState() + + with self.assertLogs() as captured: + result = sync._handle_pcs_required(config, api_mock, data.AUTHENTICATED_USER, sync_state) # noqa: SLF001 + + self.assertFalse(result) + self.assertTrue(any("Failed to request PCS consent" in msg for msg in captured.output)) + + def test_handle_pcs_required_adp_consented_cookies_success(self): + """Test _handle_pcs_required returns True when ADP consent is granted and cookies obtained.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + + def side_effect_post(url, **kwargs): + response_mock = unittest.mock.MagicMock() + if "requestWebAccessState" in url: + response_mock.json.return_value = data.PCS_WEBACCESS_STATE_ADP_CONSENTED + elif "requestPCS" in url: + response_mock.json.return_value = data.PCS_REQUEST_SUCCESS + return response_mock + + api_mock.session.post.side_effect = side_effect_post + config = self.config.copy() + sync_state = sync.SyncState() + + result = sync._handle_pcs_required(config, api_mock, data.AUTHENTICATED_USER, sync_state) # noqa: SLF001 + + self.assertTrue(result) + + def test_handle_pcs_required_adp_consented_cookies_pending(self): + """Test _handle_pcs_required returns False when consent granted but cookies not yet ready.""" + api_mock = unittest.mock.MagicMock() + api_mock.setup_endpoint = "https://setup.icloud.com/setup/ws/1" + api_mock.params = {} + + def side_effect_post(url, **kwargs): + response_mock = unittest.mock.MagicMock() + if "requestWebAccessState" in url: + response_mock.json.return_value = data.PCS_WEBACCESS_STATE_ADP_CONSENTED + elif "requestPCS" in url: + response_mock.json.return_value = data.PCS_REQUEST_PENDING + return response_mock + + api_mock.session.post.side_effect = side_effect_post + config = self.config.copy() + sync_state = sync.SyncState() + + with self.assertLogs() as captured: + result = sync._handle_pcs_required(config, api_mock, data.AUTHENTICATED_USER, sync_state) # noqa: SLF001 + + self.assertFalse(result) + self.assertTrue(any("PCS cookies not yet ready" in msg for msg in captured.output)) + + @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) + @patch(target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER) + @patch("src.sync.ICloudPyService", side_effect=data.ICloudPyServiceMock) + @patch("src.sync.read_config") + @patch("requests.post", side_effect=tests.mocked_usage_post) + def test_sync_with_no_adp( + self, + mock_usage_post, + mock_read_config, + mock_service, + mock_get_username, + mock_get_password, + ): + """Test sync proceeds normally when account has no ADP (default state).""" + config = self.config.copy() + mock_read_config.return_value = config + if ENV_ICLOUD_PASSWORD_KEY in os.environ: + del os.environ[ENV_ICLOUD_PASSWORD_KEY] + # AUTHENTICATED_USER returns PCS_WEBACCESS_STATE_NO_ADP in the mock session + self.assertIsNone(sync.sync()) + + @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) + @patch(target="src.config_parser.get_username", return_value=data.REQUIRES_PCS_CONSENT_USER) + @patch("src.sync.ICloudPyService", side_effect=data.ICloudPyServiceMock) + @patch("src.sync.read_config") + @patch("requests.post", side_effect=tests.mocked_usage_post) + def test_sync_pcs_consent_required( + self, + mock_usage_post, + mock_read_config, + mock_service, + mock_get_username, + mock_get_password, + ): + """Test sync exits oneshot mode when PCS consent is required (ADP enabled, no consent).""" + config = self.config.copy() + mock_read_config.return_value = config + if ENV_ICLOUD_PASSWORD_KEY in os.environ: + del os.environ[ENV_ICLOUD_PASSWORD_KEY] + + with self.assertLogs() as captured: + sync.sync() + + self.assertTrue(any("Advanced Data Protection" in msg for msg in captured.output)) + + @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) + @patch(target="src.config_parser.get_username", return_value=data.AUTHENTICATED_WITH_ADP_USER) + @patch("src.sync.ICloudPyService", side_effect=data.ICloudPyServiceMock) + @patch("src.sync.read_config") + @patch("requests.post", side_effect=tests.mocked_usage_post) + def test_sync_pcs_consent_granted( + self, + mock_usage_post, + mock_read_config, + mock_service, + mock_get_username, + mock_get_password, + ): + """Test sync proceeds when ADP is enabled and consent is already granted.""" + config = self.config.copy() + mock_read_config.return_value = config + if ENV_ICLOUD_PASSWORD_KEY in os.environ: + del os.environ[ENV_ICLOUD_PASSWORD_KEY] + # AUTHENTICATED_WITH_ADP_USER returns PCS_WEBACCESS_STATE_ADP_CONSENTED then PCS_REQUEST_SUCCESS + self.assertIsNone(sync.sync())