test

integration_tests/cosmoscli.py

@@ -1870,3 +1870,17 @@ def set_e2ee_key(self, key, **kwargs):
         if rsp["code"] == 0:
             rsp = self.event_query_tx_for(rsp["txhash"])
         return rsp
+
+    def keygen(self, **kwargs):
+        return self.raw("keygen", home=self.data_dir, **kwargs).strip().decode()
+
+    def encrypt(self, input, receipts, **kwargs):
+        return self.raw(
+            "encrypt",
+            input,
+            *[val for a in [["--r", val] for val in receipts] for val in a],
+            **kwargs,
+        )
+
+    def decrypt(self, input, **kwargs):
+        return self.raw("decrypt", input, home=self.data_dir, **kwargs).strip().decode()

integration_tests/test_e2ee.py

@@ -1,4 +1,5 @@
 import base64
+import tempfile
 
 
 def test_set_key(cronos):
@@ -8,3 +9,19 @@ def test_set_key(cronos):
     adr = cli.address("community")
     p = cli.query_e2ee_key(adr)
     assert p["key"] == key
+
+
+def test_encrypt_decrypt(cronos):
+    cli = cronos.cosmos_cli()
+    key_dir = tempfile.mkdtemp(dir=cronos.base_dir)
+    key0 = f"{key_dir}/key0"
+    key1 = f"{key_dir}/key1"
+    pubkey0 = cli.keygen(o=key0)
+    pubkey1 = cli.keygen(o=key1)
+    input = f"{key_dir}/input"
+    decrypted = f"{key_dir}/input.age"
+    data = "Hello, World!"
+    with open(input, "w") as file:
+        file.write(data)
+    cli.encrypt(input, [pubkey0, pubkey1], o=decrypted)
+    assert cli.decrypt(decrypted, i=key0) == cli.decrypt(decrypted, i=key1) == data