generated from coulomb/repo-seed
94 lines
3.4 KiB
Python
94 lines
3.4 KiB
Python
import json
|
|
import unittest
|
|
|
|
from user_engine.adapters.local import InMemoryUserEngineStore
|
|
from user_engine.migrations import USER_ENGINE_STORE_RECORD_TYPES
|
|
from user_engine.store_records import (
|
|
StoreRecord,
|
|
domain_record_from_store_record,
|
|
store_record_for,
|
|
validate_store_record_codecs,
|
|
)
|
|
from user_engine.testing.store_conformance import (
|
|
PROFILE_SECRET_VALUE,
|
|
RAW_FACTOR_VALUE,
|
|
TENANT,
|
|
USER_ID,
|
|
reference_store_records,
|
|
)
|
|
|
|
|
|
class StoreRecordSerializationTests(unittest.TestCase):
|
|
def test_codecs_cover_migration_manifest_record_types(self):
|
|
self.assertEqual(validate_store_record_codecs(), ())
|
|
|
|
def test_reference_records_round_trip_through_json_safe_payloads(self):
|
|
store = InMemoryUserEngineStore()
|
|
store.migrate()
|
|
records = reference_store_records(store)
|
|
expected_types = set(USER_ENGINE_STORE_RECORD_TYPES)
|
|
|
|
encoded_types = set()
|
|
for value in records.values():
|
|
record = store_record_for(value)
|
|
encoded_types.add(record.record_type)
|
|
json.dumps(record.payload)
|
|
decoded = domain_record_from_store_record(
|
|
StoreRecord(
|
|
record_type=record.record_type,
|
|
record_key=record.record_key,
|
|
payload=json.loads(json.dumps(record.payload)),
|
|
tenant=record.tenant,
|
|
user_id=record.user_id,
|
|
application_id=record.application_id,
|
|
scope_type=record.scope_type,
|
|
scope_id=record.scope_id,
|
|
)
|
|
)
|
|
self.assertEqual(decoded, value)
|
|
|
|
self.assertEqual(encoded_types, expected_types)
|
|
|
|
def test_record_metadata_supports_provider_indexes(self):
|
|
store = InMemoryUserEngineStore()
|
|
store.migrate()
|
|
records = reference_store_records(store)
|
|
|
|
tenant_account = store_record_for(records["tenant_account"])
|
|
active_context = store_record_for(records["access_context"])
|
|
profile_value = store_record_for(records["profile_value"])
|
|
factor = store_record_for(records["factor"])
|
|
|
|
self.assertEqual(tenant_account.record_key, f'["{TENANT}","{USER_ID}"]')
|
|
self.assertEqual(tenant_account.tenant, TENANT)
|
|
self.assertEqual(tenant_account.user_id, USER_ID)
|
|
self.assertEqual(active_context.tenant, TENANT)
|
|
self.assertEqual(active_context.user_id, USER_ID)
|
|
self.assertEqual(active_context.scope_type, "tenant")
|
|
self.assertEqual(profile_value.scope_type, "global")
|
|
self.assertEqual(factor.user_id, USER_ID)
|
|
|
|
def test_durable_payloads_are_raw_state_not_diagnostics(self):
|
|
store = InMemoryUserEngineStore()
|
|
store.migrate()
|
|
records = reference_store_records(store)
|
|
|
|
factor = store_record_for(records["factor"])
|
|
access_profile = store_record_for(records["access_profile"])
|
|
|
|
self.assertIn(RAW_FACTOR_VALUE, json.dumps(factor.payload))
|
|
self.assertIn(PROFILE_SECRET_VALUE, json.dumps(access_profile.payload))
|
|
|
|
def test_unknown_record_type_fails_closed(self):
|
|
with self.assertRaises(ValueError):
|
|
domain_record_from_store_record(
|
|
StoreRecord(record_type="unknown", record_key="1", payload={})
|
|
)
|
|
|
|
with self.assertRaises(TypeError):
|
|
store_record_for(object())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|