Coverage for src/galaxy_beep_frame/frame/base.py: 100%

1"""Frame class.""" 

2 

3from dataclasses import InitVar, dataclass, field 

4from typing import TYPE_CHECKING 

5 

6from .state import FrameState 

7 

8if TYPE_CHECKING: 

9 from galaxy_beep_frame.frame import ContinuationIndicator 

10 from galaxy_beep_frame.header_type import PossibleHeaderType 

11 

12 

13@dataclass 

14class Frame: 

15 """ 

16 A frame of the BEEP protocol. 

17 

18 Notes 

19 ----- 

20 See https://datatracker.ietf.org/doc/html/rfc3080. 

21 

22 """ 

23 

24 payload: bytes 

25 state: FrameState = field(init=False) 

26 

27 channel_number: InitVar[int] 

28 message_number: InitVar[int] 

29 sequence_number: InitVar[int] 

30 answer_number: InitVar[int | None] 

31 continuation_indicator: InitVar[ContinuationIndicator] 

32 header_type: InitVar[PossibleHeaderType] 

33 

34 def __post_init__( # noqa: PLR0913, PLR0917 

35 self, 

36 channel_number: int, 

37 message_number: int, 

38 sequence_number: int, 

39 answer_number: int | None, 

40 continuation_indicator: ContinuationIndicator, 

41 possible_header_type: PossibleHeaderType, 

42 ) -> None: 

43 """Post-initialization process.""" 

44 self.state = FrameState( 

45 frame=self, 

46 channel_number=channel_number, 

47 message_number=message_number, 

48 sequence_number=sequence_number, 

49 answer_number=answer_number, 

50 continuation_indicator=continuation_indicator, 

51 possible_header_type=possible_header_type, 

52 ) 

53 

54 def build(self) -> bytes: 

55 """Build a Frame.""" 

56 header = self.state.build(size=len(self.payload)) 

57 return header + b"\r\n" + self.payload + b"END\r\n"