make vault key a param and start adding some unit tests

src/cli.py

@@ -79,14 +79,14 @@ def main():
 # print("Anonymize:")
 # anonymizer = AnonymizerEngine()
 # anonymizer.add_anonymizer(VaultEncrypt)
-# operators = {"DEFAULT": OperatorConfig("vault_encrypt", {"vault_url": VAULT_URL})}
+# operators = {"DEFAULT": OperatorConfig("vault_encrypt", {"vault_url": VAULT_URL, "key": "orders"})}
 # anon_res = anonymizer.anonymize(t, res, operators)
 # print(anon_res.text)
 
 
 # print("Deanonymize:")
 # deanonymizer = DeanonymizeEngine()
 # deanonymizer.add_deanonymizer(VaultDecrypt)
-# de_ops = {"DEFAULT": OperatorConfig("vault_decrypt", {"vault_url": VAULT_URL})}
+# de_ops = {"DEFAULT": OperatorConfig("vault_decrypt", {"vault_url": VAULT_URL, "key": "orders"})}
 # deanon_res = deanonymizer.deanonymize(anon_res.text, anon_res.items, de_ops)
 # print(deanon_res.text)

src/operators/vault.py

@@ -19,14 +19,15 @@ def _base64ify(self, bytes_or_str):
 
     def operate(self, text: str, params: Dict = None) -> str:
         vault_url = params.get("vault_url")
-        client = hvac.Client(url=vault_url)
+        key = params.get("key")
 
+        client = hvac.Client(url=vault_url)
         encrypt_data_response = client.secrets.transit.encrypt_data(
-            name='orders',
+            name=key,
             plaintext=self._base64ify(text),
         )
-
         ciphertext = encrypt_data_response['data']['ciphertext']
+
         return ciphertext
 
     def validate(self, params: Dict = None) -> None:
@@ -40,6 +41,12 @@ def validate(self, params: Dict = None) -> None:
         else:
             raise InvalidParamException(f"Invalid input, vault_url must be a string.")
 
+        key = params.get("key")
+        if isinstance(key, str) and key:
+            pass
+        else:
+            raise InvalidParamException(f"Invalid input, key must be a valid encryption key name.")
+
     def operator_name(self) -> str:
         return "vault_encrypt"
 
@@ -51,15 +58,16 @@ def operator_type(self) -> OperatorType:
 class VaultDecrypt(Operator):
     def operate(self, text: str, params: Dict = None) -> str:
         vault_url = params.get("vault_url")
-        client = hvac.Client(url=vault_url)
+        key = params.get("key")
 
+        client = hvac.Client(url=vault_url)
         decrypt_data_response = client.secrets.transit.decrypt_data(
-            name='orders',
+            name=key,
             ciphertext=text,
         )
-
         encodedtext = decrypt_data_response['data']['plaintext']
         plaintext = base64.b64decode(encodedtext).decode('utf8')
+
         return plaintext
 
     def validate(self, params: Dict = None) -> None:
@@ -73,6 +81,12 @@ def validate(self, params: Dict = None) -> None:
         else:
             raise InvalidParamException(f"Invalid input, vault_url must be a string.")
 
+        key = params.get("key")
+        if isinstance(key, str) and key:
+            pass
+        else:
+            raise InvalidParamException(f"Invalid input, key must be a valid encryption key name.")
+
     def operator_name(self) -> str:
         return "vault_decrypt"
 

tests/operators/vault_test.py

@@ -0,0 +1,47 @@
+import pytest
+
+from presidio_anonymizer.entities import InvalidParamException
+from operators.vault import VaultEncrypt, VaultDecrypt
+
+
+class TestVaultEncrypt:
+    def test_given_valid_key_raises_no_exceptions(self):
+        VaultEncrypt().validate(params={"vault_url": "http://127.0.0.1:8200", "key": "foobar"})
+
+    def test_given_invalid_key_raises_exceptions(self):
+        with pytest.raises(
+            InvalidParamException,
+            match="Invalid input, key must be a valid encryption key name.",
+        ):
+            VaultEncrypt().validate(params={"vault_url": "http://127.0.0.1:8200", "key": 1})
+
+    def test_given_valid_url_raises_no_exceptions(self):
+        VaultEncrypt().validate(params={"vault_url": "http://127.0.0.1:8200", "key": "foobar"})
+
+    def test_given_invalid_url_raises_exceptions(self):
+        with pytest.raises(
+            InvalidParamException,
+            match="Invalid input, vault_url must be a valid URL.",
+        ):
+            VaultEncrypt().validate(params={"vault_url": "http:/127.0.0.1:8200", "key": "foobar"})
+
+class TestVaultDecrypt:
+    def test_given_valid_key_raises_no_exceptions(self):
+        VaultDecrypt().validate(params={"vault_url": "http://127.0.0.1:8200", "key": "foobar"})
+
+    def test_given_invalid_key_raises_exceptions(self):
+        with pytest.raises(
+            InvalidParamException,
+            match="Invalid input, key must be a valid encryption key name.",
+        ):
+            VaultDecrypt().validate(params={"vault_url": "http://127.0.0.1:8200", "key": 1})
+
+    def test_given_valid_url_raises_no_exceptions(self):
+        VaultDecrypt().validate(params={"vault_url": "http://127.0.0.1:8200", "key": "foobar"})
+
+    def test_given_invalid_url_raises_exceptions(self):
+        with pytest.raises(
+            InvalidParamException,
+            match="Invalid input, vault_url must be a valid URL.",
+        ):
+            VaultDecrypt().validate(params={"vault_url": "http:/127.0.0.1:8200", "key": "foobar"})