Merge branch 'dev' into 10-termination-codes

Author: dynata-bradley

CmixAPIClient/api.py

@@ -163,6 +163,19 @@ def add_extra_url_params(self, url, params):
 
         return url
 
+    def get_survey_data_layouts(self, survey_id):
+        self.check_auth_headers()
+        data_layouts_url = '{}/surveys/{}/data-layouts'.format(CMIX_SERVICES['survey'][self.url_type], survey_id)
+        data_layouts_response = requests.get(data_layouts_url, headers=self._authentication_headers)
+        if data_layouts_response.status_code != 200:
+            raise CmixError(
+                'CMIX returned a non-200 response code while getting data_layouts: {} and error {}'.format(
+                    data_layouts_response.status_code,
+                    data_layouts_response.text
+                )
+            )
+        return data_layouts_response.json()
+
     def get_survey_definition(self, survey_id):
         self.check_auth_headers()
         definition_url = '{}/surveys/{}/definition'.format(CMIX_SERVICES['survey'][self.url_type], survey_id)
@@ -209,6 +222,19 @@ def get_survey_status(self, survey_id):
             raise CmixError('Get Survey Status returned without a status. Response: {}'.format(status_response.json()))
         return status.lower()
 
+    def get_survey_sources(self, survey_id):
+        self.check_auth_headers()
+        sources_url = '{}/surveys/{}/sources'.format(CMIX_SERVICES['survey'][self.url_type], survey_id)
+        sources_response = requests.get(sources_url, headers=self._authentication_headers)
+        if sources_response.status_code != 200:
+            raise CmixError(
+                'CMIX returned a non-200 response code while getting sources: {} and error {}'.format(
+                    sources_response.status_code,
+                    sources_response.text
+                )
+            )
+        return sources_response.json()
+
     def get_survey_completes(self, survey_id):
         return self.get_survey_respondents(survey_id, "COMPLETE", True)
 
@@ -255,25 +281,14 @@ def create_export_archive(self, survey_id, export_type):
             )
         archive_json = archive_response.json()
 
-        layout_url = '{}/surveys/{}/data-layouts/'.format(CMIX_SERVICES['survey']["BASE_URL"], survey_id)
-        layout_response = requests.get(layout_url, headers=self._authentication_headers)
-        if layout_response.status_code != 200:
-            raise CmixError(
-                'CMIX returned a non-200 response code: {} and error {}'.format(
-                    layout_response.status_code,
-                    layout_response.text
-                )
-            )
+        layout_json = self.get_survey_data_layouts(survey_id)
         layout_id = None
-        for layout in layout_response.json():
+        for layout in layout_json:
             if layout.get('name') == 'Default':
                 layout_id = layout.get('id')
         if layout_id is None:
             raise CmixError(
-                'Layouts response did not contain a "Default" layout. Response Code: {}, Body {}'.format(
-                    layout_response.status_code,
-                    layout_response.content
-                )
+                'Layouts response did not contain a "Default" layout.'
             )
 
         archive_json['dataLayoutId'] = layout_id

README.md

@@ -29,10 +29,12 @@ A Python client library for the [Dynata Cmix API](https://wiki2.criticalmix.net/
     fetch_banner_filter(survey_id, question_a, question_b, response_id)
     fetch_raw_results(survey_id, payload)
     get_surveys(status, *args, **kwargs)
+    get_survey_data_layouts(survey_id)
     get_survey_definition(survey_id)
     get_survey_xml(survey_id)
     get_survey_simulations(survey_id)
     get_survey_termination_codes(survey_id)
+    get_survey_sources(survey_id)
     get_survey_test_url(survey_id)
     get_survey_respondents(survey_id, respondent_type, live)
     get_survey_status(survey_id)

tests/test_api.py

@@ -116,6 +116,32 @@ def test_create_export_archive_errors_handled(self):
                 with self.assertRaises(CmixError):
                     self.cmix_api.create_export_archive(self.survey_id, 'XLSX_READABLE')
 
+    def test_get_survey_data_layouts(self):
+        self.cmix_api._authentication_headers = {'Authentication': 'Bearer test'}
+
+        # success case
+        with mock.patch('CmixAPIClient.api.requests') as mock_request:
+            mock_get = mock.Mock()
+            mock_get.status_code = 200
+            mock_get.json.return_value = {}
+            mock_request.get.return_value = mock_get
+
+            self.cmix_api.get_survey_data_layouts(self.survey_id)
+
+            base_url = CMIX_SERVICES['survey']['BASE_URL']
+            surveys_url = '{}/surveys/{}/data-layouts'.format(base_url, self.survey_id)
+            mock_request.get.assert_any_call(surveys_url, headers=self.cmix_api._authentication_headers)
+
+        # error case (survey not found)
+        with mock.patch('CmixAPIClient.api.requests') as mock_request:
+            mock_get = mock.Mock()
+            mock_get.status_code = 404
+            mock_get.json.return_value = {}
+            mock_request.get.return_value = mock_get
+
+            with self.assertRaises(CmixError):
+                self.cmix_api.get_survey_data_layouts(self.survey_id)
+
     def test_get_survey_status(self):
         self.cmix_api._authentication_headers = {'Authentication': 'Bearer test'}
 
@@ -143,6 +169,32 @@ def test_get_survey_status_error_handled(self):
             with self.assertRaises(CmixError):
                 self.cmix_api.get_survey_status(self.survey_id)
 
+    def test_get_survey_sources(self):
+        self.cmix_api._authentication_headers = {'Authentication': 'Bearer test'}
+
+        # success case
+        with mock.patch('CmixAPIClient.api.requests') as mock_request:
+            mock_get = mock.Mock()
+            mock_get.status_code = 200
+            mock_get.json.return_value = {}
+            mock_request.get.return_value = mock_get
+
+            self.cmix_api.get_survey_sources(self.survey_id)
+
+            base_url = CMIX_SERVICES['survey']['BASE_URL']
+            surveys_url = '{}/surveys/{}/sources'.format(base_url, self.survey_id)
+            mock_request.get.assert_any_call(surveys_url, headers=self.cmix_api._authentication_headers)
+
+        # error case (survey not found)
+        with mock.patch('CmixAPIClient.api.requests') as mock_request:
+            mock_get = mock.Mock()
+            mock_get.status_code = 404
+            mock_get.json.return_value = {}
+            mock_request.get.return_value = mock_get
+
+            with self.assertRaises(CmixError):
+                self.cmix_api.get_survey_sources(self.survey_id)
+
     def test_get_survey_test_url(self):
         self.cmix_api._authentication_headers = {'Authentication': 'Bearer test'}
         correct_test_link = '{}/#/?cmixSvy={}&cmixTest={}'.format(