From 1dac42bf2f6988c87fde338dff61474bd6b8a5a1 Mon Sep 17 00:00:00 2001 From: = Date: Sat, 21 Jun 2025 22:54:55 +0300 Subject: [PATCH] refactor: improve type assertions in tests and session backends import structure --- crudadmin/session/backends/__init__.py | 14 ++- tests/crud/test_session_backend_parameters.py | 113 +++++++++++++----- 2 files changed, 98 insertions(+), 29 deletions(-) diff --git a/crudadmin/session/backends/__init__.py b/crudadmin/session/backends/__init__.py index 1093a50..7639ecf 100644 --- a/crudadmin/session/backends/__init__.py +++ b/crudadmin/session/backends/__init__.py @@ -1,3 +1,15 @@ """Session storage backends for different storage systems.""" -__all__ = ["MemorySessionStorage", "RedisSessionStorage", "MemcachedSessionStorage"] +from .database import DatabaseSessionStorage +from .hybrid import HybridSessionStorage +from .memcached import MemcachedSessionStorage +from .memory import MemorySessionStorage +from .redis import RedisSessionStorage + +__all__ = ( + "MemorySessionStorage", + "RedisSessionStorage", + "MemcachedSessionStorage", + "DatabaseSessionStorage", + "HybridSessionStorage", +) diff --git a/tests/crud/test_session_backend_parameters.py b/tests/crud/test_session_backend_parameters.py index e0de4df..9f8318f 100644 --- a/tests/crud/test_session_backend_parameters.py +++ b/tests/crud/test_session_backend_parameters.py @@ -3,6 +3,7 @@ import pytest from crudadmin import CRUDAdmin +from crudadmin.session import backends as session_backends from tests.crud.test_admin import create_test_db_config @@ -25,17 +26,23 @@ async def test_redis_url_configuration(self, async_session): try: # Test basic URL admin.use_redis_sessions(redis_url="redis://localhost:6379/0") - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) # Test URL with password admin.use_redis_sessions(redis_url="redis://user:pass@localhost:6379/1") - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) # Test complex URL admin.use_redis_sessions( redis_url="redis://admin:secret123@redis.example.com:6380/2" ) - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) except ImportError: pytest.skip("Redis not available") @@ -56,23 +63,33 @@ async def test_redis_individual_parameters(self, async_session): try: # Test basic parameters admin.use_redis_sessions(host="localhost", port=6379, db=0) - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) # Test with password admin.use_redis_sessions( host="localhost", port=6379, db=1, password="secret" ) - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) # Test partial parameters (others should use defaults) admin.use_redis_sessions(host="custom-host") - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) admin.use_redis_sessions(port=6380) - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) admin.use_redis_sessions(db=3) - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) except ImportError: pytest.skip("Redis not available") @@ -92,7 +109,9 @@ async def test_redis_defaults(self, async_session): try: admin.use_redis_sessions() - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) except ImportError: pytest.skip("Redis not available") @@ -172,13 +191,18 @@ async def test_redis_additional_parameters(self, async_session): admin.use_redis_sessions( redis_url="redis://localhost:6379/0", pool_size=20, connect_timeout=10 ) - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) # Test with additional parameters via individual params admin.use_redis_sessions( host="localhost", port=6379, db=0, pool_size=15, connect_timeout=5 ) - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) except ImportError: pytest.skip("Redis not available") @@ -384,15 +408,21 @@ async def test_memcached_servers_configuration(self, async_session): try: # Test single server admin.use_memcached_sessions(servers=["localhost:11211"]) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) # Test multiple servers admin.use_memcached_sessions(servers=["localhost:11211", "server2:11211"]) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) # Test custom ports admin.use_memcached_sessions(servers=["localhost:11212", "server2:11213"]) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) except ImportError: pytest.skip("Memcached not available") @@ -413,22 +443,32 @@ async def test_memcached_individual_parameters(self, async_session): try: # Test basic parameters admin.use_memcached_sessions(host="localhost", port=11211) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) # Test custom host admin.use_memcached_sessions(host="memcached.example.com", port=11211) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) # Test custom port admin.use_memcached_sessions(host="localhost", port=11212) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) # Test partial parameters admin.use_memcached_sessions(host="custom-host") - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) admin.use_memcached_sessions(port=11213) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) except ImportError: pytest.skip("Memcached not available") @@ -448,7 +488,9 @@ async def test_memcached_defaults(self, async_session): try: admin.use_memcached_sessions() - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) except ImportError: pytest.skip("Memcached not available") @@ -510,11 +552,15 @@ async def test_memcached_additional_parameters(self, async_session): try: # Test with additional parameters via servers admin.use_memcached_sessions(servers=["localhost:11211"], pool_size=20) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) # Test with additional parameters via individual params admin.use_memcached_sessions(host="localhost", port=11211, pool_size=15) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) except ImportError: pytest.skip("Memcached not available") @@ -602,20 +648,27 @@ async def test_track_sessions_in_db_flag_management(self, async_session): # Switch to database sessions admin.use_database_sessions() assert admin.track_sessions_in_db is True - assert "DatabaseSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.DatabaseSessionStorage + ) # Switch to memory sessions (should reset flag) admin.use_memory_sessions() assert admin.track_sessions_in_db is False - assert "MemorySessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemorySessionStorage + ) # Switch to Redis with explicit tracking try: admin.use_redis_sessions(host="localhost", track_sessions_in_db=True) assert admin.track_sessions_in_db is True # Should be HybridSessionStorage when Redis + DB tracking - storage_name = type(admin.session_manager.storage).__name__ - assert storage_name in ["HybridSessionStorage", "RedisSessionStorage"] + storage_name = type(admin.session_manager.storage) + assert storage_name in [ + session_backends.HybridSessionStorage, + session_backends.RedisSessionStorage, + ] except ImportError: pytest.skip("Redis not available") @@ -639,7 +692,9 @@ async def test_redis_positional_argument(self, async_session): try: # Old way should still work admin.use_redis_sessions("redis://localhost:6379/0") - assert "RedisSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.RedisSessionStorage + ) except ImportError: pytest.skip("Redis not available") @@ -659,7 +714,9 @@ async def test_memcached_positional_argument(self, async_session): try: # Old way should still work admin.use_memcached_sessions(["localhost:11211"]) - assert "MemcachedSessionStorage" in str(type(admin.session_manager.storage)) + assert isinstance( + admin.session_manager.storage, session_backends.MemcachedSessionStorage + ) except ImportError: pytest.skip("Memcached not available")