47 lines
2 KiB
Python
47 lines
2 KiB
Python
import pathlib
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).parents[1]))
|
|
from provisioner import _username, dispatch, DriftResult, Result
|
|
|
|
|
|
class Fake:
|
|
def provision(self, payload): return Result("p", "u", "password_setup_required", False)
|
|
def suspend(self, subject): return Result("p", subject, "suspended", False)
|
|
def reactivate(self, subject): return Result("p", subject, "active", False)
|
|
def deprovision(self, subject): return Result("p", subject, "deprovisioned", False)
|
|
def drift(self, payload): return DriftResult("p", payload["external_subject"], "drifted", ("group:missing:t:users",))
|
|
def reconcile(self, payload): return DriftResult("p", payload["external_subject"], "reconciled", (), ("group:added:t:users",))
|
|
|
|
|
|
class ProvisionerTests(unittest.TestCase):
|
|
def test_username_is_stable_and_sanitized(self):
|
|
self.assertEqual("bernd.worsch", _username("Bernd.Worsch@binky-hedgehog.com"))
|
|
|
|
def test_dispatch_requires_idempotency(self):
|
|
with self.assertRaisesRegex(ValueError, "idempotency_key"):
|
|
dispatch(Fake(), "/v1/identities/suspend", {
|
|
"external_subject": "u", "correlation_id": "c"
|
|
})
|
|
|
|
def test_lifecycle_dispatch(self):
|
|
result = dispatch(Fake(), "/v1/identities/suspend", {
|
|
"external_subject": "u",
|
|
"idempotency_key": "1234567890123456",
|
|
"correlation_id": "c",
|
|
})
|
|
self.assertEqual("suspended", result.status)
|
|
|
|
def test_drift_and_reconcile_dispatch(self):
|
|
payload = {
|
|
"external_subject": "u",
|
|
"tenant": "t",
|
|
"primary_email": "u@example.com",
|
|
"idempotency_key": "1234567890123456",
|
|
"correlation_id": "c",
|
|
}
|
|
drift = dispatch(Fake(), "/v1/identities/drift", payload)
|
|
reconciled = dispatch(Fake(), "/v1/identities/reconcile", payload)
|
|
self.assertEqual("drifted", drift.status)
|
|
self.assertEqual("reconciled", reconciled.status)
|