107 lines
4 KiB
Python
107 lines
4 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from urllib.error import URLError
|
|
|
|
from user_engine.adapters.provisioning import HTTPIdentityProvisioningAdapter
|
|
from user_engine.ports import ProvisioningRequest
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): return None
|
|
|
|
|
|
class ProvisioningAdapterTests(unittest.TestCase):
|
|
@patch("user_engine.adapters.provisioning.urlopen")
|
|
def test_provision_transports_idempotency_without_provider_leakage(self, opener):
|
|
opener.return_value = Response(json.dumps({
|
|
"provider": "netkingdom-lldap",
|
|
"external_subject": "person-1",
|
|
"status": "password_setup_required",
|
|
"resumed": True,
|
|
"password_setup_url": "https://kc.example/setup/password?token=opaque",
|
|
}).encode())
|
|
adapter = HTTPIdentityProvisioningAdapter(
|
|
base_url="http://provisioner", bearer_token="secret\n"
|
|
)
|
|
result = adapter.provision(ProvisioningRequest(
|
|
user_id="person-1",
|
|
tenant="tenant:friendly:binky",
|
|
primary_email="person@example.test",
|
|
display_name="Person",
|
|
idempotency_key="idem-1234567890123456",
|
|
correlation_id="corr-1",
|
|
))
|
|
self.assertTrue(result.resumed)
|
|
self.assertEqual(
|
|
"https://kc.example/setup/password?token=opaque",
|
|
result.password_setup_url,
|
|
)
|
|
request = opener.call_args.args[0]
|
|
self.assertEqual("Bearer secret", request.headers["Authorization"])
|
|
self.assertIn(b"idem-1234567890123456", request.data)
|
|
|
|
@patch("user_engine.adapters.provisioning.urlopen")
|
|
def test_drift_and_reconcile_use_provider_neutral_desired_state(self, opener):
|
|
opener.side_effect = [
|
|
Response(json.dumps({
|
|
"provider": "netkingdom-lldap",
|
|
"external_subject": "person",
|
|
"status": "drifted",
|
|
"drift": ["status:mismatch"],
|
|
"changed": [],
|
|
}).encode()),
|
|
Response(json.dumps({
|
|
"provider": "netkingdom-lldap",
|
|
"external_subject": "person",
|
|
"status": "reconciled",
|
|
"drift": [],
|
|
"changed": ["group:removed:netkingdom-suspended"],
|
|
}).encode()),
|
|
]
|
|
adapter = HTTPIdentityProvisioningAdapter(
|
|
base_url="http://provisioner", bearer_token="secret"
|
|
)
|
|
request = ProvisioningRequest(
|
|
user_id="person",
|
|
tenant="tenant:friendly:binky",
|
|
primary_email="person@example.test",
|
|
display_name="Person",
|
|
idempotency_key="idem-1234567890123456",
|
|
correlation_id="corr-1",
|
|
roles=("user",),
|
|
)
|
|
self.assertEqual(
|
|
("status:mismatch",),
|
|
adapter.drift(request, external_subject="person").drift,
|
|
)
|
|
reconciled = adapter.reconcile(request, external_subject="person")
|
|
self.assertEqual("reconciled", reconciled.status)
|
|
self.assertEqual(
|
|
("group:removed:netkingdom-suspended",), reconciled.changed
|
|
)
|
|
|
|
@patch("user_engine.adapters.provisioning.urlopen")
|
|
def test_idempotent_request_retries_one_transient_provider_failure(self, opener):
|
|
opener.side_effect = [
|
|
URLError("temporary"),
|
|
Response(json.dumps({
|
|
"provider": "netkingdom-lldap",
|
|
"external_subject": "person",
|
|
"status": "active",
|
|
}).encode()),
|
|
]
|
|
adapter = HTTPIdentityProvisioningAdapter(
|
|
base_url="http://provisioner",
|
|
bearer_token="secret",
|
|
retry_delay=0,
|
|
)
|
|
result = adapter.reactivate(
|
|
external_subject="person",
|
|
idempotency_key="idem-1234567890123456",
|
|
correlation_id="corr-1",
|
|
)
|
|
self.assertEqual("active", result.status)
|
|
self.assertEqual(2, opener.call_count)
|