refactor CI manifests

Signed-off-by: Tianle Huang <tianleh@amazon.com>

src/ci_workflow/ci_input_manifest.py

@@ -14,34 +14,22 @@
 
 
 class CiInputManifest(CiManifest):
-    def __init__(self, keep, snapshot, component, file, component_command):
-        self.keep = keep
-        self.snapshot = snapshot
-        self.component = component
-        self.file = file
-        self.component_command = component_command
+    def __init__(self, file, args):
+        super().__init__(InputManifest.from_file(file), args)
 
-    def __from_file(self):
-        self.manifest = InputManifest.from_file(self.file)
+    def __check__(self):
 
-    def check(self):
-        self.__from_file()
+        target = CiTarget(version=self.manifest.build.version, snapshot=self.args.snapshot)
 
-        target = CiTarget(version=self.manifest.build.version, snapshot=self.snapshot)
-
-        with TemporaryDirectory(keep=self.keep, chdir=True) as work_dir:
+        with TemporaryDirectory(keep=self.args.keep, chdir=True) as work_dir:
             logging.info(f"Sanity-testing in {work_dir.name}")
 
             logging.info(f"Sanity testing {self.manifest.build.name}")
 
-            for component in self.manifest.components.select(focus=self.component):
+            for component in self.manifest.components.select(focus=self.args.component):
                 logging.info(f"Sanity testing {component.name}")
 
-                try:
-                    ci_check_list = CiCheckLists.from_component(component, target)
-                    ci_check_list.checkout(work_dir.name)
-                    ci_check_list.check()
-                    logging.info("Done.")
-                except:
-                    logging.error(f"Error checking {component.name}, retry with: {self.component_command(component.name)}")
-                    raise
+                ci_check_list = CiCheckLists.from_component(component, target)
+                ci_check_list.checkout(work_dir.name)
+                ci_check_list.check()
+                logging.info("Done.")

src/ci_workflow/ci_manifest.py

@@ -5,12 +5,17 @@
 # compatible open source license.
 
 import abc
+import logging
 
 
 class CiManifest(abc.ABC):
-    def __init__(self):
-        pass
+    def __init__(self, manifest, args):
+        self.manifest = manifest
+        self.args = args
 
-    @abc.abstractmethod
     def check(self):
-        pass
+        try:
+            self.__check__()
+        except:
+            logging.error("CI Manifest check failed")
+            raise

src/ci_workflow/ci_manifests.py

@@ -0,0 +1,23 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+
+
+import re
+
+from ci_workflow.ci_input_manifest import CiInputManifest
+from ci_workflow.ci_test_manifest import CiTestManifest
+
+
+class CiManifests:
+    def __klass(filename):
+        if re.search("-test.yml$", filename):
+            return CiTestManifest
+        else:
+            return CiInputManifest
+
+    @classmethod
+    def from_file(cls, file, args):
+        return cls.__klass(file.name)(file, args)

src/ci_workflow/ci_test_manifest.py

@@ -12,17 +12,10 @@
 
 
 class CiTestManifest(CiManifest):
-    def __init__(self, file):
-        self.file = file
+    def __init__(self, file, args):
+        super().__init__(TestManifest.from_file(file), args)
 
-    def __from_file(self):
-        self.manifest = TestManifest.from_file(self.file)
-
-    def check(self):
-        try:
-            self.__from_file()
-            logging.info("TestManifest schema validation succeeded")
-            logging.info("Done.")
-        except:
-            logging.error(f"TestManifest check failed for {self.file.name}")
-            raise
+    def __check__(self):
+        assert self.manifest
+        logging.info("TestManifest schema validation succeeded")
+        logging.info("Done.")

src/run_ci.py

@@ -6,27 +6,19 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
-import re
+
 import sys
 
 from ci_workflow.ci_args import CiArgs
-from ci_workflow.ci_input_manifest import CiInputManifest
-from ci_workflow.ci_test_manifest import CiTestManifest
+from ci_workflow.ci_manifests import CiManifests
 from system import console
 
 
 def main():
     args = CiArgs()
     console.configure(level=args.logging_level)
 
-    if __is_test_manifest(args.manifest.name):
-        CiTestManifest(args.manifest).check()
-    else:
-        CiInputManifest(args.keep, args.snapshot, args.component, args.manifest, args.component_command).check()
-
-
-def __is_test_manifest(path):
-    return re.search("-test.yml$", path)
+    CiManifests.from_file(args.manifest, args).check()
 
 
 if __name__ == "__main__":