feat: test snapshot file contents

kubernetes/.pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+# https://docs.python.org/3/library/logging.html#levels
+log_cli = true
+log_cli_level = 20

kubernetes/vault_server_mock.py

@@ -1,11 +1,30 @@
 import subprocess
+import hvac
+import requests
 
 VAULT_PORT = 8200
+VAULT_ADDR = f"http://127.0.0.1:{VAULT_PORT}"
 VAULT_CONFIG = "./vault_config.hcl"
 VAULT_TOKEN = "root"
 VAULT_DATA_DIR = "./vault_data"
 
 class VaultServer():
+    """
+    Vault server mock.
+
+    Runs on http://127.0.0.1:8200 and can be initialized with Vault token as
+    first argument.
+    """
+
+    def __init__(self, *args):
+        self.token = VAULT_TOKEN
+        if len(args) >= 1:
+            self.token = args[0]
+        else:
+            self.token = VAULT_TOKEN
+
+        self.headers = {"X-Vault-Token": self.token}
+
     def reset_data(self, dir: str = VAULT_DATA_DIR):
         """
         Reset Vault server mock raft data directory.
@@ -14,15 +33,51 @@ def reset_data(self, dir: str = VAULT_DATA_DIR):
         subprocess.run(f"rm -rf {dir}/*", shell=True)
         print(f"Vault data dir reset: {dir}")
 
-    def run(self, port: int = VAULT_PORT, config: str = VAULT_CONFIG,
-            token: str = VAULT_TOKEN):
+    def run(self, port: int = VAULT_PORT, config: str = VAULT_CONFIG):
         """
         Start the Vault server mock with data dir and config.
         """
 
-        command = f"$(which vault) server -dev -dev-root-token-id={token} -config={config}"
+        command = f"$(which vault) server -dev -dev-root-token-id={self.token} -config={config}"
         self.proc = subprocess.Popen(command, shell=True)
 
+    def init_hvac_client(self):
+        """
+        Initialize hvac client as root on the mock server
+        """
+
+        self.hvac_client = hvac.Client(url=VAULT_ADDR)
+        self.hvac_client.token = self.token
+
+        assert self.hvac_client.is_authenticated()
+
+    def setup_kubernetes_auth(self):
+        """"
+        Configure a Kubernetes auth backend for testing purposes.
+        """
+
+        self.hvac_client.sys.enable_auth_method(
+                method_type="kubernetes",
+                path="kubernetes",
+        )
+
+        #data = {
+        #  "kubernetes_host": "127.0.0.1",
+        #}
+        # configure Kubernetes auth backend
+
+        data = {
+          "bound_service_account_names": "default",
+          "bound_service_account_namespaces": "*",
+          "policies": ["root"],
+        }
+
+        # add login role
+        # https://developer.hashicorp.com/vault/api-docs/auth/kubernetes#create-update-role
+        ret = requests.post(f"{VAULT_ADDR}/v1/auth/kubernetes/role/default",
+                            json=data,
+                            headers=self.headers)
+
     def status(self):
         """
         Return Vault server mock status.

kubernetes/vault_snapshot.py

@@ -79,7 +79,7 @@ def __init__(self, **kwargs):
         ####VAULT_TOKEN=$(vault write -field=token  auth/kubernetes/login role="${VAULT_ROLE}" jwt="${JWT}")
         #Kubernetes(hvac_client.adapter).login(role=role, jwt=jwt)
 
-        self.logger.debug(f"Connecting to Vault API {self.vault_addr}")
+        self.logger.info(f"Connecting to Vault API {self.vault_addr}")
         self.hvac_client = hvac.Client(url=self.vault_addr)
         self.hvac_client.token = self.vault_token
 
@@ -97,11 +97,11 @@ def snapshot(self):
         with self.hvac_client.sys.take_raft_snapshot() as resp:
             assert resp.ok
 
-            self.logger.debug("Raft snapshot status code: %d" % resp.status_code)
+            self.logger.info("Raft snapshot status code: %d" % resp.status_code)
 
             date_str = datetime.datetime.now(datetime.UTC).strftime("%F-%H%M")
             file_name = "vault_%s.snapshot" % (date_str)
-            self.logger.debug(f"File name: {file_name}")
+            self.logger.info(f"File name: {file_name}")
 
             # Upload the file
             # * https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3/client/put_object.html
@@ -111,7 +111,7 @@ def snapshot(self):
                     Bucket=self.s3_bucket,
                     Key=file_name,
                 )
-                self.logger.debug("s3 put_object response: %s", response)
+                self.logger.info("s3 put_object response: %s", response)
             except ClientError as e:
                 logging.error(e)
 
@@ -123,7 +123,7 @@ def snapshot(self):
                             aws_secret_access_key=self.s3_secret_access_key)
         bucket = s3.Bucket(self.s3_bucket)
         for key in bucket.objects.all():
-            self.logger.debug(key.key)
+            self.logger.info(key.key)
             # todo: do the S3_EXPIRE_DAYS magic
 
         return file_name

kubernetes/vault_snapshot_test.py

@@ -1,12 +1,18 @@
+import logging
+from io import BytesIO
+import tarfile
 import pytest
 import boto3
 import hvac
+import time
 from moto import mock_aws
 from unittest.mock import patch, create_autospec
 
 from vault_snapshot import VaultSnapshot
 from vault_server_mock import VaultServer
 
+logger = logging.getLogger(__name__)
+
 class TestVaultSnapshots:
     """
     Test Vault snapshot functionality.
@@ -21,6 +27,10 @@ def mock_vault_server(self):
         self.mock.reset_data()
         # run mock
         self.mock.run()
+        # configure Vault with auth backend
+        time.sleep(3)
+        self.mock.init_hvac_client()
+        self.mock.setup_kubernetes_auth()
         # return process status, 'None' means process is still running
         yield self.mock.status()
         # when tests are done, teardown the Vault server
@@ -52,9 +62,12 @@ def test_snapshot(self, mock_vault_server):
         )
         file_name = vault_snapshot.snapshot()
 
-        body = conn.Object(bucket_name,
-                           file_name).get()#["Body"].read()#.decode("utf-8")
+        s3obj = conn.Object(bucket_name, file_name).get()
+        body = s3obj["Body"]
+        file_obj = BytesIO(body.read())
 
-        #print(body)
-
-        #assert body == "is awesome" 
+        snapshot_files = tarfile.open(fileobj=file_obj).getmembers()
+        for f in snapshot_files:
+            logger.info(f"- {f}")
+
+        assert len(snapshot_files) >= 4