Coverage for src/tests/conftest.py: 99%

1"""Module for sharing fixtures in this module.""" 

2 

3from typing import AsyncGenerator, AsyncIterator, Iterator 

4 

5import pytest 

6import pytest_asyncio 

7 

8from kwai_bc_identity.users.user import UserEntity 

9from kwai_bc_identity.users.user_account import UserAccountEntity 

10from kwai_bc_identity.users.user_account_db_repository import ( 

11 UserAccountDbRepository, 

12) 

13from kwai_core.db.database import Database 

14from kwai_core.db.uow import UnitOfWork 

15from kwai_core.domain.value_objects.email_address import EmailAddress 

16from kwai_core.domain.value_objects.name import Name 

17from kwai_core.domain.value_objects.owner import Owner 

18from kwai_core.domain.value_objects.password import Password 

19from kwai_core.events.publisher import Publisher 

20from kwai_core.events.redis_bus import RedisBus 

21from kwai_core.mail.mailer import Mailer 

22from kwai_core.mail.recipient import Recipient, Recipients 

23from kwai_core.settings import get_settings 

24from kwai_core.template.jinja2_engine import Jinja2Engine 

25from kwai_core.template.template_engine import TemplateEngine 

26from redis.asyncio import Redis 

27 

28 

29@pytest_asyncio.fixture(scope="session") 

30async def database(): 

31 """Fixture for getting a database.""" 

32 db = Database(get_settings().db) 

33 yield db 

34 await db.close() 

35 

36 

37@pytest_asyncio.fixture(scope="session") 

38async def redis() -> AsyncGenerator[Redis, None]: 

39 """Fixture for a redis instance.""" 

40 settings = get_settings() 

41 redis = Redis( 

42 host=settings.redis.host, 

43 port=settings.redis.port, 

44 password=settings.redis.password, 

45 ) 

46 yield redis 

47 await redis.aclose() 

48 

49 

50@pytest_asyncio.fixture(scope="session") 

51async def bus(redis: Redis) -> RedisBus: 

52 """Fixture for a message bus.""" 

53 return RedisBus(redis) 

54 

55 

56@pytest_asyncio.fixture(scope="session") 

57async def publisher(bus: RedisBus) -> Publisher: 

58 """Fixture for a publisher.""" 

59 return bus 

60 

61 

62@pytest.fixture(scope="session") 

63def mailer() -> Iterator[Mailer]: 

64 """Fixture for getting a mailer.""" 

65 from kwai_core.mail.smtp_mailer import SmtpMailer 

66 

67 settings = get_settings() 

68 mailer = SmtpMailer( 

69 host=settings.email.host, port=settings.email.port, tls=settings.email.tls 

70 ) 

71 try: 

72 mailer.connect() 

73 if settings.email.user: 

74 mailer.login(settings.email.user, settings.email.password) 

75 yield mailer 

76 finally: 

77 mailer.disconnect() 

78 

79 

80@pytest.fixture 

81def recipients() -> Recipients: 

82 """Fixture for getting recipients.""" 

83 return Recipients( 

84 from_=Recipient(email=EmailAddress("webmaster@kwai.com"), name="Webmaster"), 

85 to=[Recipient(email=EmailAddress("jigoro.kano@kwai.com"), name="Jigoro Kano")], 

86 ) 

87 

88 

89@pytest.fixture(scope="module") 

90def template_engine() -> TemplateEngine: 

91 """Fixture for getting a template engine.""" 

92 settings = get_settings() 

93 return Jinja2Engine(website=settings.website) 

94 

95 

96@pytest_asyncio.fixture(scope="module") 

97async def user_account(database: Database) -> AsyncIterator[UserAccountEntity]: 

98 """Fixture that provides a user account in the database. 

99 

100 The user will be removed again after running the tests. 

101 """ 

102 user_account = UserAccountEntity( 

103 user=UserEntity( 

104 email=EmailAddress("jigoro.kano@kwai.com"), 

105 name=Name(first_name="Jigoro", last_name="Kano"), 

106 ), 

107 password=Password.create_from_string("Nage-waza/1882"), 

108 ) 

109 repo = UserAccountDbRepository(database) 

110 

111 async with UnitOfWork(database): 

112 user_account = await repo.create(user_account) 

113 

114 yield user_account 

115 

116 async with UnitOfWork(database): 

117 await repo.delete(user_account) 

118 

119 

120@pytest_asyncio.fixture(scope="module") 

121async def user(user_account: UserAccountEntity) -> UserEntity: 

122 """Fixture that provides a user account in the database. 

123 

124 The user will be removed again after running the tests. 

125 """ 

126 return user_account.user 

127 

128 

129@pytest_asyncio.fixture(scope="module") 

130async def owner(user: UserEntity) -> Owner: 

131 """Fixture for an owner.""" 

132 return Owner(id=user.id, uuid=user.uuid, name=user.name)