MCP Server

"""Tests for the OAuth models.""" import pytest import uuid from datetime import datetime, timedelta from app.models.oauth import OAuthClient, AuthorizationCode, Token from app.models.user import User @pytest.fixture(scope="function") def test_user(db_session): """Create a test user for OAuth tests.""" user = User( email="oauth_test@example.com", username="oauth_test_user", hashed_password="hashed_password" ) db_session.add(user) db_session.commit() return user @pytest.fixture(scope="function") def test_client(db_session): """Create a test OAuth client.""" client = OAuthClient( client_secret="test_client_secret", client_name="Test Client", redirect_uris=["http://localhost:8000/callback"], scopes=["read", "write"] ) db_session.add(client) db_session.commit() return client def test_oauth_client_creation(db_session): """Test that an OAuth client can be created.""" client = OAuthClient( client_secret="test_client_secret", client_name="Test Client", redirect_uris=["http://localhost:8000/callback"], scopes=["read", "write"] ) db_session.add(client) db_session.commit() # Query the client db_client = db_session.query(OAuthClient).filter( OAuthClient.client_name == "Test Client" ).first() assert db_client is not None assert db_client.client_name == "Test Client" assert db_client.client_secret == "test_client_secret" assert db_client.redirect_uris == ["http://localhost:8000/callback"] assert db_client.scopes == ["read", "write"] assert db_client.is_confidential is True assert isinstance(db_client.client_id, uuid.UUID) assert isinstance(db_client.id, uuid.UUID) def test_authorization_code_creation(db_session, test_user, test_client): """Test that an authorization code can be created.""" expiration_time = (datetime.utcnow() + timedelta(minutes=10)).isoformat() auth_code = AuthorizationCode( code="test_auth_code", client_id=test_client.id, user_id=test_user.id, redirect_uri="http://localhost:8000/callback", scope="read write", expires_at=expiration_time, code_challenge="test_challenge", code_challenge_method="S256" ) db_session.add(auth_code) db_session.commit() # Query the code db_code = db_session.query(AuthorizationCode).filter( AuthorizationCode.code == "test_auth_code" ).first() assert db_code is not None assert db_code.code == "test_auth_code" assert db_code.client_id == test_client.id assert db_code.user_id == test_user.id assert db_code.redirect_uri == "http://localhost:8000/callback" assert db_code.scope == "read write" assert db_code.expires_at == expiration_time assert db_code.code_challenge == "test_challenge" assert db_code.code_challenge_method == "S256" assert db_code.is_expired is False def test_authorization_code_expiration(db_session, test_user, test_client): """Test the authorization code expiration property.""" # Create expired code (10 minutes in the past) past_time = (datetime.utcnow() - timedelta(minutes=10)).isoformat() expired_code = AuthorizationCode( code="expired_code", client_id=test_client.id, user_id=test_user.id, redirect_uri="http://localhost:8000/callback", scope="read", expires_at=past_time ) db_session.add(expired_code) db_session.commit() # Create valid code (10 minutes in the future) future_time = (datetime.utcnow() + timedelta(minutes=10)).isoformat() valid_code = AuthorizationCode( code="valid_code", client_id=test_client.id, user_id=test_user.id, redirect_uri="http://localhost:8000/callback", scope="read", expires_at=future_time ) db_session.add(valid_code) db_session.commit() # Test is_expired property assert expired_code.is_expired is True assert valid_code.is_expired is False def test_token_creation(db_session, test_user, test_client): """Test that a token can be created.""" access_expires = (datetime.utcnow() + timedelta(hours=1)).isoformat() refresh_expires = (datetime.utcnow() + timedelta(days=7)).isoformat() token = Token( access_token="test_access_token", refresh_token="test_refresh_token", client_id=test_client.id, user_id=test_user.id, scope="read write", access_token_expires_at=access_expires, refresh_token_expires_at=refresh_expires ) db_session.add(token) db_session.commit() # Query the token db_token = db_session.query(Token).filter( Token.access_token == "test_access_token" ).first() assert db_token is not None assert db_token.access_token == "test_access_token" assert db_token.refresh_token == "test_refresh_token" assert db_token.client_id == test_client.id assert db_token.user_id == test_user.id assert db_token.scope == "read write" assert db_token.access_token_expires_at == access_expires assert db_token.refresh_token_expires_at == refresh_expires assert db_token.is_revoked is False assert db_token.is_access_token_expired is False assert db_token.is_refresh_token_expired is False def test_token_expiration(db_session, test_user, test_client): """Test token expiration properties.""" # Create token with expired access token but valid refresh token past_access = (datetime.utcnow() - timedelta(hours=1)).isoformat() future_refresh = (datetime.utcnow() + timedelta(days=7)).isoformat() token = Token( access_token="expired_access_token", refresh_token="valid_refresh_token", client_id=test_client.id, user_id=test_user.id, scope="read", access_token_expires_at=past_access, refresh_token_expires_at=future_refresh ) db_session.add(token) db_session.commit() assert token.is_access_token_expired is True assert token.is_refresh_token_expired is False # Create token with both tokens expired past_access = (datetime.utcnow() - timedelta(hours=1)).isoformat() past_refresh = (datetime.utcnow() - timedelta(days=1)).isoformat() expired_token = Token( access_token="expired_access_token2", refresh_token="expired_refresh_token", client_id=test_client.id, user_id=test_user.id, scope="read", access_token_expires_at=past_access, refresh_token_expires_at=past_refresh ) db_session.add(expired_token) db_session.commit() assert expired_token.is_access_token_expired is True assert expired_token.is_refresh_token_expired is True def test_client_relationships(db_session, test_user, test_client): """Test the relationships between OAuth models.""" # Create authorization code auth_code = AuthorizationCode( code="rel_auth_code", client_id=test_client.id, user_id=test_user.id, redirect_uri="http://localhost:8000/callback", scope="read", expires_at=(datetime.utcnow() + timedelta(minutes=10)).isoformat() ) # Create token token = Token( access_token="rel_access_token", refresh_token="rel_refresh_token", client_id=test_client.id, user_id=test_user.id, scope="read", access_token_expires_at=(datetime.utcnow() + timedelta(hours=1)).isoformat(), refresh_token_expires_at=(datetime.utcnow() + timedelta(days=7)).isoformat() ) db_session.add_all([auth_code, token]) db_session.commit() # Refresh the client to update relationships db_session.refresh(test_client) # Test client-authorization_codes relationship assert len(test_client.authorization_codes) == 1 assert test_client.authorization_codes[0].code == "rel_auth_code" # Test client-tokens relationship assert len(test_client.tokens) == 1 assert test_client.tokens[0].access_token == "rel_access_token"