Add CSITs for Truststore Merger

Issue-ID: DCAEGEN2-2253
Signed-off-by: Joanna Jeremicz <joanna.jeremicz@nokia.com>
Change-Id: Ie2d648108c99f4a1e61fe855f318078e6ffc0e21

4 files changed, 95 insertions, 0 deletions

diff --git a/tests/oom-platform-cert-service/truststoremerger/libraries/EnvsReader.py b/tests/oom-platform-cert-service/truststoremerger/libraries/EnvsReader.py
new file mode 100644
index 00000000..cc60eed6
--- /dev/null
+++ b/tests/oom-platform-cert-service/truststoremerger/libraries/EnvsReader.py
@@ -0,0 +1,11 @@
+
+class EnvsReader:
+
+  def read_env_list_from_file(self, path):
+    f = open(path, "r")
+    r_list = []
+    for line in f:
+      line = line.strip()
+      if line[0] != "#":
+        r_list.append(line)
+    return r_list
diff --git a/tests/oom-platform-cert-service/truststoremerger/libraries/JksTruststoreValidator.py b/tests/oom-platform-cert-service/truststoremerger/libraries/JksTruststoreValidator.py
new file mode 100644
index 00000000..e18ca12c
--- /dev/null
+++ b/tests/oom-platform-cert-service/truststoremerger/libraries/JksTruststoreValidator.py
@@ -0,0 +1,18 @@
+
+import jks
+
+class JksTruststoreValidator:
+
+  def get_truststore(self, truststore_path, password_path):
+    truststore = jks.KeyStore.load(truststore_path, open(password_path, 'rb').read())
+    return truststore.certs
+
+  def assert_jks_truststores_equal(self, result_truststore_path, password_path, expected_truststore_path):
+    result_certs = self.get_truststore(result_truststore_path, password_path)
+    expected_certs = self.get_truststore(expected_truststore_path, password_path)
+    if len(result_certs) != len(expected_certs):
+      return False
+    for k in result_certs:
+      if not (k in expected_certs and result_certs[k].cert == expected_certs[k].cert):
+        return False
+    return True
diff --git a/tests/oom-platform-cert-service/truststoremerger/libraries/PemTruststoreValidator.py b/tests/oom-platform-cert-service/truststoremerger/libraries/PemTruststoreValidator.py
new file mode 100644
index 00000000..8dc9623d
--- /dev/null
+++ b/tests/oom-platform-cert-service/truststoremerger/libraries/PemTruststoreValidator.py
@@ -0,0 +1,19 @@
+import re
+
+BEGIN_CERT = "-----BEGIN CERTIFICATE-----"
+END_CERT = "-----END CERTIFICATE-----"
+
+class PemTruststoreValidator:
+
+  def assert_pem_truststores_equal(self, result_pem_path, expected_pem_path):
+    result_certs = self.get_list_of_pem_certificates(result_pem_path)
+    expected_certs = self.get_list_of_pem_certificates(expected_pem_path)
+    result_certs.sort()
+    expected_certs.sort()
+    if len(result_certs) != len(expected_certs):
+      return False
+    return result_certs == expected_certs
+
+
+  def get_list_of_pem_certificates(self, path):
+    return re.findall(BEGIN_CERT + '(.+?)' + END_CERT, open(path, 'rb').read(), re.DOTALL)
diff --git a/tests/oom-platform-cert-service/truststoremerger/libraries/TrustMergerManager.py b/tests/oom-platform-cert-service/truststoremerger/libraries/TrustMergerManager.py
new file mode 100644
index 00000000..f7a493c4
--- /dev/null
+++ b/tests/oom-platform-cert-service/truststoremerger/libraries/TrustMergerManager.py
@@ -0,0 +1,47 @@
+import docker
+import os
+import shutil
+from EnvsReader import EnvsReader
+from docker.types import Mount
+
+ARCHIVES_PATH = os.getenv("WORKSPACE") + "/archives/"
+
+
+class TrustMergerManager:
+
+  def __init__(self, mount_path, truststores_path):
+    self.mount_path = mount_path
+    self.truststores_path = truststores_path
+
+  def run_merger_container(self, merger_image, merger_name, path_to_env):
+    self.remove_mount_dir()
+    shutil.copytree(self.truststores_path, self.mount_path)
+    client = docker.from_env()
+    environment = EnvsReader().read_env_list_from_file(path_to_env)
+    container = client.containers.run(
+        image=merger_image,
+        name=merger_name,
+        environment=environment,
+        user='root',  # Run container as root to avoid permission issues with volume mount access
+        mounts=[Mount(target='/var/certs', source=self.mount_path, type='bind')],
+        detach=True
+    )
+    exitcode = container.wait()
+    return exitcode
+
+  def create_mount_dir(self):
+    if not os.path.exists(self.mount_path):
+      os.makedirs(self.mount_path)
+
+  def remove_mount_dir(self):
+    if os.path.exists(self.mount_path):
+      shutil.rmtree(self.mount_path)
+
+  def remove_merger_container_and_save_logs(self, container_name, log_file_name):
+    client = docker.from_env()
+    container = client.containers.get(container_name)
+    text_file = open(ARCHIVES_PATH + "merger_container_" + log_file_name + ".log", "w")
+    text_file.write(container.logs())
+    text_file.close()
+    container.remove()
+    self.remove_mount_dir()