Coverage for src/integrify/epoint/handlers.py: 100%

1import base64 

2import json 

3 

4import httpx 

5 

6from integrify.api import APIPayloadHandler 

7from integrify.epoint import env 

8from integrify.epoint.helper import generate_signature 

9from integrify.epoint.schemas.enums import TransactionStatus, TransactionStatusExtended 

10from integrify.epoint.schemas.request import ( 

11 GetTransactionStatusRequestSchema, 

12 PayAndSaveCardRequestSchema, 

13 PaymentRequestSchema, 

14 PayoutRequestSchema, 

15 PayWithSavedCardRequestSchema, 

16 RefundRequestSchema, 

17 SaveCardRequestSchema, 

18 SplitPayAndSaveCardRequestSchema, 

19 SplitPayRequestSchema, 

20 SplitPayWithSavedCardRequestSchema, 

21) 

22from integrify.epoint.schemas.response import ( 

23 BaseResponseSchema, 

24 MinimalResponseSchema, 

25 RedirectUrlResponseSchema, 

26 RedirectUrlWithCardIdResponseSchema, 

27 SplitPayWithSavedCardResponseSchema, 

28 TransactionStatusResponseSchema, 

29) 

30from integrify.schemas import APIResponse, PayloadBaseModel, _ResponseT 

31 

32 

33class BasePayloadHandler(APIPayloadHandler): 

34 def __init__(self, req_model: type[PayloadBaseModel], resp_model: type[_ResponseT]): 

35 super().__init__(req_model, resp_model) 

36 

37 def pre_handle_payload(self, *args, **kwds): 

38 return { 

39 'public_key': env.EPOINT_PUBLIC_KEY, 

40 'language': env.EPOINT_INTERFACE_LANG, 

41 } 

42 

43 def post_handle_payload(self, data: dict): 

44 b64data = base64.b64encode(json.dumps(data).encode()).decode() 

45 return { 

46 'data': b64data, 

47 'signature': generate_signature(b64data), 

48 } 

49 

50 def handle_response(self, resp: httpx.Response) -> APIResponse[_ResponseT]: 

51 api_resp: APIResponse[MinimalResponseSchema] = super().handle_response(resp) # type: ignore[assignment] 

52 

53 # EPoint həmişə 200 qaytarır, error olsa belə 

54 if isinstance(api_resp.body.status, TransactionStatusExtended): 

55 api_resp.ok = api_resp.body.status != TransactionStatusExtended.SERVER_ERROR 

56 else: 

57 api_resp.ok = api_resp.body.status == TransactionStatus.SUCCESS 

58 

59 return api_resp # type: ignore[return-value] 

60 

61 

62class PaymentPayloadHandler(BasePayloadHandler): 

63 def __init__(self): 

64 super().__init__(PaymentRequestSchema, RedirectUrlResponseSchema) 

65 

66 

67class GetTransactionStatusPayloadHandler(BasePayloadHandler): 

68 def __init__(self): 

69 super().__init__(GetTransactionStatusRequestSchema, TransactionStatusResponseSchema) 

70 

71 

72class SaveCardPayloadHandler(BasePayloadHandler): 

73 def __init__(self): 

74 super().__init__(SaveCardRequestSchema, RedirectUrlWithCardIdResponseSchema) 

75 

76 

77class PayWithSavedCardPayloadHandler(BasePayloadHandler): 

78 def __init__(self): 

79 super().__init__(PayWithSavedCardRequestSchema, BaseResponseSchema) 

80 

81 

82class PayAndSaveCardPayloadHandler(BasePayloadHandler): 

83 def __init__(self): 

84 super().__init__(PayAndSaveCardRequestSchema, RedirectUrlWithCardIdResponseSchema) 

85 

86 

87class PayoutPayloadHandler(BasePayloadHandler): 

88 def __init__(self): 

89 super().__init__(PayoutRequestSchema, BaseResponseSchema) 

90 

91 

92class RefundPayloadHandler(BasePayloadHandler): 

93 def __init__(self): 

94 super().__init__(RefundRequestSchema, MinimalResponseSchema) 

95 

96 

97class SplitPayPayloadHandler(BasePayloadHandler): 

98 def __init__(self): 

99 super().__init__(SplitPayRequestSchema, RedirectUrlResponseSchema) 

100 

101 

102class SplitPayWithSavedCardPayloadHandler(BasePayloadHandler): 

103 def __init__(self): 

104 super().__init__( 

105 SplitPayWithSavedCardRequestSchema, 

106 SplitPayWithSavedCardResponseSchema, 

107 ) 

108 

109 

110class SplitPayAndSaveCardPayloadHandler(BasePayloadHandler): 

111 def __init__(self): 

112 super().__init__( 

113 SplitPayAndSaveCardRequestSchema, 

114 RedirectUrlWithCardIdResponseSchema, 

115 )