wag-managment-api-service-version-3/testers/test_token_management.py

import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
from datetime import datetime
import time
from unittest.mock import MagicMock, patch

from api_objects import OccupantTokenObject, EmployeeTokenObject, UserType
from ApiServices.api_handlers.auth_actions.token import AccessObjectActions
from api_services.redis.functions import RedisActions
from api_library.date_time_actions.date_functions import system_arrow
from api_configs import Auth


class MockRequest:
    def __init__(self, headers=None):
        self.headers = headers or {}


def test_save_object_to_redis_success():
    """Test successful token save"""
    # Create test data
    access_token = "test_token_123"
    model_object = OccupantTokenObject(
        domain="test.com",
        user_type=UserType.occupant.value,
        user_uu_id="test-uuid",
        credentials={},
        user_id=1,
        person_id=1,
        person_uu_id="person-uuid",
        request={},
        available_occupants={},
    )

    # Test save
    result = AccessObjectActions.save_object_to_redis(
        access_token=access_token, model_object=model_object, expiry_minutes=1
    )
    assert result is True

    # Verify saved data
    saved_data = RedisActions.get_object_via_access_key(
        MockRequest(headers={Auth.ACCESS_TOKEN_TAG: access_token})
    )
    assert saved_data.user_uu_id == model_object.user_uu_id
    assert saved_data.domain == model_object.domain


def test_save_object_to_redis_expiry():
    """Test token expiration"""
    # Create test data
    access_token = "test_token_456"
    model_object = OccupantTokenObject(
        domain="test.com",
        user_type=UserType.occupant.value,
        user_uu_id="test-uuid-2",
        credentials={},
        user_id=2,
        person_id=2,
        person_uu_id="person-uuid-2",
        request={},
        available_occupants={},
    )

    # Save with 2 second expiry
    AccessObjectActions.save_object_to_redis(
        access_token=access_token,
        model_object=model_object,
        expiry_minutes=2 / 60,  # 2 seconds
    )

    # Verify token exists
    token1 = RedisActions.get_object_via_access_key(
        MockRequest(headers={Auth.ACCESS_TOKEN_TAG: access_token})
    )
    assert token1 is not None

    # Wait for expiration
    time.sleep(3)

    # Verify token is expired
    with pytest.raises(Exception) as exc_info:
        RedisActions.get_object_via_access_key(
            MockRequest(headers={Auth.ACCESS_TOKEN_TAG: access_token})
        )
    assert any(
        msg in str(exc_info.value) for msg in ["Token expired", "Invalid credentials"]
    )


def test_get_object_via_user_uu_id():
    """Test retrieving all tokens for a user"""
    user_id = "test-uuid-3"

    # Create multiple tokens for same user
    tokens = []
    for i in range(3):
        access_token = f"test_token_{i}"
        model_object = OccupantTokenObject(
            domain=f"test{i}.com",
            user_type=UserType.occupant.value,
            user_uu_id=user_id,
            credentials={},
            user_id=3,
            person_id=3,
            person_uu_id="person-uuid-3",
            request={},
            available_occupants={},
        )
        AccessObjectActions.save_object_to_redis(access_token, model_object)
        tokens.append(access_token)

    # Get all tokens
    user_tokens = AccessObjectActions.get_object_via_user_uu_id(user_id)
    assert len(user_tokens) == 3

    # Verify each token
    for token in tokens:
        found = False
        for key, data in user_tokens.items():
            if token in key:
                found = True
                break
        assert found, f"Token {token} not found in user tokens"


def test_invalid_token_handling():
    """Test handling of invalid tokens"""
    # Test missing headers
    with pytest.raises(HTTPException) as exc_info:
        AccessObjectActions.get_token_object(MockRequest())
    assert "Unauthorized user" in str(exc_info.value.detail["message"])

    # Test missing token
    with pytest.raises(HTTPException) as exc_info:
        AccessObjectActions.get_token_object(MockRequest(headers={}))
    assert "Unauthorized user" in str(exc_info.value.detail["message"])

    # Test invalid token
    with pytest.raises(HTTPException) as exc_info:
        AccessObjectActions.get_token_object(
            MockRequest(headers={Auth.ACCESS_TOKEN_TAG: "invalid_token"})
        )
    assert "Invalid credentials" in str(exc_info.value.detail["message"])


def test_timezone_handling():
    """Test timezone awareness in token handling"""
    access_token = "test_token_tz"
    model_object = OccupantTokenObject(
        domain="test.com",
        user_type=UserType.occupant.value,
        user_uu_id="test-uuid-tz",
        credentials={},
        user_id=4,
        person_id=4,
        person_uu_id="person-uuid-4",
        request={},
        available_occupants={},
    )

    # Save token
    AccessObjectActions.save_object_to_redis(access_token, model_object)

    # Get token and verify timezone data
    token_data = RedisActions.get_object_via_access_key(
        MockRequest(headers={Auth.ACCESS_TOKEN_TAG: access_token})
    )

    # Verify expiry time is in future
    assert system_arrow.from_timestamp(token_data.expires_at) > system_arrow.now()


if __name__ == "__main__":
    pytest.main([__file__, "-v"])