generated from coulomb/repo-seed
108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
import unittest
|
|
|
|
from user_engine.adapters.postgres import PostgresUserEngineStore
|
|
from user_engine.domain import User
|
|
from user_engine.testing.postgres_provider import (
|
|
POSTGRES_TEST_DSN_ENV,
|
|
POSTGRES_TEST_RESET_ENV,
|
|
close_postgres_provider_connection,
|
|
connect_postgres_provider,
|
|
postgres_provider_test_config,
|
|
reset_user_engine_postgres_tables,
|
|
)
|
|
from user_engine.testing.store_conformance import (
|
|
assert_user_engine_store_conformance,
|
|
)
|
|
|
|
|
|
class ProviderPostgresConfigTests(unittest.TestCase):
|
|
def test_config_skips_without_dsn(self):
|
|
config, reason = postgres_provider_test_config({})
|
|
|
|
self.assertIsNone(config)
|
|
self.assertIn(POSTGRES_TEST_DSN_ENV, reason or "")
|
|
|
|
def test_config_requires_reset_acknowledgement(self):
|
|
config, reason = postgres_provider_test_config(
|
|
{POSTGRES_TEST_DSN_ENV: "postgresql://example.test/db"}
|
|
)
|
|
|
|
self.assertIsNone(config)
|
|
self.assertIn(POSTGRES_TEST_RESET_ENV, reason or "")
|
|
|
|
def test_config_accepts_dsn_and_reset_acknowledgement(self):
|
|
config, reason = postgres_provider_test_config(
|
|
{
|
|
POSTGRES_TEST_DSN_ENV: "postgresql://example.test/db",
|
|
POSTGRES_TEST_RESET_ENV: "1",
|
|
}
|
|
)
|
|
|
|
self.assertIsNotNone(config)
|
|
self.assertIsNone(reason)
|
|
|
|
|
|
class ProviderPostgresConformanceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.config, reason = postgres_provider_test_config()
|
|
if reason:
|
|
self.skipTest(reason)
|
|
self.connections = []
|
|
|
|
def tearDown(self):
|
|
for connection in self.connections:
|
|
close_postgres_provider_connection(connection)
|
|
if self.config is not None:
|
|
cleanup = connect_postgres_provider(self.config.dsn)
|
|
try:
|
|
reset_user_engine_postgres_tables(cleanup)
|
|
finally:
|
|
close_postgres_provider_connection(cleanup)
|
|
|
|
def test_live_postgres_store_satisfies_store_conformance(self):
|
|
assert_user_engine_store_conformance(self, self._store_factory)
|
|
|
|
def test_live_postgres_migration_readiness(self):
|
|
store = self._store_factory()
|
|
|
|
self.assertFalse(store.ready)
|
|
store.migrate()
|
|
|
|
self.assertTrue(store.ready)
|
|
self.assertEqual(store.schema_version, "0001_initial")
|
|
|
|
def test_live_postgres_upsert_keeps_one_logical_record(self):
|
|
store = self._store_factory()
|
|
store.migrate()
|
|
user = User(user_id="usr_live_upsert", display_name="Original")
|
|
replacement = User(user_id="usr_live_upsert", display_name="Replacement")
|
|
|
|
store.save_user(user)
|
|
store.save_user(replacement)
|
|
|
|
self.assertEqual(store.user(user.user_id), replacement)
|
|
cursor = store.connection.cursor()
|
|
try:
|
|
cursor.execute(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM user_engine_records
|
|
WHERE record_type = %s AND record_key = %s
|
|
""",
|
|
("users", user.user_id),
|
|
)
|
|
self.assertEqual(cursor.fetchone()[0], 1)
|
|
finally:
|
|
cursor.close()
|
|
|
|
def _store_factory(self) -> PostgresUserEngineStore:
|
|
assert self.config is not None
|
|
connection = connect_postgres_provider(self.config.dsn)
|
|
reset_user_engine_postgres_tables(connection)
|
|
self.connections.append(connection)
|
|
return PostgresUserEngineStore(connection)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|