Merge pull request #834 from dod-ccpo/limit-concurrent-logins

Prevent multiple active sessions

tests/utils/test_logging.py

@@ -2,6 +2,7 @@ from io import StringIO
 import json
 import logging
 from uuid import uuid4
+from unittest.mock import Mock
 
 import pytest
 
@@ -62,13 +63,16 @@ def test_json_formatter_for_exceptions(logger, log_stream_content):
     assert log.get("details")
 
 
-def test_request_context_filter(logger, log_stream_content, request_ctx):
-    user = UserFactory.create()
-    uuid = str(uuid4())
+def test_request_context_filter(logger, log_stream_content, request_ctx, monkeypatch):
+    request_uuid = str(uuid4())
+    user_uuid = str(uuid4())
 
-    request_ctx.g.current_user = user
-    request_ctx.request.environ["HTTP_X_REQUEST_ID"] = uuid
+    user = Mock(spec=["id"])
+    user.id = user_uuid
+
+    monkeypatch.setattr("atst.utils.logging.g", Mock(current_user=user))
+    request_ctx.request.environ["HTTP_X_REQUEST_ID"] = request_uuid
     logger.info("this user is doing something")
     log = json.loads(log_stream_content())
-    assert log["user_id"] == str(user.id)
-    assert log["request_id"] == uuid
+    assert log["user_id"] == str(user_uuid)
+    assert log["request_id"] == request_uuid

tests/utils/test_session_limiter.py

@@ -0,0 +1,51 @@
+import pytest
+from redis import Redis
+from unittest.mock import Mock
+from uuid import uuid4
+
+from atst.utils.session_limiter import SessionLimiter
+from tests.factories import UserFactory
+from atst.models.user import User
+
+
+@pytest.fixture
+def mock_redis():
+    return Mock(spec=Redis)
+
+
+@pytest.fixture
+def mock_session():
+    mock = Mock(spec=["sid"])
+    mock.sid = uuid4()
+    return mock
+
+
+def test_session_limiter_deletes_users_old_session(mock_redis, mock_session):
+    last_session_id = uuid4()
+    current_session_id = uuid4()
+
+    mock_session.sid = current_session_id
+
+    session_limiter = SessionLimiter(
+        {"LIMIT_CONCURRENT_SESSIONS": True}, mock_session, mock_redis
+    )
+    user = UserFactory.create(last_session_id=last_session_id)
+    session_limiter.on_login(user)
+
+    mock_redis.delete.assert_called_with("session:{}".format(last_session_id))
+
+
+def test_session_limiter_updates_users_last_sesion_id(mock_redis, mock_session, db):
+    last_session_id = uuid4()
+    current_session_id = uuid4()
+
+    mock_session.sid = current_session_id
+
+    session_limiter = SessionLimiter(
+        {"LIMIT_CONCURRENT_SESSIONS": True}, mock_session, mock_redis
+    )
+    user = UserFactory.create(last_session_id=last_session_id)
+    session_limiter.on_login(user)
+
+    user = db.session.query(User).get(user.id)
+    assert user.last_session_id == current_session_id