Coverage for src/kwai/modules/portal/update_page.py: 100%

1"""Module for the use case "Update Page".""" 

2 

3from dataclasses import dataclass 

4 

5from kwai.core.domain.entity import Entity 

6from kwai.core.domain.value_objects.owner import Owner 

7from kwai.core.domain.value_objects.text import DocumentFormat, Locale, LocaleText 

8from kwai.modules.portal.applications.application import ApplicationIdentifier 

9from kwai.modules.portal.applications.application_repository import ( 

10 ApplicationRepository, 

11) 

12from kwai.modules.portal.page_command import PageCommand 

13from kwai.modules.portal.pages.page import PageEntity, PageIdentifier 

14from kwai.modules.portal.pages.page_repository import PageRepository 

15 

16 

17@dataclass(kw_only=True, frozen=True, slots=True) 

18class UpdatePageCommand(PageCommand): 

19 """Input for the "Update Page" use case.""" 

20 

21 id: int 

22 

23 

24class UpdatePage: 

25 """Use case for updating a page.""" 

26 

27 def __init__( 

28 self, 

29 repo: PageRepository, 

30 application_repo: ApplicationRepository, 

31 owner: Owner, 

32 ): 

33 """Initialize the use case. 

34 

35 Args: 

36 repo: A repository for updating pages. 

37 application_repo: A repository for getting the application. 

38 owner: The owner of the page. 

39 """ 

40 self._repo = repo 

41 self._application_repo = application_repo 

42 self._owner = owner 

43 

44 async def execute(self, command: UpdatePageCommand) -> PageEntity: 

45 """Execute the use case. 

46 

47 Args: 

48 command: The input for this use case. 

49 

50 Raises: 

51 PageNotFoundException: When the page does not exist. 

52 ApplicationNotFoundException: When the application does not exist. 

53 """ 

54 page = await self._repo.get_by_id(PageIdentifier(command.id)) 

55 application = await self._application_repo.get_by_id( 

56 ApplicationIdentifier(command.application) 

57 ) 

58 

59 page = Entity.replace( 

60 page, 

61 enabled=command.enabled, 

62 application=application, 

63 texts=[ 

64 LocaleText( 

65 locale=Locale(text.locale), 

66 format=DocumentFormat(text.format), 

67 title=text.title, 

68 content=text.content, 

69 summary=text.summary, 

70 author=self._owner, 

71 ) 

72 for text in command.texts 

73 ], 

74 priority=command.priority, 

75 remark=command.remark, 

76 traceable_time=page.traceable_time.mark_for_update(), 

77 ) 

78 

79 await self._repo.update(page) 

80 

81 return page