Coverage for src/tests/modules/identity/tokens/test_refresh_token_db_repository.py: 100%

1"""Module for testing the refresh token database repository.""" 

2 

3import pytest 

4 

5from kwai_bc_identity.tokens.refresh_token_repository import ( 

6 RefreshTokenNotFoundException, 

7 RefreshTokenRepository, 

8) 

9 

10 

11pytestmark = pytest.mark.db 

12 

13 

14async def test_create(make_refresh_token_in_db): 

15 """Test the create method.""" 

16 refresh_token = await make_refresh_token_in_db() 

17 assert refresh_token is not None, "There should be a refresh token entity" 

18 

19 

20async def test_get_by_token_identifier( 

21 refresh_token_repo: RefreshTokenRepository, make_refresh_token_in_db 

22): 

23 """Test get_by_token_identifier.""" 

24 refresh_token = await make_refresh_token_in_db() 

25 token = await refresh_token_repo.get_by_token_identifier(refresh_token.identifier) 

26 assert token, "There should be a refresh token" 

27 

28 

29async def test_query( 

30 refresh_token_repo: RefreshTokenRepository, 

31 make_refresh_token_in_db, 

32): 

33 """Test query.""" 

34 await make_refresh_token_in_db() 

35 tokens = [token async for token in refresh_token_repo.get_all(limit=10)] 

36 assert len(tokens) > 0, "There should be at least 1 token" 

37 

38 

39async def test_delete( 

40 refresh_token_repo: RefreshTokenRepository, make_refresh_token_in_db 

41): 

42 """Test delete.""" 

43 refresh_token = await make_refresh_token_in_db() 

44 await refresh_token_repo.delete(refresh_token) 

45 

46 with pytest.raises(RefreshTokenNotFoundException): 

47 await refresh_token_repo.get_by_token_identifier(refresh_token.identifier)