Create engine directory

container_ci_suite/api.py

@@ -34,7 +34,7 @@
 from pathlib import Path
 from tempfile import mkdtemp, mktemp
 
-from container_ci_suite.container_engine import PodmanCLIWrapper
+from container_ci_suite.engines.container import PodmanCLIWrapper
 from container_ci_suite.utils import (
     run_command,
     get_file_content,

container_ci_suite/engines/imagestreams.py

@@ -0,0 +1,109 @@
+#!/bin/env python3
+
+# MIT License
+#
+# Copyright (c) 2018-2019 Red Hat, Inc.
+
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import json
+import os
+
+from pathlib import Path
+from typing import Dict, List, Any
+
+IMAGESTREAMS_DIR: str = "imagestreams"
+
+
+class ImageStreamEngine(object):
+    version: str = ""
+
+    def __init__(self, working_dir: Path = Path(".")):
+        self.results: Dict[Any, Any] = {}
+        self.working_dir = working_dir
+
+    def get_latest_version(self) -> str:
+        latest_version = ""
+        print(f"Working dir for latest imagestream is {self.working_dir}.")
+        with open(self.working_dir / "Makefile") as f:
+            for line in f:
+                if not line.startswith("VERSIONS ="):
+                    continue
+                versions = line.split("=")[1].strip().split()
+                latest_version = versions[-1]
+        print(f"The latest version is {latest_version}.")
+        return latest_version
+
+    def load_json_file(self, filename: Path) -> Any:
+        with open(str(filename)) as f:
+            data = json.load(f)
+            isinstance(data, Dict)
+            return data
+
+    def check_version(self, json_dict: Dict[Any, Any]) -> List[str]:
+        res = []
+        for tags in json_dict["spec"]["tags"]:
+            print(
+                f"check_version: Compare tags['name']:'{tags['name']}' against version:'{self.version}'"
+            )
+            # The name can be"<stream>" or "<stream>-elX" or "<stream>-ubiX"
+            if tags["name"] == self.version or tags["name"].startswith(
+                self.version + "-"
+            ):
+                res.append(tags)
+        return res
+
+    def check_latest_tag(self, json_dict: Dict[Any, Any]) -> bool:
+        latest_tag_correct: bool = False
+        for tags in json_dict["spec"]["tags"]:
+            if tags["name"] != "latest":
+                continue
+            print(
+                f"check_latest_tag: Compare tags['name']:'{tags['name']}' against version:'{self.version}'"
+            )
+            # The latest can link to either "<stream>" or "<stream>-elX" or "<stream>-ubiX"
+            if tags["from"]["name"] == self.version or tags["from"]["name"].startswith(
+                self.version + "-"
+            ):
+                latest_tag_correct = True
+                print("Latest tag found.")
+        return latest_tag_correct
+
+    def check_imagestreams(self, version: str) -> int:
+        self.version = version
+        p = Path("..")
+        json_files = p.glob(f"{IMAGESTREAMS_DIR}/*.json")
+        if not json_files:
+            print(f"No json files present in {IMAGESTREAMS_DIR}.")
+            return 0
+        for f in json_files:
+            if os.environ.get("TARGET") in ("rhel7", "centos7") and "aarch64" in str(f):
+                print("Imagestream aarch64 is not supported on rhel7")
+                continue
+            print(f"Checking file {str(f)}.")
+            json_dict = self.load_json_file(f)
+            if not (self.check_version(json_dict) and self.check_latest_tag(json_dict)):
+                print(
+                    f"The latest version is not present in {str(f)} or in latest tag."
+                )
+                self.results[f] = False
+        if self.results:
+            return 1
+        print("Imagestreams contains the latest version.")
+        return 0

container_ci_suite/helm.py

@@ -33,7 +33,7 @@
 import container_ci_suite.utils as utils
 
 from container_ci_suite.openshift import OpenShiftAPI
-from container_ci_suite.openshift_ops import OpenShiftOperations
+from container_ci_suite.engines.openshift import OpenShiftOperations
 
 logging.basicConfig(format="%(levelname)s:%(message)s", level=logging.DEBUG)
 logger = logging.getLogger(__name__)

container_ci_suite/imagestreams.py

@@ -22,88 +22,21 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-import json
-import os
 
 from pathlib import Path
-from typing import Dict, List, Any
+
+
+from container_ci_suite.engines.imagestreams import ImageStreamEngine
 
 IMAGESTREAMS_DIR: str = "imagestreams"
 
 
 class ImageStreamChecker(object):
-    version: str = ""
-
     def __init__(self, working_dir: Path = Path(".")):
-        self.results: Dict[Any, Any] = {}
-        self.working_dir = working_dir
+        self.engine = ImageStreamEngine(working_dir=working_dir)
 
     def get_latest_version(self) -> str:
-        latest_version = ""
-        print(f"Working dir for latest imagestream is {self.working_dir}.")
-        with open(self.working_dir / "Makefile") as f:
-            for line in f:
-                if not line.startswith("VERSIONS ="):
-                    continue
-                versions = line.split("=")[1].strip().split()
-                latest_version = versions[-1]
-        print(f"The latest version is {latest_version}.")
-        return latest_version
-
-    def load_json_file(self, filename: Path) -> Any:
-        with open(str(filename)) as f:
-            data = json.load(f)
-            isinstance(data, Dict)
-            return data
-
-    def check_version(self, json_dict: Dict[Any, Any]) -> List[str]:
-        res = []
-        for tags in json_dict["spec"]["tags"]:
-            print(
-                f"check_version: Compare tags['name']:'{tags['name']}' against version:'{self.version}'"
-            )
-            # The name can be"<stream>" or "<stream>-elX" or "<stream>-ubiX"
-            if tags["name"] == self.version or tags["name"].startswith(
-                self.version + "-"
-            ):
-                res.append(tags)
-        return res
-
-    def check_latest_tag(self, json_dict: Dict[Any, Any]) -> bool:
-        latest_tag_correct: bool = False
-        for tags in json_dict["spec"]["tags"]:
-            if tags["name"] != "latest":
-                continue
-            print(
-                f"check_latest_tag: Compare tags['name']:'{tags['name']}' against version:'{self.version}'"
-            )
-            # The latest can link to either "<stream>" or "<stream>-elX" or "<stream>-ubiX"
-            if tags["from"]["name"] == self.version or tags["from"]["name"].startswith(
-                self.version + "-"
-            ):
-                latest_tag_correct = True
-                print("Latest tag found.")
-        return latest_tag_correct
+        return self.engine.get_latest_version()
 
     def check_imagestreams(self, version: str) -> int:
-        self.version = version
-        p = Path(".")
-        json_files = p.glob(f"{IMAGESTREAMS_DIR}/*.json")
-        if not json_files:
-            print(f"No json files present in {IMAGESTREAMS_DIR}.")
-            return 0
-        for f in json_files:
-            if os.environ.get("TARGET") in ("rhel7", "centos7") and "aarch64" in str(f):
-                print("Imagestream aarch64 is not supported on rhel7")
-                continue
-            print(f"Checking file {str(f)}.")
-            json_dict = self.load_json_file(f)
-            if not (self.check_version(json_dict) and self.check_latest_tag(json_dict)):
-                print(
-                    f"The latest version is not present in {str(f)} or in latest tag."
-                )
-                self.results[f] = False
-        if self.results:
-            return 1
-        print("Imagestreams contains the latest version.")
-        return 0
+        return self.engine.check_imagestreams(version=version)

container_ci_suite/openshift.py

@@ -29,8 +29,8 @@
 from pathlib import Path
 from typing import Dict, Any, List
 
-from container_ci_suite.container_engine import PodmanCLIWrapper
-from container_ci_suite.openshift_ops import OpenShiftOperations
+from container_ci_suite.engines.container import PodmanCLIWrapper
+from container_ci_suite.engines.openshift import OpenShiftOperations
 from container_ci_suite.utils import run_oc_command
 
 

setup.py

@@ -43,7 +43,7 @@ def get_requirements():
     description='A python3 container CI tool for testing images.',
     long_description=long_description,
     long_description_content_type='text/markdown',
-    version="0.5.1",
+    version="0.6.0",
     keywords='tool,containers,images,tests',
     packages=find_packages(exclude=["tests"]),
     url="https://github.com/sclorg/container-ci-suite",

tests/test_helm.py

@@ -29,7 +29,7 @@
 
 from container_ci_suite.helm import HelmChartsAPI
 from container_ci_suite.openshift import OpenShiftAPI
-from container_ci_suite.openshift_ops import OpenShiftOperations
+from container_ci_suite.engines.openshift import OpenShiftOperations
 
 test_dir = Path(os.path.abspath(os.path.dirname(__file__)))
 

tests/test_openshift.py

@@ -27,8 +27,8 @@
 from flexmock import flexmock
 
 from container_ci_suite.openshift import OpenShiftAPI
-from container_ci_suite.openshift_ops import OpenShiftOperations
-from container_ci_suite.container_engine import PodmanCLIWrapper
+from container_ci_suite.engines.openshift import OpenShiftOperations
+from container_ci_suite.engines.container import PodmanCLIWrapper
 from container_ci_suite import utils
 
 

tests/test_openshift_ops.py

@@ -24,7 +24,7 @@
 
 from flexmock import flexmock
 
-from container_ci_suite.openshift_ops import OpenShiftOperations
+from container_ci_suite.engines.openshift import OpenShiftOperations
 
 
 class TestOpenShiftOpsSuite(object):