pki_ca/tests/integration/test_zenroom_live.py

72 lines
2.5 KiB
Python

import os
import unittest
import sys
from pathlib import Path
# Import from ca_core
code_path = Path(__file__).parent.parent.parent / "ca_core"
sys.path.insert(0, str(code_path))
from crypto.zenroom_service_client import ZenroomServiceClient
def _live_enabled() -> bool:
return os.environ.get("RUN_LIVE_ZENROOM", "").strip().lower() in {"1", "true", "yes"}
@unittest.skipUnless(_live_enabled(), "Set RUN_LIVE_ZENROOM=1 to run live Zenroom service smoke tests")
class TestZenroomLiveServices(unittest.TestCase):
@classmethod
def setUpClass(cls):
base_url = os.environ.get("ZENROOM_BASE_URL", "http://localhost:3300").strip()
api_prefix = os.environ.get("ZENROOM_API_PREFIX", "/api").strip()
timeout_s = int(os.environ.get("ZENROOM_TIMEOUT_S", "20"))
cls.client = ZenroomServiceClient(
base_url=base_url,
api_prefix=api_prefix,
timeout_s=timeout_s,
)
def test_end_to_end_8_calls(self):
sender_kp = self.client.generate_keypair("LiveUser123456")
sender_pub = self.client.generate_public_key(sender_kp["keyring"])
plaintext = "Dear Bob, your name is too short, goodbye - Alice."
sm = self.client.symmetric_encrypt(
header="A very important secret",
message=plaintext,
shared_key="myVerySecretPassword",
)
pt = self.client.symmetric_decrypt(secret_message=sm, shared_key="myVerySecretPassword")
self.assertEqual(pt, plaintext)
secret = self.client.asymmetric_encrypt(
receiver_public_key=sender_pub,
sender_keyring=sender_kp["keyring"],
message="Hello from live test",
header="Live header",
)
dec = self.client.asymmetric_decrypt(
sender_public_key=sender_pub,
receiver_keyring=sender_kp["keyring"],
secret=secret,
)
self.assertEqual(dec["header"], "Live header")
self.assertEqual(dec["text"], "Hello from live test")
signed = self.client.sign_objects(
signer_keyring=sender_kp["keyring"],
objects={"myMessage": "Signed live message"},
)
sig = signed["myMessage.signature"]
ok = self.client.verify_signature(
message_field="myMessage",
message_value=signed["myMessage"],
signature={"r": sig["r"], "s": sig["s"]},
signer_public_key=sender_pub,
)
self.assertTrue(ok)