Fix unit tests

Author: sk-keeper

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):

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/record_type_utils.py

@@ -1,4 +1,5 @@
 import json
+from typing import List
 
 from .. import api
 
@@ -99,7 +100,7 @@ def get_record_type_example(vault: vault_online.VaultOnline, record_type_name: s
         raise ValueError(f'No record type found with name {record_type_name}. Use "record-type-info" to list all record types')
 
 
-def get_record_types(vault:vault_online.VaultOnline) -> list[vault_types.RecordType]:
+def get_record_types(vault:vault_online.VaultOnline) -> List[vault_types.RecordType]:
         records = []  # (recordTypeId, name, scope)
         record_types = vault.vault_data.get_record_types()
 

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

@@ -1,7 +1,7 @@
 import argparse
 from enum import Enum
 import time
-from typing import Optional
+from typing import Optional, List, Set, Tuple
 
 from keepersdk import utils
 from keepersdk.proto.enterprise_pb2 import GENERAL
@@ -454,7 +454,7 @@ def _confirm_remove_all_clients(clients_count: int) -> bool:
         return user_choice.lower() == USER_CHOICE_YES
 
     @staticmethod
-    def remove_client(vault: vault_online.VaultOnline, uid: str, client_names_and_ids: list[str], force: bool = False):
+    def remove_client(vault: vault_online.VaultOnline, uid: str, client_names_and_ids: List[str], force: bool = False):
         """Remove client devices from a KSM application."""
         ksm_management.KSMClientManagement.remove_clients_from_ksm_app(
             vault=vault, 
@@ -539,7 +539,7 @@ def _get_app_uid_from_kwargs(self, vault, app_uid_or_name: Optional[str]) -> str
 
         return ksm_app.record_uid
 
-    def _parse_secret_uids(self, secret_uids_str: Optional[str]) -> list[str]:
+    def _parse_secret_uids(self, secret_uids_str: Optional[str]) -> List[str]:
         """Parse secret UIDs from string."""
         if not secret_uids_str:
             return []
@@ -552,7 +552,7 @@ def _find_ksm_application(self, vault: vault_online.VaultOnline, app_uid_or_name
             None
         )
 
-    def _handle_add_share(self, context: KeeperParams, app_uid: str, secret_uids: list[str], is_editable: bool) -> None:
+    def _handle_add_share(self, context: KeeperParams, app_uid: str, secret_uids: List[str], is_editable: bool) -> None:
         """Handle adding shares to a KSM application."""
         if not context.vault:
             raise ValueError("Vault is not initialized.")
@@ -600,7 +600,7 @@ def _handle_share_error(kae: base.errors.KeeperApiError) -> bool:
             raise ValueError(f"Failed to share secrets: {kae}")
 
     @staticmethod
-    def remove_share(vault: vault_online.VaultOnline, app_uid: str, secret_uids: list[str]) -> None:
+    def remove_share(vault: vault_online.VaultOnline, app_uid: str, secret_uids: List[str]) -> None:
         """Remove shares from a KSM application."""
         if not secret_uids:
             logger.warning("No secret UIDs provided for removal.")
@@ -623,7 +623,7 @@ def remove_share(vault: vault_online.VaultOnline, app_uid: str, secret_uids: lis
         logger.info("Shared secrets were successfully removed from the application\n")
 
     @staticmethod
-    def _get_current_shared_uids(vault: vault_online.VaultOnline, app_uid: str) -> set:
+    def _get_current_shared_uids(vault: vault_online.VaultOnline, app_uid: str) -> Set:
         """Get currently shared UIDs for the application."""
         app_infos = ksm_management.get_app_info(vault=vault, app_uid=app_uid)
         if not app_infos:
@@ -636,14 +636,14 @@ def _get_current_shared_uids(vault: vault_online.VaultOnline, app_uid: str) -> s
         }
 
     @staticmethod
-    def _validate_share_uids(secret_uids: list[str], current_shared_uids: set) -> tuple[list, list]:
+    def _validate_share_uids(secret_uids: List[str], current_shared_uids: Set) -> Tuple[List, List]:
         """Validate secret UIDs against currently shared UIDs."""
         valid_uids = [uid for uid in secret_uids if uid in current_shared_uids]
         invalid_uids = [uid for uid in secret_uids if uid not in current_shared_uids]
         return valid_uids, invalid_uids
 
     @staticmethod
-    def _log_invalid_uids(invalid_uids: list[str]) -> None:
+    def _log_invalid_uids(invalid_uids: List[str]) -> None:
         """Log warnings for invalid UIDs."""
         for uid in invalid_uids:
             logger.warning(f"Secret UID '{uid}' is not shared with this application. Skipping.")

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

@@ -3,7 +3,7 @@
 import math
 import re
 from enum import Enum
-from typing import Optional
+from typing import Optional, List, Set
 
 from keepersdk import utils
 from keepersdk.authentication import keeper_auth
@@ -132,7 +132,7 @@ def execute(self, context: KeeperParams, **kwargs) -> None:
         if not context.vault:
             raise ValueError("Vault is not initialized.")
         vault = context.vault
-        
+
         uid_or_name = kwargs.get('record')
         if not uid_or_name:
             return self.get_parser().print_help()
@@ -154,7 +154,7 @@ def execute(self, context: KeeperParams, **kwargs) -> None:
         share_expiration = share_management_utils.get_share_expiration(
             kwargs.get('expire_at'), kwargs.get('expire_in')
         )
-        
+
         request = RecordShares.prep_request(
             vault=vault, 
             enterprise=context.enterprise_data,
@@ -403,7 +403,7 @@ def _get_share_admin_obj_uids(self, vault: vault_online.VaultOnline,
         except (ValueError, AttributeError) as e:
             raise ValueError(f'get_share_admin: msg = {e}') from e
 
-    def _get_record_uids(self, vault: vault_online.VaultOnline, name: str) -> set[str]:
+    def _get_record_uids(self, vault: vault_online.VaultOnline, name: str) -> Set[str]:
         """Get record UIDs by name or UID."""
         record_uids = set()
         if not vault or not vault.vault_data:
@@ -1042,7 +1042,7 @@ def _find_client_id(self, applications, share_name: str) -> Optional[bytes]:
 
         return self._resolve_partial_matches(partial_matches, share_name)
 
-    def _resolve_partial_matches(self, partial_matches: list[bytes], original_name: str) -> Optional[bytes]:
+    def _resolve_partial_matches(self, partial_matches: List[bytes], original_name: str) -> Optional[bytes]:
         """
         Resolve partial matches to a single client ID.