Sms structuring and tests

Author: maxkahan

http_client/src/vonage_http_client/http_client.py

@@ -104,18 +104,20 @@ def post(
         host: str,
         request_path: str = '',
         params: dict = None,
-        auth_type: str = 'jwt',
+        auth_type: Literal['jwt', 'basic', 'signature'] = 'jwt',
+        body_type: Literal['json', 'data'] = 'json',
     ) -> Union[dict, None]:
-        return self.make_request('POST', host, request_path, params, auth_type)
+        return self.make_request('POST', host, request_path, params, auth_type, body_type)
 
     def get(
         self,
         host: str,
         request_path: str = '',
         params: dict = None,
-        auth_type: str = 'jwt',
+        auth_type: Literal['jwt', 'basic', 'signature'] = 'jwt',
+        body_type: Literal['json', 'data'] = 'json',
     ) -> Union[dict, None]:
-        return self.make_request('GET', host, request_path, params, auth_type)
+        return self.make_request('GET', host, request_path, params, auth_type, body_type)
 
     @validate_call
     def make_request(
@@ -125,6 +127,7 @@ def make_request(
         request_path: str = '',
         params: Optional[dict] = None,
         auth_type: Literal['jwt', 'basic', 'signature'] = 'jwt',
+        body_type: Literal['json', 'data'] = 'json',
     ):
         url = f'https://{host}{request_path}'
         logger.debug(
@@ -137,22 +140,20 @@ def make_request(
         elif auth_type == 'signature':
             params['api_key'] = self._auth.api_key
             params['sig'] = self._auth.sign_params(params)
-            with self._session.request(
-                request_type,
-                url,
-                data=params,
-                headers=self._headers,
-                timeout=self._timeout,
-            ) as response:
-                return self._parse_response(response)
-
-        with self._session.request(
-            request_type,
-            url,
-            json=params,
-            headers=self._headers,
-            timeout=self._timeout,
-        ) as response:
+
+        request_params = {
+            'method': request_type,
+            'url': url,
+            'headers': self._headers,
+            'timeout': self._timeout,
+        }
+
+        if body_type == 'json':
+            request_params['json'] = params
+        else:
+            request_params['data'] = params
+
+        with self._session.request(**request_params) as response:
             return self._parse_response(response)
 
     def append_to_user_agent(self, string: str):

http_client/tests/test_auth.py

@@ -163,33 +163,21 @@ def test_sign_params_with_dynamic_timestamp(mock_time):
     assert signed_params == 'bc7e95bb4e341090b3a202a2885903a5'
 
 
-def test_check_signature_with_valid_signature():
+def test_check_signature_valid_signature():
     auth = Auth(api_key=api_key, signature_secret=signature_secret)
     params = {
-        'param1': 'value1',
-        'param2': 'value2',
-        'sig': 'valid_signature',
+        'param': 'value',
         'timestamp': 1234567890,
+        'sig': '655a4d0b7f064dff438defc52b012cf5',
     }
-
     assert auth.check_signature(params) == True
 
 
-# def test_check_signature_with_invalid_signature():
-#     auth = Auth(signature_secret='signature_secret')
-#     params = {'param1': 'value1', 'param2': 'value2', 'sig': 'invalid_signature'}
-#     expected_signature = hmac.new(
-#         b'signature_secret', b'param1value1param2value2', hashlib.sha256
-#     ).hexdigest()
-
-#     assert auth.check_signature(params) == False
-
-
-# def test_check_signature_with_empty_signature():
-#     auth = Auth(signature_secret='signature_secret')
-#     params = {'param1': 'value1', 'param2': 'value2', 'sig': ''}
-#     expected_signature = hmac.new(
-#         b'signature_secret', b'param1value1param2value2', hashlib.sha256
-#     ).hexdigest()
-
-#     assert auth.check_signature(params) == False
+def test_check_signature_invalid_signature():
+    auth = Auth(api_key=api_key, signature_secret=signature_secret)
+    params = {
+        'param': 'value',
+        'timestamp': 1234567890,
+        'sig': 'invalid_signature',
+    }
+    assert auth.check_signature(params) == False

http_client/tests/test_http_client.py

@@ -131,6 +131,7 @@ def test_make_post_request_with_signature():
         request_path='/post_signed_params',
         params=params,
         auth_type='signature',
+        body_type='data',
     )
     assert res['hello'] == 'world!'
 

sms/src/vonage_sms/models.py

@@ -0,0 +1,54 @@
+from typing import Literal, Optional
+
+from pydantic import (
+    BaseModel,
+    Field,
+    ValidationInfo,
+    field_validator,
+    model_validator,
+)
+
+
+class SmsMessage(BaseModel):
+    """Message object containing the data and options for an SMS message."""
+
+    to: str
+    from_: str = Field(..., serialization_alias='from')
+    text: str
+    sig: Optional[str] = Field(None, min_length=16, max_length=60)
+    client_ref: Optional[str] = Field(
+        None, serialization_alias='client-ref', max_length=100
+    )
+    type: Optional[Literal['text', 'binary', 'unicode']] = None
+    ttl: Optional[int] = Field(None, ge=20000, le=604800000)
+    status_report_req: Optional[bool] = Field(
+        None, serialization_alias='status-report-req'
+    )
+    callback: Optional[str] = Field(None, max_length=100)
+    message_class: Optional[int] = Field(
+        None, serialization_alias='message-class', ge=0, le=3
+    )
+    body: Optional[str] = None
+    udh: Optional[str] = None
+    protocol_id: Optional[int] = Field(
+        None, serialization_alias='protocol-id', ge=0, le=255
+    )
+    account_ref: Optional[str] = Field(None, serialization_alias='account-ref')
+    entity_id: Optional[str] = Field(None, serialization_alias='entity-id')
+    content_id: Optional[str] = Field(None, serialization_alias='content-id')
+
+    @field_validator('body', 'udh')
+    @classmethod
+    def validate_body(cls, value, info: ValidationInfo):
+        data = info.data
+        if 'type' not in data or not data['type'] == 'binary':
+            raise ValueError(
+                'This parameter can only be set when the "type" parameter is set to "binary".'
+            )
+        return value
+
+    @model_validator(mode='after')
+    def validate_type(self) -> 'SmsMessage':
+        if self.type == 'binary' and self.body is None and self.udh is None:
+            raise ValueError('This parameter is required for binary messages.')
+        return self

sms/src/vonage_sms/responses.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+from typing import List, Optional
+
+
+@dataclass
+class MessageResponse:
+    to: str
+    message_id: str
+    status: str
+    remaining_balance: str
+    message_price: str
+    network: str
+    client_ref: Optional[str] = None
+    account_ref: Optional[str] = None
+
+
+@dataclass
+class SmsResponse:
+    message_count: str
+    messages: List[MessageResponse]

sms/src/vonage_sms/sms.py

@@ -1,65 +1,19 @@
 from copy import deepcopy
-from dataclasses import dataclass
-from typing import List, Literal, Optional
 
-from pydantic import BaseModel, Field, field_validator, validate_call
+from pydantic import validate_call
 from vonage_http_client.http_client import HttpClient
 
 from .errors import PartialFailureError, SmsError
-
-
-class SmsMessage(BaseModel):
-    to: str
-    from_: str = Field(..., alias="from")
-    text: str
-    sig: Optional[str] = Field(None, min_length=16, max_length=60)
-    client_ref: Optional[str] = Field(None, alias="client-ref", max_length=100)
-    type: Optional[Literal['text', 'binary', 'unicode']] = None
-    ttl: Optional[int] = Field(None, ge=20000, le=604800000)
-    status_report_req: Optional[bool] = Field(None, alias='status-report-req')
-    callback: Optional[str] = Field(None, max_length=100)
-    message_class: Optional[int] = Field(None, alias='message-class', ge=0, le=3)
-    body: Optional[str] = None
-    udh: Optional[str] = None
-    protocol_id: Optional[int] = Field(None, alias='protocol-id', ge=0, le=255)
-    account_ref: Optional[str] = Field(None, alias='account-ref')
-    entity_id: Optional[str] = Field(None, alias='entity-id')
-    content_id: Optional[str] = Field(None, alias='content-id')
-
-    @field_validator('body', 'udh')
-    @classmethod
-    def validate_body(cls, value, values):
-        if 'type' not in values or not values['type'] == 'binary':
-            raise ValueError(
-                'This parameter can only be set when the "type" parameter is set to "binary".'
-            )
-        if values['type'] == 'binary' and not value:
-            raise ValueError('This parameter is required for binary messages.')
-
-
-@dataclass
-class MessageResponse:
-    to: str
-    message_id: str
-    status: str
-    remaining_balance: str
-    message_price: str
-    network: str
-    client_ref: Optional[str] = None
-    account_ref: Optional[str] = None
-
-
-@dataclass
-class SmsResponse:
-    message_count: str
-    messages: List[MessageResponse]
+from .models import SmsMessage
+from .responses import MessageResponse, SmsResponse
 
 
 class Sms:
     """Calls Vonage's SMS API."""
 
     def __init__(self, http_client: HttpClient) -> None:
         self._http_client = deepcopy(http_client)
+        self._body_type = 'data'
         if self._http_client._auth._signature_secret:
             self._auth_type = 'signature'
         else:
@@ -73,6 +27,7 @@ def send(self, message: SmsMessage) -> SmsResponse:
             '/sms/json',
             message.model_dump(by_alias=True),
             self._auth_type,
+            self._body_type,
         )
 
         if int(response['message-count']) > 1:

sms/tests/data/default.json

@@ -0,0 +1 @@
+{}