Fix Pretest Static Analysis found formats

Author: sg893052

sonic_installer/main.py

@@ -375,7 +375,7 @@ def migrate_sonic_packages(bootloader, binary_image_version):
             docker_started = True
             run_command_or_raise(["cp", packages_path, os.path.join(new_image_mount, TMP_DIR, packages_file)])
             run_command_or_raise(["mkdir", "-p", custom_manifests_path])
-            run_command_or_raise(["cp", "-arf", custom_manifests_path , new_image_package_directory_path])
+            run_command_or_raise(["cp", "-arf", custom_manifests_path, new_image_package_directory_path])
             run_command_or_raise(["touch", os.path.join(new_image_mount, "tmp", DOCKERD_SOCK)])
             run_command_or_raise(["mount", "--bind",
                                 os.path.join(VAR_RUN_PATH, DOCKERD_SOCK),

sonic_package_manager/main.py

@@ -9,10 +9,6 @@
 import click
 import click_log
 import tabulate
-from urllib.parse import urlparse
-import paramiko
-import requests
-import getpass
 from natsort import natsorted
 
 from sonic_package_manager.database import PackageEntry, PackageDatabase
@@ -166,7 +162,7 @@ def repository(ctx):
 @click.pass_context
 def manifests(ctx):
     """ Custom local Manifest management commands. """
-    
+
     pass
 
 @cli.group()
@@ -292,13 +288,12 @@ def changelog(ctx,
         exit_cli(f'Failed to print package changelog: {err}', fg='red')
 
 
-
 @manifests.command('create')
 @click.pass_context
 @click.argument('name', type=click.Path())
 @click.option('--from-json', type=str, help='specify manifest json file')
 @root_privileges_required
-def create(ctx, name, from_json):
+def create_manifest(ctx, name, from_json):
     """Create a new custom local manifest file."""
 
     manager: PackageManager = ctx.obj
@@ -307,17 +302,14 @@ def create(ctx, name, from_json):
     except Exception as e:
         click.echo("Error: Manifest {} creation failed - {}".format(name, str(e)))
         return
-        
-#At the end of sonic-package-manager install, a new manifest file is created with the name.
-#At the end of sonic-package-manager uninstall name, this manifest file name and name.edit will be deleted.
-#At the end of sonic-package-manager update, we need to mv maniests name.edit to name in case of success, else keep it as such.
-#So during sonic-package-manager update, we could take old package from name and new package from edit and at the end, follow 3rd point
+
+
 @manifests.command('update')
 @click.pass_context
 @click.argument('name', type=click.Path())
 @click.option('--from-json', type=str, required=True)
 @root_privileges_required
-def update(ctx, name, from_json):
+def update_manifest(ctx, name, from_json):
     """Update an existing custom local manifest file with new one."""
 
     manager: PackageManager = ctx.obj
@@ -327,11 +319,12 @@ def update(ctx, name, from_json):
         click.echo(f"Error occurred while updating manifest '{name}': {e}")
         return
 
+
 @manifests.command('delete')
 @click.pass_context
 @click.argument('name', type=click.Path())
 @root_privileges_required
-def delete(ctx, name):
+def delete_manifest(ctx, name):
     """Delete a custom local manifest file."""
     manager: PackageManager = ctx.obj
     try:
@@ -352,6 +345,7 @@ def show_manifest(ctx, name):
     except FileNotFoundError:
         click.echo("Manifest file '{}' not found.".format(name))
 
+
 @manifests.command('list')
 @click.pass_context
 @root_privileges_required
@@ -421,8 +415,8 @@ def remove(ctx, name):
               help='Use locally created custom manifest file. ',
               hidden=True)
 @click.option('--name',
-                 type=str,
-                 help='custom name for the package')
+              type=str,
+              help='custom name for the package')
 @add_options(PACKAGE_SOURCE_OPTIONS)
 @add_options(PACKAGE_COMMON_OPERATION_OPTIONS)
 @add_options(PACKAGE_COMMON_INSTALL_OPTIONS)
@@ -471,13 +465,13 @@ def install(ctx,
 
     if use_local_manifest:
         if not name:
-            click.echo(f'name argument is not provided to use local manifest')
+            click.echo('name argument is not provided to use local manifest')
             return
         original_file = os.path.join(MANIFESTS_LOCATION, name)
         if not os.path.exists(original_file):
             click.echo(f'Local Manifest file for {name} does not exists to install')
             return
-        
+
     try:
         manager.install(package_expr,
                         from_repository,
@@ -490,6 +484,14 @@ def install(ctx,
     except KeyboardInterrupt:
         exit_cli('Operation canceled by user', fg='red')
 
+# At the end of sonic-package-manager install, a new manifest file is created with the name.
+# At the end of sonic-package-manager uninstall name,
+# this manifest file name and name.edit will be deleted.
+# At the end of sonic-package-manager update,
+# we need to mv maniests name.edit to name in case of success, else keep it as such.
+# So during sonic-package-manager update,
+# we could take old package from name and new package from edit and at the end, follow 3rd point
+
 
 @cli.command()
 @add_options(PACKAGE_COMMON_OPERATION_OPTIONS)

sonic_package_manager/manager.py

@@ -67,14 +67,13 @@
 )
 import click
 import json
-import os
 import requests
 import getpass
 import paramiko
 import urllib.parse
 from scp import SCPClient
 from sonic_package_manager.manifest import Manifest, MANIFESTS_LOCATION, DEFAULT_MANIFEST_FILE
-LOCAL_JSON="/tmp/local_json"
+LOCAL_JSON = "/tmp/local_json"
 
 @contextlib.contextmanager
 def failure_ignore(ignore: bool):
@@ -458,9 +457,9 @@ def install_from_source(self,
         self.database.commit()
 
     @under_lock
-    def update(self, 
-                name: str,
-                **kwargs):
+    def update(self,
+               name: str,
+               **kwargs):
         """ Update SONiC Package referenced by name. The update
         can be forced if force argument is True.
 
@@ -481,7 +480,6 @@ def update(self,
             click.echo("Package {} is not installed".format(name))
             return
 
-    
     def remove_unused_docker_image(self, package):
         image_id_used = any(entry.image_id == package.image_id for entry in self.database if entry.name != package.name)
         if not image_id_used:
@@ -579,7 +577,7 @@ def upgrade_from_source(self,
         Raises:
             PackageManagerError
         """
-    
+
         if update_only:
             new_package = self.get_installed_package(name, use_edit=True)
         else:
@@ -679,10 +677,10 @@ def upgrade_from_source(self,
                 if not skip_host_plugins:
                     self._install_cli_plugins(new_package)
                     exits.callback(rollback(self._uninstall_cli_plugin, new_package))
-                
+
                 if old_package.image_id != new_package.image_id:
                     self.remove_unused_docker_image(old_package)
-                
+
                 exits.pop_all()
         except Exception as err:
             raise PackageUpgradeError(f'Failed to upgrade {new_package.name}: {err}')
@@ -697,7 +695,7 @@ def upgrade_from_source(self,
         if update_only:
             manifest_path = os.path.join(MANIFESTS_LOCATION, name)
             edit_path = os.path.join(MANIFESTS_LOCATION, name + ".edit")
-            os.rename(edit_path,manifest_path)
+            os.rename(edit_path, manifest_path)
 
     @under_lock
     @opt_check
@@ -820,9 +818,9 @@ def migrate_package(old_package_entry,
                     new_package.version = old_package.version
                     migrate_package(old_package, new_package)
                 else:
-                    #self.install(f'{new_package.name}={new_package_default_version}')
-                    repo_tag_formed="{}:{}".format(new_package.repository, new_package.default_reference)
-                    self.install(None,repo_tag_formed,name=new_package.name)
+                    # self.install(f'{new_package.name}={new_package_default_version}')
+                    repo_tag_formed = "{}:{}".format(new_package.repository, new_package.default_reference)
+                    self.install(None, repo_tag_formed, name=new_package.name)
             else:
                 # No default version and package is not installed.
                 # Migrate old package same version.
@@ -1108,12 +1106,12 @@ def download_file(self, url, local_path):
         remote_path = parsed_url.path
         supported_protocols = ['http', 'https', 'scp', 'sftp']
 
-        #clear the temporary local file
+        # clear the temporary local file
         if os.path.exists(local_path):
             os.remove(local_path)
 
         if not protocol:
-            #check for local file
+            # check for local file
             if os.path.exists(url):
                 os.rename(url, local_path)
                 return True
@@ -1191,7 +1189,6 @@ def create_package_manifest(self, name, from_json):
             click.echo("Error: Manifest file '{}' already exists.".format(name))
             return
 
-
         if from_json:
             ret = self.download_file(from_json, LOCAL_JSON)
             if ret is False:
@@ -1202,18 +1199,17 @@ def create_package_manifest(self, name, from_json):
         data = {}
         with open(from_json, 'r') as file:
             data = json.load(file)
-            #Validate with manifest scheme
+            # Validate with manifest scheme
             Manifest.marshal(data)
 
-            #Make sure the 'name' is overwritten into the dict
+            # Make sure the 'name' is overwritten into the dict
             data['package']['name'] = name
             data['service']['name'] = name
 
             with open(mfile_name, 'w') as file:
                 json.dump(data, file, indent=4)
         click.echo(f"Manifest '{name}' created successfully.")
 
-
     def update_package_manifest(self, name, from_json):
         if name == "default_manifest":
             click.echo("Default Manifest updation is not allowed")
@@ -1223,7 +1219,7 @@ def update_package_manifest(self, name, from_json):
         if not os.path.exists(original_file):
             click.echo(f'Local Manifest file for {name} does not exists to update')
             return
-        #download json file from remote/local path
+        # download json file from remote/local path
         ret = self.download_file(from_json, LOCAL_JSON)
         if ret is False:
             return
@@ -1232,10 +1228,10 @@ def update_package_manifest(self, name, from_json):
         with open(from_json, 'r') as file:
             data = json.load(file)
 
-        #Validate with manifest scheme
+        # Validate with manifest scheme
         Manifest.marshal(data)
 
-        #Make sure the 'name' is overwritten into the dict
+        # Make sure the 'name' is overwritten into the dict
         data['package']['name'] = name
         data['service']['name'] = name
 
@@ -1246,8 +1242,8 @@ def update_package_manifest(self, name, from_json):
                 json.dump(data, edit_file, indent=4)
             click.echo(f"Manifest '{name}' updated successfully.")
         else:
-            #If package is not installed,
-            ## update the name file directly
+            # If package is not installed,
+            # update the name file directly
             with open(original_file, 'w') as orig_file:
                 json.dump(data, orig_file, indent=4)
             click.echo(f"Manifest '{name}' updated successfully.")

sonic_package_manager/manifest.py

@@ -289,4 +289,3 @@ def get_manifest_from_local_file(name):
                 }
             }
         return desired_dict
-

sonic_package_manager/metadata.py

@@ -5,7 +5,6 @@
 import json
 import tarfile
 from typing import Dict, List, Optional
-import os
 from sonic_package_manager import utils
 from sonic_package_manager.errors import MetadataError
 from sonic_package_manager.logger import log
@@ -64,7 +63,8 @@ def __init__(self, docker, registry_resolver):
         self.docker = docker
         self.registry_resolver = registry_resolver
 
-    def from_local(self, image: str, use_local_manifest: bool = False, name: Optional[str] = None, use_edit: bool = False) -> Metadata:
+    def from_local(self, image: str, use_local_manifest: bool = False,
+                   name: Optional[str] = None, use_edit: bool = False) -> Metadata:
         """ Reads manifest from locally installed docker image.
 
         Args:
@@ -88,7 +88,7 @@ def from_local(self, image: str, use_local_manifest: bool = False, name: Optiona
             if name:
                 labels = Manifest.get_manifest_from_local_file(name)
                 if labels is None:
-                    raise MetadataError('No manifest found in image labels')    
+                    raise MetadataError('No manifest found in image labels')
             else:
                 raise MetadataError('No manifest found in image labels')
 
@@ -125,7 +125,7 @@ def from_registry(self,
                 raise MetadataError('The name(custom) option is required as there is no metadata found in image labels')
             labels = Manifest.get_manifest_from_local_file(name)
         if labels is None:
-            raise MetadataError('No manifest found in image labels')    
+            raise MetadataError('No manifest found in image labels')
         return self.from_labels(labels)
 
     def from_tarball(self, image_path: str, use_local_manifest: bool = False, name: Optional[str] = None) -> Metadata:
@@ -149,10 +149,11 @@ def from_tarball(self, image_path: str, use_local_manifest: bool = False, name:
             labels = image_config['config'].get('Labels')
             if labels is None or len(labels) == 0 or 'com.azure.sonic.manifest' not in labels:
                 if name is None:
-                    raise MetadataError('The name(custom) option is required as there is no metadata found in image labels')
+                    raise MetadataError('The name(custom) option is \
+                            required as there is no metadata found in image labels')
                 labels = Manifest.get_manifest_from_local_file(name)
                 if labels is None:
-                    raise MetadataError('No manifest found in image labels')    
+                    raise MetadataError('No manifest found in image labels')
             return self.from_labels(labels)
 
     @classmethod
@@ -202,5 +203,5 @@ def from_labels(cls, labels: Dict[str, str]) -> Metadata:
             log.debug(f"Found YANG modules: {labels_yang_modules.keys()}")
         else:
             log.debug("No YANG modules found")
-    
+
         return Metadata(Manifest.marshal(manifest_dict), components, yang_modules)

sonic_package_manager/source.py

@@ -117,7 +117,9 @@ def __init__(self,
 
     def get_metadata(self) -> Metadata:
         """ Returns manifest read from tarball. """
-        return self.metadata_resolver.from_tarball(self.tarball_path, use_local_manifest=self.use_local_manifest, name=self.name)
+        return self.metadata_resolver.from_tarball(self.tarball_path,
+                                                   use_local_manifest=self.use_local_manifest,
+                                                   name=self.name)
 
     def install_image(self, package: Package):
         """ Installs image from local tarball source. """