Merge pull request #84 from ovh/support-headers

feat: added support for headers when calling ovh.Client.raw_call

ovh/client.py

@@ -471,7 +471,7 @@ def call(self, method, path, data=None, need_auth=True):
         else:
             raise APIError(json_result.get('message'), response=result)
 
-    def raw_call(self, method, path, data=None, need_auth=True):
+    def raw_call(self, method, path, data=None, need_auth=True, headers=None):
         """
         Lowest level call helper. If ``consumer_key`` is not ``None``, inject
         authentication headers and sign the request.
@@ -490,12 +490,17 @@ def raw_call(self, method, path, data=None, need_auth=True):
         :param str path: api entrypoint to call, relative to endpoint base path
         :param data: any json serializable data to send as request's body
         :param boolean need_auth: if False, bypass signature
+        :param dict headers: A dict containing the headers that should be sent to
+                             the OVH API. ``raw_call`` will override the
+                             OVH API authentication headers, as well as
+                             the Content-Type header.
         """
         body = ''
         target = self._endpoint + path
-        headers = {
-            'X-Ovh-Application': self._application_key
-        }
+
+        if headers is None:
+            headers = {}
+        headers['X-Ovh-Application'] = self._application_key
 
         # include payload
         if data is not None:

tests/test_client.py

@@ -433,6 +433,23 @@ def test_raw_call(self, m_req):
         r = api.raw_call(FAKE_METHOD, FAKE_PATH, None, False)
         self.assertEqual(r, "Let's assume requests will return this")
 
+    @mock.patch('ovh.client.Session.request', return_value="Let's assume requests will return this")
+    def test_raw_call_with_headers(self, m_req):
+        api = Client(ENDPOINT, APPLICATION_KEY, APPLICATION_SECRET)
+        r = api.raw_call(FAKE_METHOD, FAKE_PATH, None, False, headers={
+            'Custom-Header': "1",
+        })
+        self.assertEqual(r, "Let's assume requests will return this")
+        m_req.assert_called_once_with(
+            FAKE_METHOD, BASE_URL+FAKE_PATH,
+            headers={
+                'Custom-Header': "1",
+                'X-Ovh-Application': APPLICATION_KEY,
+            },
+            data='',
+            timeout=TIMEOUT
+        )
+
     # Perform real API tests.
     def test_endpoints(self):
         for endpoint in ENDPOINTS.keys():