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

1"""Module with tests for the access token database repository.""" 

2 

3import pytest 

4 

5from kwai_bc_identity.tokens.access_token_repository import ( 

6 AccessTokenNotFoundException, 

7 AccessTokenRepository, 

8) 

9 

10 

11pytestmark = pytest.mark.db 

12 

13 

14async def test_create(make_access_token_in_db): 

15 """Test the creation of an access_token.""" 

16 access_token = await make_access_token_in_db() 

17 assert access_token is not None, "There should be an access token entity" 

18 

19 

20async def test_get_by_token_identifier( 

21 access_token_repo: AccessTokenRepository, 

22 make_access_token_in_db, 

23): 

24 """Test get_by_token_identifier.""" 

25 access_token = await make_access_token_in_db() 

26 token = await access_token_repo.get_by_identifier(access_token.identifier) 

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

28 

29 

30async def test_query( 

31 access_token_repo: AccessTokenRepository, 

32 make_access_token_in_db, 

33): 

34 """Test query.""" 

35 await make_access_token_in_db() 

36 tokens = [token async for token in access_token_repo.get_all(limit=10)] 

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

38 

39 

40async def test_delete( 

41 access_token_repo: AccessTokenRepository, make_access_token_in_db 

42): 

43 """Test delete.""" 

44 access_token = await make_access_token_in_db() 

45 await access_token_repo.delete(access_token) 

46 

47 with pytest.raises(AccessTokenNotFoundException): 

48 await access_token_repo.get_by_identifier(access_token.identifier)