src/cmd-koji-upload: port to Python3

src/cmd-koji-upload

@@ -1,20 +1,11 @@
-#!/usr/bin/python -u
-# pylint: disable=C0103
-
+#!/usr/bin/python3 -u
 """
 cmd-koji-upload performs the required steps to make COSA a Koji Content
 Generator. When running this in an automated fashion, you will need a Kerberos
 keytab file for a service account.
 
-Python2: due to an issue with Kerberos Auth issue which returns errors like:
-[auth_gssapi:error] NO AUTH DATA Client did not send any authentication headers
-[auth_gssapi:error] GSS ERROR gss_localname() failed:...
-We are unable to use Python3. After inquiring with the Koji team, we were
-advised to use Python2. This code should port rather cleanly to Python3 once
-the Kerberos issue is cleared.
-
 Dependencies:
- - python-koji, krb5-workstation and python-krbV
+ - python3-koji, koji-utils,  krb5-workstation and python3-koji-cli-plugins
  - Kerberos credentials against Koji _in_ keytab format
  - content generator permissions for your user or service account
 
@@ -36,7 +27,7 @@ import sys
 import tempfile
 
 cosa_dir = os.path.dirname(os.path.abspath(__file__))
-sys.path.insert(0, ("%s/cosalib" % cosa_dir))
+sys.path.insert(0, f"{cosa_dir}/cosalib")
 sys.path.insert(0, cosa_dir)
 
 try:
@@ -46,14 +37,6 @@ except ImportError:
     sys.path.insert(0, '/usr/lib/coreos-assembler/cosalib')
     from cosalib.build import _Build
 
-try:
-    # Available in Koji v1.17, https://pagure.io/koji/issue/975
-    from koji_cli.lib import unique_path
-    from koji_cli.lib import progress_callback as cb
-except ImportError:
-    from koji_cli.lib import _unique_path as unique_path
-    from koji_cli.lib import _progress_callback as cb
-
 try:
     with open("/etc/redhat-release", "r") as f:
         HOST_OS = str(f.read()).strip()
@@ -130,8 +113,8 @@ class Build(_Build):
         try:
             shutil.rmtree(self._tmpdir)
         except Exception as e:
-            raise Exception("failed to remove temporary directory: %s",
-                            self._tmpdir, e)
+            raise Exception(
+                f"failed to remove temporary directory: {self._tmpdir}", e)
 
     def rename_mutator(self, fname):
         """
@@ -144,11 +127,11 @@ class Build(_Build):
         """
         for ending in RENAME_RAW:
             if fname.endswith(ending):
-                lname = "%s.raw" % fname
+                lname = f"{fname}.raw"
                 try:
                     os.symlink(fname, lname)
                 except Exception as e:
-                    raise Exception("failed create symlink: %s" % e)
+                    raise Exception("failed create symlink for {lname}", e)
 
                 return lname, True
         return fname, None
@@ -171,7 +154,7 @@ class Build(_Build):
             base_name = os.path.basename(fname)
             base = os.path.splitext(base_name)[0]
             ftype = (os.path.splitext(base)[1]).replace(".", "")
-            ctype = "%s.%s" % (ftype, x)
+            ctype = f"{ftype}.{x}"
             if ctype not in KOJI_CG_TYPES and ftype in KOJI_CG_TYPES:
                 if x == "gz":
                     compressor = gzip
@@ -182,12 +165,13 @@ class Build(_Build):
                 try:
                     infile = compressor.open(fname)
                     outfile = open(new_path, 'wb+')
-                    log.info("using %s module to mutate %s to %s",
-                             compressor.__name__, base_name, new_path)
+                    log.info((f"using {compressor.__name__} module to mutate",
+                              f"{base_name} to {new_path}"))
                     shutil.copyfileobj(infile, outfile)
                 except Exception as e:
-                    raise Exception("failed to decompress file %s to %s: %s",
-                                    fname, new_path, e)
+                    raise Exception(
+                        f"failed to decompress file {fname} to {new_path}", e)
+
                 finally:
                     infile.close()
                     outfile.close()
@@ -251,8 +235,8 @@ class Build(_Build):
         for ffile in os.listdir(self.build_dir):
             files.append(os.path.join(self.build_dir, ffile))
             if os.path.islink(ffile):
-                log.debug(" * EXCLUDING symlink '%s'", ffile)
-                log.debug(" *    target '%s'", os.path.realpath(ffile))
+                log.debug(f" * EXCLUDING symlink {ffile}")
+                log.debug(f" *    target {os.path.realpath(ffile)}")
                 continue
 
         # add the coreos assembler files
@@ -262,33 +246,32 @@ class Build(_Build):
         # process the files that were found
         for ffile in files:
             lpath = os.path.abspath(ffile)
-            log.debug("Considering file file '%s'", lpath)
+            log.debug(f"Considering file file {lpath}")
 
             # if the file is mutated (renamed, uncompressed, etc)
             # we want to use that file name instead
             mutated_path = self.mutate_for_koji(lpath)
             if mutated_path != lpath:
                 lpath = mutated_path
-            log.debug(" * using %s for upload name", lpath)
+            log.debug(f" * using {lpath} for upload name")
 
             # and check that a file should be uploaded
             upload_path = os.path.basename(lpath)
             if not self.supported_upload(lpath):
-                log.debug(" * EXCLUDING file '%s'", ffile)
+                log.debug(f" * EXCLUDING file {ffile}")
                 log.debug("   File type is not supported by Koji")
                 continue
 
-            # os.path.getsize uses 1kB instead of 1KB. So we use stat instead.
-            fsize = subprocess.check_output(["stat", "--dereference",  "--format", '%s', lpath])
+            fsize = '{}'.format(os.stat(lpath).st_size)
             log.debug(" * calculating checksum")
             self._found_files[lpath] = {
                 "local_path": lpath,
                 "upload_path": upload_path,
                 "md5": md5sum_file(lpath),
                 "size": int(fsize)
             }
-            log.debug(" * size is %s", self._found_files[lpath]["size"])
-            log.debug(" * md5 is %s", self._found_files[lpath]["md5"])
+            log.debug(f" * size is {self._found_files[lpath]['size']}")
+            log.debug(f" * md5 is {self._found_files[lpath]['md5']}")
 
 
 def set_logger(level):
@@ -320,12 +303,12 @@ def kinit(keytab, principle):
     if keytab is None:
         raise Exception("keytab file is not defined")
 
-    log.info("using kerberos auth via %s", keytab)
+    log.info(f"using kerberos auth via {keytab}")
     try:
         _ = subprocess.check_output([
             "kinit", "-f", "-t", keytab, "-k", principle])
         klist_out = subprocess.check_output(["klist", "-a"])
-        log.debug("authenticated: \n%s", klist_out.decode("utf-8"))
+        log.debug(f"authenticated: \n{(klist_out.decode('utf-8'))}")
     except Exception as err:
         raise Exception("failed to auth: ", err)
 
@@ -403,10 +386,10 @@ class Upload():
         if ext in COMPRESSION_EXT:
             # find sub extension, e.g. "tar" in "tar.gz"
             sub_ext = os.path.splitext(obj.get("upload_path"))[0].lstrip('.')
-            ext = "%s.%s" % (sub_ext, ext)
+            ext = f"{sub_ext}.{ext}"
 
-        log.debug("File %s should be of type %s: %s ", obj.get("upload_path"),
-                  ext, obj)
+        log.debug((f"File {(obj.get('upload_path'))} should be of",
+                   f"type {ext}: {obj}"))
         (description, ext, arch, etype) = KOJI_CG_TYPES.get(
             ext, [None, None, None, None])
 
@@ -458,7 +441,7 @@ class Upload():
         now = datetime.datetime.utcnow()
         stamp = now.strftime("%s")
 
-        log.debug("Preparing manifest for %s files", len(self.image_files))
+        log.debug(f"Preparing manifest for {(len(self.image_files))} files")
         self._manifest = {
             "metadata_version": 0,
             "build": {
@@ -475,7 +458,7 @@ class Upload():
                 "owner": self._owner,
                 "source": source['origin'],
                 "start_time": stamp,
-                "version": "%s" % self.build.build_id
+                "version": f"{self.build.build_id}"
             },
             "buildroots": [{
                 "id": 1,
@@ -533,7 +516,7 @@ class Upload():
             assert session.logged_in
             log.info("logged into koji server")
         except Exception as e:
-            raise Exception("failed to authenticate to koji: %s" % e)
+            raise Exception(f"failed to authenticate to koji: {e}")
 
         if session is None:
             raise Exception("failed to get session from koji server")
@@ -545,7 +528,7 @@ class Upload():
         """ Verify that a tag exists in this Koji instance """
         taginfo = self._session.getTag(tag)
         if not taginfo:
-            raise RuntimeError('tag %s is not present in Koji' % tag)
+            raise RuntimeError(f"tag {tag} is not present in Koji")
 
     def upload(self):
         """
@@ -558,15 +541,13 @@ class Upload():
         a symlink is created in temporary directory and is used as the upload
         file.
         """
-        serverdir = unique_path("%s-cosa" % self.build.build_id)
-        callback = cb
-
-        log.debug('uploading files to %s', serverdir)
+        serverdir = klib.unique_path(f"{self.build.build_id}-cosa")
+        log.debug(f"uploading files to {serverdir}")
         for _, meta in (self.build).get_artifacts():
             local_path = meta['local_path']    # the local file to upload
             remote_path = meta['upload_path']  # the name of the file to upload
-            log.info("Uploading %s to %s/%s", local_path, serverdir,
-                     remote_path)
+            log.info(f"Uploading {local_path} to {serverdir}/{remote_path}")
+            callback = klib._progress_callback
             self.session.uploadWrapper(local_path, serverdir, name=remote_path,
                                        callback=callback)
             if callback:
@@ -576,7 +557,7 @@ class Upload():
         self._remote_directory = serverdir
         cginfo = self.session.CGImport(self.manifest, serverdir)
         log.info(json.dumps(cginfo, sort_keys=True, indent=3))
-        log.info("recorded build %s", cginfo['nvr'])
+        log.info(f"recorded build {cginfo['nvr']}")
         return cginfo
 
 
@@ -600,7 +581,7 @@ To use this program, you will need:
 
 Example:
     $ cmd-koji-upload \
-        --build_root=/src/build \
+        --buildroot=/src/build \
         --keytab keytab \
         --owner me@FEDORA.COM \
         --tag rhaos-4.1-rhel-8-build \
@@ -654,7 +635,7 @@ Environment variables are supported:
     build.build_artifacts()
     upload = Upload(build, args.owner, args.tag, args.profile)
     if args.dump:
-        print(json.dumps(upload.manifest, sort_keys=True, indent=3))
+        print((json.dumps(upload.manifest, sort_keys=True, indent=3)))
         return
 
     if args.no_upload is False:

src/cmdlib.py

@@ -101,8 +101,7 @@ def fatal(msg):
     :type msg: str
     :raises: SystemExit
     """
-    print('fatal: {}'.format(msg), file=sys.stderr)
-    raise SystemExit(1)
+    raise SystemExit(msg)
 
 
 def info(msg):
@@ -112,7 +111,7 @@ def info(msg):
     :param msg: The message to show to output
     :type msg: str
     """
-    print('info: {}'.format(msg), file=sys.stderr)
+    sys.stderr.write(f"info: {msg}")
 
 
 def rfc3339_time(t=None):

src/deps.txt

@@ -56,4 +56,4 @@ ShellCheck
 python3-flake8 python3-pytest python3-pytest-cov
 
 # Support for Koji uploads.
-krb5-libs krb5-workstation python2-koji python2-krbv
+krb5-libs krb5-workstation koji-utils python3-koji python3-koji-cli-plugins

tests/test_cmdlib.py

@@ -61,11 +61,11 @@ def test_fatal(capsys):
     Ensure that fatal does indeed attempt to exit
     """
     test_string = str(uuid.uuid4())
-    with pytest.raises(SystemExit):
+    err = None
+    with pytest.raises(SystemExit) as err:
         cmdlib.fatal(test_string)
     # Check that our test string is in stderr
-    captured = capsys.readouterr()
-    assert test_string in captured.err
+    assert test_string in str(err)
 
 
 def test_info(capsys):