Fix unite tests

Author: sk-keeper

keepercli-package/src/keepercli/commands/account_commands.py

@@ -2,7 +2,7 @@
 import datetime
 import os
 import re
-from typing import Tuple, Optional, List, Any
+from typing import Tuple, Optional, List, Any, Dict
 
 from keepersdk import crypto, utils, errors
 from keepersdk.authentication import keeper_auth  
@@ -394,7 +394,7 @@ def _get_current_salt(self, auth: keeper_auth.KeeperAuth, is_sso_user: bool):
             else:
                 raise kae
 
-    def _handle_sso_deletion(self, kwargs: dict, auth: keeper_auth.KeeperAuth, is_sso_user: bool, current_salt) -> bool:
+    def _handle_sso_deletion(self, kwargs: Dict, auth: keeper_auth.KeeperAuth, is_sso_user: bool, current_salt) -> bool:
         """Handle SSO password deletion request."""
         if not kwargs.get('delete_sso'):
             return False

keepercli-package/src/keepercli/commands/audit_alert.py

@@ -30,8 +30,8 @@ def __init__(self):
 class AuditSettingMixin:
     LAST_USERNAME = ""
     LAST_ENTERPRISE_ID = 0
-    SETTINGS = None  # type: Optional[dict]
-    EVENT_TYPES = None  # type: Optional[List[Tuple[int, str]]]
+    SETTINGS: Optional[Dict] = None
+    EVENT_TYPES: Optional[List[Tuple[int, str]]] = None
 
     @staticmethod
     def load_settings(auth: keeper_auth.KeeperAuth, reload: bool=False) -> Optional[Dict[str, Any]]:

keepercli-package/src/keepercli/commands/base.py

@@ -4,7 +4,7 @@
 import re
 import shlex
 import sys
-from typing import Optional, Any, Dict, Iterable, Tuple
+from typing import Optional, Any, Dict, Iterable, Tuple, Set, List
 
 from keepersdk import errors
 
@@ -146,15 +146,15 @@ def __init__(self, parser: argparse.ArgumentParser):
     def execute(self, context: KeeperParams, **kwargs):
         pass
 
-    def _get_all_option_strings(self, parser: argparse.ArgumentParser) -> set:
+    def _get_all_option_strings(self, parser: argparse.ArgumentParser) -> Set:
         """Get all valid option strings from the parser and its parents."""
         options = set()
         for action in parser._actions:
             for opt in action.option_strings:
                 options.add(opt)
         return options
 
-    def _validate_strict_options(self, arg_list: list, valid_options: set) -> None:
+    def _validate_strict_options(self, arg_list: List, valid_options: Set) -> None:
         """Validate that all option-like arguments are recognized.
         
         Raises ParseError if:

keepercli-package/src/keepercli/commands/breachwatch.py

@@ -2,7 +2,7 @@
 import base64
 import getpass
 import json
-from typing import Any, List, Optional, Set
+from typing import Any, List, Optional, Set, Dict
 
 from keepersdk.enterprise import breachwatch_report
 from keepersdk.proto import breachwatch_pb2, client_pb2
@@ -16,7 +16,7 @@
 
 logger = api.get_logger()
 
-STATUS_TO_TEXT: dict[int, str] = { 
+STATUS_TO_TEXT: Dict[int, str] = {
     client_pb2.BWStatus.GOOD: "GOOD", 
     client_pb2.BWStatus.WEAK: "WEAK", 
     client_pb2.BWStatus.BREACHED: "BREACHED" 
@@ -176,7 +176,7 @@ def execute(self, context: KeeperParams, **kwargs) -> Any:
         else:
             logger.info("No breach watch requests to process")
 
-    def _get_record_names(self, kwargs: dict) -> list[str]:
+    def _get_record_names(self, kwargs: Dict) -> List[str]:
         """Extract record names from kwargs."""
         records = kwargs.get('records')
         if not records:
@@ -187,14 +187,14 @@ def _get_record_names(self, kwargs: dict) -> list[str]:
 
         return records
 
-    def _resolve_record_uids(self, record_names: list[str], context: KeeperParams) -> Set[str]:
+    def _resolve_record_uids(self, record_names: List[str], context: KeeperParams) -> Set[str]:
         """Resolve record names to UIDs using the context."""
         record_uids: Set[str] = set()
         for record_name in record_names:
             record_uids.update(record_utils.resolve_records(record_name, context))
         return record_uids
 
-    def _get_breached_records(self, vault: vault_online.VaultOnline) -> dict[str, str]:
+    def _get_breached_records(self, vault: vault_online.VaultOnline) -> Dict[str, str]:
         """Get breached records and their passwords."""
         record_passwords = {}
 
@@ -219,7 +219,7 @@ def _extract_record_password(self, vault: vault_online.VaultOnline, record_uid:
 
         return None
 
-    def _create_breach_watch_requests(self, vault: vault_online.VaultOnline, record_passwords: dict[str, str], record_uids: Set[str]) -> list[breachwatch_pb2.BreachWatchRecordRequest]:
+    def _create_breach_watch_requests(self, vault: vault_online.VaultOnline, record_passwords: Dict[str, str], record_uids: Set[str]) -> List[breachwatch_pb2.BreachWatchRecordRequest]:
         """Create breach watch record requests for the given records."""
         bw_requests = []
 
@@ -296,7 +296,7 @@ def _get_existing_breach_watch_euid(self, vault: vault_online.VaultOnline, recor
 
         return None
 
-    def _process_breach_watch_requests(self, vault: vault_online.VaultOnline, bw_requests: list[breachwatch_pb2.BreachWatchRecordRequest]) -> None:
+    def _process_breach_watch_requests(self, vault: vault_online.VaultOnline, bw_requests: List[breachwatch_pb2.BreachWatchRecordRequest]) -> None:
         """Process the breach watch requests."""
         # Queue audit event
         self._queue_audit_event(vault)
@@ -309,7 +309,7 @@ def _queue_audit_event(self, vault: vault_online.VaultOnline) -> None:
         if audit_plugin:
             audit_plugin.schedule_audit_event('bw_record_ignored')
 
-    def _send_breach_watch_requests(self, vault: vault_online.VaultOnline, bw_requests: list[breachwatch_pb2.BreachWatchRecordRequest]) -> None:
+    def _send_breach_watch_requests(self, vault: vault_online.VaultOnline, bw_requests: List[breachwatch_pb2.BreachWatchRecordRequest]) -> None:
         """Send breach watch requests in chunks."""
         while bw_requests:
             chunk = bw_requests[0:999]
@@ -321,7 +321,7 @@ def _send_breach_watch_requests(self, vault: vault_online.VaultOnline, bw_reques
             except Exception as e:
                 logger.error(f'Error sending breach watch chunk: {e}')
 
-    def _send_breach_watch_chunk(self, vault: vault_online.VaultOnline, chunk: list[breachwatch_pb2.BreachWatchRecordRequest]) -> breachwatch_pb2.BreachWatchUpdateResponse:
+    def _send_breach_watch_chunk(self, vault: vault_online.VaultOnline, chunk: List[breachwatch_pb2.BreachWatchRecordRequest]) -> breachwatch_pb2.BreachWatchUpdateResponse:
         """Send a chunk of breach watch requests."""
         rq = breachwatch_pb2.BreachWatchUpdateRequest()
         rq.breachWatchRecordRequest.extend(chunk)
@@ -368,7 +368,7 @@ def _validate_context(self, context: KeeperParams) -> None:
         if not context.auth.auth_context.license.get('breachWatchEnabled'):
             raise ValueError("Breach watch is not enabled. Please contact your administrator to enable this feature.")
 
-    def _get_and_validate_record_uids(self, kwargs: dict) -> list[str]:
+    def _get_and_validate_record_uids(self, kwargs: Dict) -> List[str]:
         """Extract and validate record UIDs from kwargs."""
         record_uids = kwargs.get('records')
         if not record_uids:
@@ -479,7 +479,7 @@ def _is_vault_ready(self, context: KeeperParams) -> bool:
             raise base.CommandError('Breach watch is not enabled. Please contact your administrator to enable this feature.')
         return True
 
-    def _get_passwords_to_scan(self, kwargs: dict) -> list[str]:
+    def _get_passwords_to_scan(self, kwargs: Dict) -> List[str]:
         """Get passwords from command line arguments or prompt user."""
         passwords = kwargs.get('passwords', [])
         if passwords:
@@ -493,7 +493,7 @@ def _get_passwords_to_scan(self, kwargs: dict) -> list[str]:
             logger.info('')
         return []
 
-    def _scan_passwords(self, breach_watch, passwords: list[str]) -> list:
+    def _scan_passwords(self, breach_watch, passwords: List[str]) -> List:
         """Scan passwords and return results with EUIDs for cleanup."""
         scan_results = []
         for result in breach_watch.scan_passwords(passwords):
@@ -505,7 +505,7 @@ def _is_valid_scan_result(self, result) -> bool:
         """Validate scan result structure."""
         return result and len(result) == 2
 
-    def _display_results(self, scan_results: list, echo_passwords: bool) -> None:
+    def _display_results(self, scan_results: List, echo_passwords: bool) -> None:
         """Display scan results in a formatted way."""
         for result in scan_results:
             password, scan_result = result
@@ -523,7 +523,7 @@ def _get_status_text(self, scan_result) -> str:
         status_code = client_pb2.BWStatus.BREACHED if is_breached else client_pb2.BWStatus.GOOD
         return STATUS_TO_TEXT.get(status_code, "Unknown")
 
-    def _cleanup_scan_data(self, breach_watch, scan_results: list) -> None:
+    def _cleanup_scan_data(self, breach_watch, scan_results: List) -> None:
         """Clean up scan data by deleting EUIDs."""
         euids = self._extract_euids(scan_results)
         if euids:
@@ -532,7 +532,7 @@ def _cleanup_scan_data(self, breach_watch, scan_results: list) -> None:
             except Exception as e:
                 logger.warning(f"Failed to cleanup scan data: {e}")
 
-    def _extract_euids(self, scan_results: list) -> list:
+    def _extract_euids(self, scan_results: List) -> List:
         """Extract EUIDs from scan results for cleanup."""
         euids = []
         for result in scan_results:

keepercli-package/src/keepercli/commands/enterprise_info.py

@@ -137,7 +137,7 @@ def execute(self, context: KeeperParams, **kwargs):
                 teams[queued_team.node_id] = []
             queued_teams[queued_team.node_id].append(queued_team)
 
-        def tree_node(node: enterprise_types.Node) -> Tuple[str, Dict[str, dict]]:
+        def tree_node(node: enterprise_types.Node) -> Tuple[str, Dict[str, Dict]]:
             node_name = node.name
             if not node_name:
                 node_name = enterprise_data.enterprise_info.enterprise_name

keepercli-package/src/keepercli/commands/enterprise_push.py

@@ -3,7 +3,7 @@
 import os
 import copy
 import re
-from typing import Any
+from typing import Any, List, Set, Dict, Optional
 
 from . import base
 from .. import api
@@ -100,7 +100,7 @@ def load_template_records_from_file(file_path: str) -> list:
 TRANSFER_RECORD_SUCCESS = "transfer_record_success"
 
 
-def _substitute_value(value: str, values: dict[str, str]) -> str:
+def _substitute_value(value: str, values: Dict[str, str]) -> str:
     """Replace all ${key} placeholders in a string with values from the given dict."""
     result = value
     while True:
@@ -113,7 +113,7 @@ def _substitute_value(value: str, values: dict[str, str]) -> str:
     return result
 
 
-def _substitute_in_dict(container: dict, values: dict[str, str]) -> None:
+def _substitute_in_dict(container: Dict, values: Dict[str, str]) -> None:
     """Recursively substitute placeholders in dict (and nested dicts/lists) in place."""
     for key, val in list(container.items()):
         if isinstance(val, str):
@@ -126,7 +126,7 @@ def _substitute_in_dict(container: dict, values: dict[str, str]) -> None:
             container[key] = _substitute_in_list(val, values)
 
 
-def _substitute_in_list(container: list, values: dict[str, str]) -> list:
+def _substitute_in_list(container: list, values: Dict[str, str]) -> List:
     """Return a new list with placeholders substituted."""
     result = []
     for item in container:
@@ -142,7 +142,7 @@ def _substitute_in_list(container: list, values: dict[str, str]) -> list:
     return result
 
 
-def _get_substitution_values(enterprise: enterprise_types.IEnterpriseData, email: str) -> dict[str, str]:
+def _get_substitution_values(enterprise: enterprise_types.IEnterpriseData, email: str) -> Dict[str, str]:
     """Build substitution map for a user: user_email, user_name, generate_password."""
     values = {
         "user_email": email,
@@ -156,14 +156,14 @@ def _get_substitution_values(enterprise: enterprise_types.IEnterpriseData, email
 
 
 def _substitute_record_params(
-    enterprise: enterprise_types.IEnterpriseData, email: str, record_data: dict
+    enterprise: enterprise_types.IEnterpriseData, email: str, record_data: Dict
 ) -> None:
     """Fill template parameters in record_data for the given user (in place)."""
     values = _get_substitution_values(enterprise, email)
     _substitute_in_dict(record_data, values)
 
 
-def _resolve_user_to_email(enterprise: enterprise_types.IEnterpriseData, user_id: str) -> str | None:
+def _resolve_user_to_email(enterprise: enterprise_types.IEnterpriseData, user_id: str) -> Optional[str]:
     """Resolve user identifier (email, name, or enterprise_user_id) to username (email)."""
     user_id_lower = user_id.lower()
     for u in enterprise.users.get_all_entities():
@@ -176,7 +176,7 @@ def _resolve_user_to_email(enterprise: enterprise_types.IEnterpriseData, user_id
     return None
 
 
-def _resolve_team_to_uid(enterprise: enterprise_types.IEnterpriseData, team_id: str) -> str | None:
+def _resolve_team_to_uid(enterprise: enterprise_types.IEnterpriseData, team_id: str) -> Optional[str]:
     """Resolve team identifier (name or team_uid) to team_uid."""
     for t in enterprise.teams.get_all_entities() or []:
         if team_id == t.team_uid or team_id.lower() == t.name.lower():
@@ -187,9 +187,9 @@ def _resolve_team_to_uid(enterprise: enterprise_types.IEnterpriseData, team_id:
 def _collect_recipient_emails(
     enterprise: enterprise_types.IEnterpriseData,
     current_username: str,
-    user_ids: list[str],
-    team_ids: list[str],
-) -> set[str]:
+    user_ids: List[str],
+    team_ids: List[str],
+) -> Set[str]:
     """Resolve user_ids and team_ids to a set of recipient emails. Excludes current user."""
     emails = set()
 
@@ -231,8 +231,8 @@ def _collect_recipient_emails(
 def _build_typed_records_for_user(
     enterprise: enterprise_types.IEnterpriseData,
     email: str,
-    record_data: list[dict[str, Any]],
-) -> list[TypedRecord]:
+    record_data: List[Dict[str, Any]],
+) -> List[TypedRecord]:
     """Substitute template params and convert JSON templates to typed records."""
     user_records = []
     for template in record_data:
@@ -247,10 +247,10 @@ def _build_typed_records_for_user(
 def _build_records_add_request(
     auth: keeper_auth.KeeperAuth,
     vault: vault_online.VaultOnline,
-    typed_records: list[TypedRecord],
+    typed_records: List[TypedRecord],
     user_ec_key: Any,
     user_rsa_key: Any,
-    record_keys_out: dict[str, bytes],
+    record_keys_out: Dict[str, bytes],
 ) -> record_pb2.RecordsAddRequest:
     """Build RecordsAddRequest and fill record_keys_out with uid -> encrypted_key for transfer."""
     rq = record_pb2.RecordsAddRequest()
@@ -303,7 +303,7 @@ def _add_transfer_and_cleanup(
     auth: keeper_auth.KeeperAuth,
     email: str,
     add_request: record_pb2.RecordsAddRequest,
-    record_keys_for_user: dict[str, Any],
+    record_keys_for_user: Dict[str, Any],
 ) -> None:
     """Execute records_add, transfer ownership to user, then unlink from admin (pre_delete + delete)."""
     rs = auth.execute_auth_rest(
@@ -378,7 +378,7 @@ def _process_one_recipient(
     auth: keeper_auth.KeeperAuth,
     vault: vault_online.VaultOnline,
     email: str,
-    record_data: list[dict[str, Any]],
+    record_data: List[Dict[str, Any]],
 ) -> None:
     """Load user key, build records, add to vault, transfer ownership to user."""
     user_key = auth.get_user_keys(email)
@@ -428,9 +428,9 @@ def push_enterprise_records(
         enterprise: enterprise_types.IEnterpriseData,
         auth: keeper_auth.KeeperAuth,
         vault: vault_online.VaultOnline,
-        user_ids: list[str],
-        team_ids: list[str],
-        record_data: list[dict[str, Any]],
+        user_ids: List[str],
+        team_ids: List[str],
+        record_data: List[Dict[str, Any]],
     ) -> None:
         """Resolve recipients, then for each user substitute template params and add/transfer records."""
         emails = list(

keepercli-package/src/keepercli/commands/enterprise_team.py

@@ -1,6 +1,6 @@
 import argparse
 import json
-from typing import Dict, List, Optional, Any, Tuple
+from typing import Dict, List, Optional, Any, Tuple, Set
 
 from keepersdk import utils, crypto
 from keepersdk.enterprise import enterprise_types, batch_management, enterprise_management
@@ -573,7 +573,7 @@ def _build_user_approval_requests(
     def _collect_team_keys_and_users(
         self, queued_team_users, teams: Dict[str, Any], 
         added_teams: Dict[str, Any], active_users: Dict[int, str]
-    ) -> Tuple[Dict[str, Any], set]:
+    ) -> Tuple[Dict[str, Any], Set]:
         """Collect team UIDs that need keys loaded and all user emails."""
         team_keys = {}
         all_users = set()
@@ -595,7 +595,7 @@ def _collect_team_keys_and_users(
 
     def _load_team_and_user_keys(
         self, vault, team_keys: Dict[str, Any], 
-        added_team_keys: Dict[str, bytes], all_users: set
+        added_team_keys: Dict[str, bytes], all_users: Set
     ) -> None:
         """Load team keys and user public keys from the vault."""
         vault.keeper_auth.load_team_keys(list(team_keys.keys()))