Code style, typing, tmt.Path usage

src/api.py

@@ -44,7 +44,7 @@ def find_test(
         # TODO: forward to docs
     # Disable Celery if not needed
     if os.environ.get("USE_CELERY") == "false":
-        html_page = service.main(
+        return service.main(
             test_url=test_url,
             test_name=test_name,
             test_ref=test_ref,
@@ -54,7 +54,6 @@ def find_test(
             out_format=out_format,
             test_path=test_path,
             plan_path=plan_path)
-        return html_page
     r = service.main.delay(
             test_url=test_url,
             test_name=test_name,
@@ -77,7 +76,7 @@ def find_test(
 
 
 @app.get("/status")
-def status(task_id: str = Query(None, alias="task-id")) -> TaskOut:
+def status(task_id: str = Query(None, alias="task-id")) -> TaskOut | str:
     r = service.main.app.AsyncResult(task_id)
     return _to_task_out(r)
 

src/generators/json_generator.py

@@ -30,9 +30,9 @@ def generate_test_json(test: Test, logger: Logger) -> str:
             "ref": test.fmf_id.ref,
         }
     }
-    data = json.dumps(data)
+    json_data = json.dumps(data)
     logger.print("JSON file generated successfully!", color="green")
-    return data
+    return json_data
 
 
 def generate_plan_json(plan: Plan, logger: Logger) -> str:
@@ -61,9 +61,9 @@ def generate_plan_json(plan: Plan, logger: Logger) -> str:
             "ref": plan.fmf_id.ref,
         }
     }
-    data = json.dumps(data)
+    json_data = json.dumps(data)
     logger.print("JSON file generated successfully!", color="green")
-    return data
+    return json_data
 
 
 def generate_testplan_json(test: tmt.Test, plan: tmt.Plan, logger: Logger) -> str:
@@ -113,6 +113,6 @@ def generate_testplan_json(test: tmt.Test, plan: tmt.Plan, logger: Logger) -> st
             }
         }
     }
-    data = json.dumps(data)
+    json_data = json.dumps(data)
     logger.print("JSON file generated successfully!", color="green")
-    return data
+    return json_data

src/generators/yaml_generator.py

@@ -28,9 +28,9 @@ def generate_test_yaml(test: tmt.Test, logger: Logger) -> str:
             "ref": test.fmf_id.ref,
         }
     }
-    data = tmt.utils.dict_to_yaml(data)
+    yaml_data = tmt.utils.dict_to_yaml(data)
     logger.print("YAML file generated successfully!", color="green")
-    return data
+    return yaml_data
 
 
 def generate_plan_yaml(plan: tmt.Plan, logger: Logger) -> str:
@@ -59,9 +59,9 @@ def generate_plan_yaml(plan: tmt.Plan, logger: Logger) -> str:
             "ref": plan.fmf_id.ref,
         }
     }
-    data = tmt.utils.dict_to_yaml(data)
+    yaml_data = tmt.utils.dict_to_yaml(data)
     logger.print("YAML file generated successfully!", color="green")
-    return data
+    return yaml_data
 
 
 def generate_testplan_yaml(test: tmt.Test, plan: tmt.Plan, logger: Logger) -> str:
@@ -111,6 +111,6 @@ def generate_testplan_yaml(test: tmt.Test, plan: tmt.Plan, logger: Logger) -> st
             }
         }
     }
-    data = tmt.utils.dict_to_yaml(data)
+    yaml_data = tmt.utils.dict_to_yaml(data)
     logger.print("YAML file generated successfully!", color="green")
-    return data
+    return yaml_data

src/service.py

@@ -1,15 +1,15 @@
 import logging
 import os
-from pathlib import Path
 
 import tmt
 from celery.app import Celery
+from tmt.utils import Path
 
 from src.generators import html_generator as html
 from src.generators import json_generator, yaml_generator
 from src.utils import git_handler as utils
 
-logger = tmt.Logger(logging.Logger("tmt-logger"))
+logger = tmt.Logger(logging.getLogger("tmt-logger"))
 
 redis_url = os.getenv("REDIS_URL", "redis://localhost:6379")
 
@@ -33,11 +33,12 @@ def get_tree(url: str, name: str, ref: str, tree_path: str) -> tmt.base.Tree:
     path = utils.get_git_repository(url, logger, ref)
 
     if tree_path is not None:
+        tree_path += '/'
         # If path is set, construct a path to the tmt Tree
-        if '.git' == path.suffix:
+        if path.suffix == '.git':
             path = path.with_suffix('')
-        path = path.as_posix() + tree_path
-        path = Path(path)
+        path = Path(path.as_posix() + tree_path)
+
 
     logger.print("Looking for tree...")
     tree = tmt.base.Tree(path=path, logger=logger)
@@ -50,7 +51,7 @@ def process_test_request(test_url: str,
                          test_ref: str,
                          test_path: str,
                          return_object: bool,
-                         out_format: str) -> str | None | tmt.Test:
+                         out_format: str) -> str | tmt.Test | None:
     """
     This function processes the request for a test and returns the HTML file or the Test object
     :param test_url: Test url
@@ -68,7 +69,7 @@ def process_test_request(test_url: str,
 
     # Find the desired Test object
     wanted_test = tree.tests(names=[test_name])[0]
-    if wanted_test is []:
+    if wanted_test == []:
         logger.print("Test not found!", color="red")
         return None
     logger.print("Test found!", color="green")
@@ -81,6 +82,7 @@ def process_test_request(test_url: str,
             return json_generator.generate_test_json(wanted_test, logger=logger)
         case "yaml":
             return yaml_generator.generate_test_yaml(wanted_test, logger=logger)
+    return None
 
 
 def process_plan_request(plan_url: str,
@@ -106,7 +108,7 @@ def process_plan_request(plan_url: str,
 
     # Find the desired Plan object
     wanted_plan = tree.plans(names=[plan_name])[0]
-    if wanted_plan is []:
+    if wanted_plan == []:
         logger.print("Plan not found!", color="red")
         return None
     logger.print("Plan found!", color="green")
@@ -116,9 +118,10 @@ def process_plan_request(plan_url: str,
         case "html":
             return html.generate_plan_html_page(wanted_plan, logger=logger)
         case "json":
-            return json_generator.generate_plan_json(wanted_plan, logger=logger)
+            return json_generator.generate_plan_json(wanted_plan,  logger=logger)
         case "yaml":
             return yaml_generator.generate_plan_yaml(wanted_plan, logger=logger)
+    return None
 
 
 def process_testplan_request(test_url,
@@ -144,7 +147,13 @@ def process_testplan_request(test_url,
     :return:
     """
     test = process_test_request(test_url, test_name, test_ref, test_path, False, out_format)
+    if not isinstance(test, tmt.Test):
+        logger.print("Invalid test object", color="red")
+        return None
     plan = process_plan_request(plan_url, plan_name, plan_ref, plan_path, False, out_format)
+    if not isinstance(plan, tmt.Plan):
+        logger.print("Invalid plan object", color="red")
+        return None
     match out_format:
         case "html":
             return html.generate_testplan_html_page(test, plan, logger=logger)
@@ -153,6 +162,8 @@ def process_testplan_request(test_url,
         case "yaml":
             return yaml_generator.generate_testplan_yaml(test, plan, logger=logger)
 
+    return None
+
 
 @app.task
 def main(test_url: str | None,
@@ -163,7 +174,7 @@ def main(test_url: str | None,
          plan_name: str | None,
          plan_ref: str | None,
          plan_path: str | None,
-         out_format: str | None) -> str | None:
+         out_format: str) -> str | tmt.Test | tmt.Plan | None:
     logger.print("Starting...", color="blue")
     if test_name is not None and plan_name is None:
         return process_test_request(test_url, test_name, test_ref, test_path, True, out_format)
@@ -172,6 +183,7 @@ def main(test_url: str | None,
     elif plan_name is not None and test_name is not None:
         return process_testplan_request(test_url, test_name, test_ref, test_path,
                                         plan_url, plan_name, plan_ref, plan_path, out_format)
+    return None
 
 
 if __name__ == "__main__":

src/utils/git_handler.py

@@ -1,9 +1,9 @@
+import contextlib
 import os
-from pathlib import Path
-from subprocess import Popen
+from shutil import rmtree
 
-import tmt.utils
 from tmt import Logger
+from tmt.utils import Command, Common, GeneralError, Path, RunError, git_clone
 
 
 def checkout_branch(path: Path, logger: Logger, ref: str) -> None:
@@ -15,12 +15,12 @@ def checkout_branch(path: Path, logger: Logger, ref: str) -> None:
     :return:
     """
     try:
-        common_instance = tmt.utils.Common(logger=logger)
+        common_instance = Common(logger=logger)
         common_instance.run(
-            command=tmt.utils.Command('git', 'checkout', ref), cwd=path)
-    except tmt.utils.RunError:
+            command=Command('git', 'checkout', ref), cwd=path)
+    except RunError as err:
         logger.print("Failed to do checkout in the repository!", color="red")
-        raise AttributeError
+        raise AttributeError from err
 
 
 def clone_repository(url: str, logger: Logger, ref: str) -> None:
@@ -38,21 +38,21 @@ def clone_repository(url: str, logger: Logger, ref: str) -> None:
         if ref != "default":
             try:
                 checkout_branch(ref=ref, path=path, logger=logger)
-            except AttributeError:
-                raise AttributeError
+            except AttributeError as err:
+                raise AttributeError from err
         logger.print("Repository already cloned!", color="yellow")
         raise FileExistsError
     try:
-        tmt.utils.git_clone(url=url, destination=path, logger=logger)
+        git_clone(url=url, destination=path, logger=logger)
         if ref != "default":
             try:
                 checkout_branch(ref=ref, path=path, logger=logger)
-            except AttributeError:
-                raise AttributeError
+            except AttributeError as err:
+                raise AttributeError from err
             checkout_branch(ref=ref, path=path, logger=logger)
-    except tmt.utils.GeneralError as e:
+    except GeneralError as e:
         logger.print("Failed to clone the repository!", color="red")
-        raise Exception
+        raise Exception from e
     logger.print("Repository cloned successfully!", color="green")
 
 
@@ -62,11 +62,9 @@ def get_path_to_repository(url: str) -> Path:
     :param url: URL to the repository
     :return: Path to the cloned repository
     """
-    repo_name = url.rsplit('/', 1)[-1]
-    path = os.path.realpath(__file__)
-    path = path.replace("src/utils/git_handler.py", "")
-    path = Path(path + os.getenv("CLONE_DIR_PATH", "./.repos/") + repo_name)
-    return path
+    repo_name = url.rstrip('/').rsplit('/', 1)[-1]
+    root_dir = Path(__file__).resolve().parents[2]  # going up from src/utils/git_handler.py
+    return root_dir / os.getenv("CLONE_DIR_PATH", "./.repos/") / repo_name
 
 
 def check_if_repository_exists(url: str) -> bool:
@@ -75,8 +73,7 @@ def check_if_repository_exists(url: str) -> bool:
     :param url: URL to the repository
     :return: True if the repository is already cloned, False otherwise
     """
-    path = get_path_to_repository(url)
-    return path.exists()
+    return get_path_to_repository(url).exists()
 
 
 def clear_tmp_dir(logger: Logger) -> None:
@@ -86,11 +83,10 @@ def clear_tmp_dir(logger: Logger) -> None:
     :return:
     """
     logger.print("Clearing the .tmp directory...")
-    path = os.path.realpath(__file__)
-    path = path.replace("src/utils/git_handler.py", "")
-    path = Path(path + os.getenv("CLONE_DIR_PATH", "./.repos/"))
+    root_dir = Path(__file__).resolve().parents[2]  # going up from src/utils/git_handler.py
+    path = root_dir / os.getenv("CLONE_DIR_PATH", "./.repos/")
     try:
-        Popen(["rm", "-rf", path])
+        rmtree(path)
     except Exception as e:
         logger.print("Failed to clear the repository clone directory!", color="red")
         raise e
@@ -104,10 +100,8 @@ def get_git_repository(url: str, logger: Logger, ref: str) -> Path:
     :param logger: Instance of Logger
     :return: Path to the cloned repository
     """
-    try:
+    with contextlib.suppress(FileExistsError):
         clone_repository(url, logger, ref)
-    except FileExistsError:
-        pass
     return get_path_to_repository(url)