black formatting

Author: whitfiea

src/mas/devops/backup.py

@@ -38,14 +38,14 @@ def copyContentsToYamlFile(file_path: str, content: dict) -> bool:
         # Create a custom dumper that uses literal style for multi-line strings
         class LiteralDumper(yaml.SafeDumper):
             pass
-        
+
         def str_representer(dumper, data):
             if '\n' in data:
                 return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')
             return dumper.represent_scalar('tag:yaml.org,2002:str', data)
-        
+
         LiteralDumper.add_representer(str, str_representer)
-        
+
         with open(file_path, 'w') as yaml_file:
             yaml.dump(content, yaml_file, default_flow_style=False, Dumper=LiteralDumper)
         return True
@@ -75,25 +75,25 @@ def filterResourceData(data: dict) -> dict:
 
     if 'status' in filteredCopy:
         del filteredCopy['status']
-    
+
     return filteredCopy
 
 
 def extract_secrets_from_dict(data, secret_names=None):
     """
     Recursively extract secret names from a dictionary structure.
     Looks for keys named 'secretName' and collects their values.
-    
+
     Args:
         data: Dictionary to search
         secret_names: Set to collect secret names (created if None)
-    
+
     Returns:
         Set of secret names found
     """
     if secret_names is None:
         secret_names = set()
-    
+
     if isinstance(data, dict):
         for key, value in data.items():
             # Check if this key is 'secretName' and has a string value
@@ -102,12 +102,12 @@ def extract_secrets_from_dict(data, secret_names=None):
             # Recursively search nested structures
             elif isinstance(value, (dict, list)):
                 extract_secrets_from_dict(value, secret_names)
-    
+
     elif isinstance(data, list):
         for item in data:
             if isinstance(item, (dict, list)):
                 extract_secrets_from_dict(item, secret_names)
-    
+
     return secret_names
 
 
@@ -116,26 +116,26 @@ def backupResources(dynClient: DynamicClient, namespace: str, kind: str, api_ver
     Backup resources of a given kind in a namespace.
     If name is provided, backs up that specific resource.
     If name is None, backs up all resources of that kind.
-    
+
     Args:
         dynClient: Kubernetes dynamic client
         namespace: Namespace to backup from
         kind: Resource kind (e.g., 'MongoCfg', 'Secret')
         api_version: API version (e.g., 'config.mas.ibm.com/v1')
         backup_path: Path to save backup files
         name: Optional specific resource name
-    
+
     Returns:
         tuple: (backed_up_count: int, not_found_count: int, failed_count: int, discovered_secrets: set)
     """
     discovered_secrets = set()
     backed_up_count = 0
     not_found_count = 0
     failed_count = 0
-    
+
     try:
         resourceAPI = dynClient.resources.get(api_version=api_version, kind=kind)
-        
+
         if name:
             # Backup specific named resource
             logger.info(f"Backing up {kind} '{name}' from namespace '{namespace}' (API version: {api_version})")
@@ -156,19 +156,19 @@ def backupResources(dynClient: DynamicClient, namespace: str, kind: str, api_ver
             logger.info(f"Backing up all {kind} resources from namespace '{namespace}' (API version: {api_version})")
             resources = resourceAPI.get(namespace=namespace)
             resources_to_process = resources.items
-        
+
         # Process each resource
         for resource in resources_to_process:
             resource_name = resource["metadata"]["name"]
             resource_dict = resource.to_dict()
-            
+
             # Extract secrets from this resource if it's not a Secret itself
             if kind != 'Secret':
                 secrets = extract_secrets_from_dict(resource_dict.get('spec', {}))
                 if secrets:
                     logger.info(f"Found {len(secrets)} secret reference(s) in {kind} '{resource_name}': {', '.join(sorted(secrets))}")
                     discovered_secrets.update(secrets)
-            
+
             # Backup the resource
             resource_file_path = f"{backup_path}/{resource_name}.yaml"
             filtered_resource = filterResourceData(resource_dict)
@@ -178,14 +178,14 @@ def backupResources(dynClient: DynamicClient, namespace: str, kind: str, api_ver
             else:
                 logger.error(f"Failed to back up {kind} '{resource_name}' to '{resource_file_path}'")
                 failed_count += 1
-        
+
         if backed_up_count > 0:
             logger.info(f"Successfully backed up {backed_up_count} {kind} resource(s)")
         elif not name:
             logger.info(f"No {kind} resources found in namespace '{namespace}'")
-        
+
         return (backed_up_count, not_found_count, failed_count, discovered_secrets)
-        
+
     except NotFoundError:
         if name:
             logger.info(f"{kind} '{name}' not found in namespace '{namespace}'")

test/src/test_backup.py

@@ -20,7 +20,7 @@ def test_create_single_directory(self, tmp_path):
         """Test creating a single backup directory"""
         test_dir = tmp_path / "backup1"
         result = createBackupDirectories([str(test_dir)])
-        
+
         assert result is True
         assert test_dir.exists()
         assert test_dir.is_dir()
@@ -34,7 +34,7 @@ def test_create_multiple_directories(self, tmp_path):
         ]
         paths = [str(d) for d in test_dirs]
         result = createBackupDirectories(paths)
-        
+
         assert result is True
         for test_dir in test_dirs:
             assert test_dir.exists()
@@ -44,7 +44,7 @@ def test_create_nested_directories(self, tmp_path):
         """Test creating nested backup directories"""
         nested_dir = tmp_path / "level1" / "level2" / "level3"
         result = createBackupDirectories([str(nested_dir)])
-        
+
         assert result is True
         assert nested_dir.exists()
         assert nested_dir.is_dir()
@@ -53,9 +53,9 @@ def test_create_existing_directory(self, tmp_path):
         """Test creating a directory that already exists"""
         test_dir = tmp_path / "existing"
         test_dir.mkdir()
-        
+
         result = createBackupDirectories([str(test_dir)])
-        
+
         assert result is True
         assert test_dir.exists()
 
@@ -67,18 +67,18 @@ def test_create_empty_list(self):
     def test_create_directory_permission_error(self, mocker):
         """Test handling of permission errors"""
         mock_makedirs = mocker.patch('os.makedirs', side_effect=PermissionError("Permission denied"))
-        
+
         result = createBackupDirectories(["/invalid/path"])
-        
+
         assert result is False
         mock_makedirs.assert_called_once()
 
     def test_create_directory_os_error(self, mocker):
         """Test handling of OS errors"""
         mocker.patch('os.makedirs', side_effect=OSError("OS error"))
-        
+
         result = createBackupDirectories(["/some/path"])
-        
+
         assert result is False
 
 
@@ -89,12 +89,12 @@ def test_write_simple_dict(self, tmp_path):
         """Test writing a simple dictionary to YAML file"""
         test_file = tmp_path / "test.yaml"
         content = {"key1": "value1", "key2": "value2"}
-        
+
         result = copyContentsToYamlFile(str(test_file), content)
-        
+
         assert result is True
         assert test_file.exists()
-        
+
         with open(test_file, 'r') as f:
             loaded_content = yaml.safe_load(f)
         assert loaded_content == content
@@ -110,9 +110,9 @@ def test_write_nested_dict(self, tmp_path):
             },
             "list": [1, 2, 3]
         }
-        
+
         result = copyContentsToYamlFile(str(test_file), content)
-        
+
         assert result is True
         with open(test_file, 'r') as f:
             loaded_content = yaml.safe_load(f)
@@ -122,9 +122,9 @@ def test_write_empty_dict(self, tmp_path):
         """Test writing an empty dictionary"""
         test_file = tmp_path / "empty.yaml"
         content = {}
-        
+
         result = copyContentsToYamlFile(str(test_file), content)
-        
+
         assert result is True
         with open(test_file, 'r') as f:
             loaded_content = yaml.safe_load(f)
@@ -135,14 +135,14 @@ def test_overwrite_existing_file(self, tmp_path):
         test_file = tmp_path / "overwrite.yaml"
         old_content = {"old": "data"}
         new_content = {"new": "data"}
-        
+
         # Write initial content
         with open(test_file, 'w') as f:
             yaml.dump(old_content, f)
-        
+
         # Overwrite with new content
         result = copyContentsToYamlFile(str(test_file), new_content)
-        
+
         assert result is True
         with open(test_file, 'r') as f:
             loaded_content = yaml.safe_load(f)
@@ -153,17 +153,17 @@ def test_write_to_nonexistent_directory(self, tmp_path):
         """Test writing to a file in a non-existent directory"""
         test_file = tmp_path / "nonexistent" / "test.yaml"
         content = {"key": "value"}
-        
+
         result = copyContentsToYamlFile(str(test_file), content)
-        
+
         assert result is False
 
     def test_write_permission_error(self, mocker):
         """Test handling of permission errors during write"""
         mocker.patch('builtins.open', side_effect=PermissionError("Permission denied"))
-        
+
         result = copyContentsToYamlFile("/invalid/path.yaml", {"key": "value"})
-        
+
         assert result is False
 
     def test_write_with_special_characters(self, tmp_path):
@@ -174,9 +174,9 @@ def test_write_with_special_characters(self, tmp_path):
             "unicode": "café ☕",
             "quotes": "value with 'quotes' and \"double quotes\""
         }
-        
+
         result = copyContentsToYamlFile(str(test_file), content)
-        
+
         assert result is True
         with open(test_file, 'r') as f:
             loaded_content = yaml.safe_load(f)
@@ -204,9 +204,9 @@ def test_filter_all_metadata_fields(self):
             },
             "spec": {"replicas": 3}
         }
-        
+
         result = filterResourceData(data)
-        
+
         assert "name" in result["metadata"]
         assert "namespace" in result["metadata"]
         assert "annotations" not in result["metadata"]
@@ -228,9 +228,9 @@ def test_filter_status_field(self):
                 "conditions": []
             }
         }
-        
+
         result = filterResourceData(data)
-        
+
         assert "status" not in result
         assert "spec" in result
         assert "metadata" in result
@@ -244,9 +244,9 @@ def test_filter_partial_metadata(self):
                 "labels": {"app": "test"}
             }
         }
-        
+
         result = filterResourceData(data)
-        
+
         assert "name" in result["metadata"]
         assert "labels" in result["metadata"]
         assert "uid" not in result["metadata"]
@@ -258,9 +258,9 @@ def test_filter_no_metadata(self):
             "kind": "Resource",
             "spec": {"replicas": 3}
         }
-        
+
         result = filterResourceData(data)
-        
+
         assert "metadata" not in result
         assert "spec" in result
         assert "apiVersion" in result
@@ -271,9 +271,9 @@ def test_filter_empty_metadata(self):
             "metadata": {},
             "spec": {"replicas": 3}
         }
-        
+
         result = filterResourceData(data)
-        
+
         assert "metadata" in result
         assert result["metadata"] == {}
 
@@ -291,9 +291,9 @@ def test_filter_preserves_other_fields(self):
                 "key2": "value2"
             }
         }
-        
+
         result = filterResourceData(data)
-        
+
         assert result["apiVersion"] == "v1"
         assert result["kind"] == "ConfigMap"
         assert result["data"] == {"key1": "value1", "key2": "value2"}
@@ -308,14 +308,14 @@ def test_filter_shallow_copy_behavior(self):
             },
             "status": {"phase": "Running"}
         }
-        
+
         result = filterResourceData(data)
-        
+
         # Due to shallow copy, nested metadata dict is modified in original
         # but top-level status is not (it's deleted from copy only)
         assert "uid" not in data["metadata"]  # Modified due to shallow copy
         assert "status" in data  # Not modified (top-level key)
-        
+
         # Result should not have uid and status
         assert "uid" not in result["metadata"]
         assert "status" not in result
@@ -345,17 +345,17 @@ def test_filter_complex_resource(self):
                 "readyReplicas": 3
             }
         }
-        
+
         result = filterResourceData(data)
-        
+
         # Check preserved fields
         assert result["apiVersion"] == "apps/v1"
         assert result["kind"] == "Deployment"
         assert result["metadata"]["name"] == "my-deployment"
         assert result["metadata"]["namespace"] == "default"
         assert result["metadata"]["labels"] == {"app": "myapp"}
         assert result["spec"]["replicas"] == 3
-        
+
         # Check removed fields
         assert "annotations" not in result["metadata"]
         assert "creationTimestamp" not in result["metadata"]