Skip to content

Improve type assertions in tests and session backends import structure #42

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jun 21, 2025
Merged

Improve type assertions in tests and session backends import structure #42

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion crudadmin/session/backends/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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",
)
113 changes: 85 additions & 28 deletions tests/crud/test_session_backend_parameters.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand All @@ -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")
Expand All @@ -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")
Expand All @@ -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")

Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand All @@ -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")
Expand All @@ -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")

Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")

Expand All @@ -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")

Expand All @@ -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")

Expand Down
Loading