Issue 7037 - [XTM Integration] Make REST API url customizable

cms/translations/xtm/xtm_api.py

 # This file is part of the Adblock Plus web scripts,
 # Copyright (C) 2006-present eyeo GmbH
 #
 # Adblock Plus is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License version 3 as
 # published by the Free Software Foundation.
 #
 # Adblock Plus is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with Adblock Plus.  If not, see <http://www.gnu.org/licenses/>.
 
 from __future__ import unicode_literals
 
 import json
+try:
+    import urlparse
+except ImportError:
+    # It's Python 3
+    import urllib.parse as urlparse
 
 import requests
 
 __all__ = [
     'XTMCloudAPI', 'XTMCloudException', 'get_token',
 ]
 
 _BASE_URL = 'https://wstest2.xtm-intl.com/rest-api/'
 
 
@@ skipping 40 matching lines @@
     }
 
     class _SuccessCodes:
         CREATE = 201
         UPLOAD = 200
         DOWNLOAD = 200
         GET_TARGET_LANGS = 200
         ADD_TARGET_LANGS = 200
         GET_WORKFLOW_IDS = 200
 
-    def __init__(self, token):
+    def __init__(self, token, base_url):
         """Constructor.
 
         Parameters
         ----------
         token: str
             Token used to authenticate with the API.
+        base_url: str
+            Url used to connect to the API.
 
         """
         self._token = token
-        self.base_url = _BASE_URL
+        self.base_url = base_url
 
     def _execute(self, url, data=None, files=None, stream=False,
                  params=None, headers=None):
         """Send request to the API and return the response.
 
         Parameters
         ----------
         url: str
             The url we're making the request to.
         data: dict
@@ skipping 111 matching lines @@
         }
 
         if files:
             file_names, files_to_upload = self._construct_files_dict(
                 files, 'translationFiles[{}]')
             data.update(file_names)
         else:
             # Hacky way to go around 415 error code
             files_to_upload = {'a': 'b'}
 
-        url = self.base_url + self._UrlPaths.CREATE
+        url = urlparse.urljoin(self.base_url, self._UrlPaths.CREATE)
 
         response = self._execute(url, data=data, files=files_to_upload)
 
         if response.status_code != self._SuccessCodes.CREATE:
             # The creation was not successful
             raise XTMCloudException(response.status_code,
                                     response.text.decode('utf-8'),
                                     'creating job')
 
         data = json.loads(response.text.encode('utf-8'))
@@ skipping 36 matching lines @@
         file_names, files_to_upload = self._construct_files_dict(
             files, 'files[{}]',
         )
 
         if len(files_to_upload.keys()) == 0:
             raise Exception('Error: No files provided for upload.')
 
         data = {'matchType': self._MATCH_TYPE[overwrite]}
         data.update(file_names)
 
-        url = self.base_url + self._UrlPaths.UPLOAD.format(project_id)
+        url = urlparse.urljoin(
+            self.base_url, self._UrlPaths.UPLOAD.format(project_id),
+        )
 
         response = self._execute(url, data=data, files=files_to_upload)
 
         if response.status_code != self._SuccessCodes.UPLOAD:
             raise XTMCloudException(response.status_code,
                                     response.text.encode('utf-8'),
                                     'uploading files')
 
         return json.loads(response.text.encode('utf-8'))['jobs']
 
@@ skipping 11 matching lines @@
         -------
         bytes
             The contents of the zip file returned by the API
 
         Raises
         ------
         XTMCloudException
             If the request is flawed in any way.
 
         """
-        url = (self.base_url + self._UrlPaths.DOWNLOAD).format(project_id)
+        url = urlparse.urljoin(
+            self.base_url, self._UrlPaths.DOWNLOAD.format(project_id),
+        )
 
         exception_msg = {
             400: 'Invalid request',
             401: 'Authentication failed',
             500: 'Internal server error',
             404: 'Project not found.',
         }
 
         response = self._execute(url, params={'fileType': files_type},
                                  stream=True)
@@ skipping 19 matching lines @@
         -------
         iterable
             With the target languages for this project.
 
         Raises
         ------
         XTMCloudException
             If the request is unsuccessful.
 
         """
-        url = (self.base_url + self._UrlPaths.GET_TARGET_LANG).format(
-            project_id,
+        url = urlparse.urljoin(
+            self.base_url, self._UrlPaths.GET_TARGET_LANG.format(project_id),
         )
 
         response = self._execute(url, stream=True)
 
         if response.status_code != self._SuccessCodes.GET_TARGET_LANGS:
             raise XTMCloudException(response.status_code, response.content,
                                     'extracting target languages')
 
         data = json.loads(response.content.encode('utf-8'))
         return {item['targetLanguage'] for item in data}
@@ skipping 10 matching lines @@
 
         Raises
         -------
         XTMCloudException
             If the request to the API was not successful.
 
         """
         data = json.dumps({
             'targetLanguages': target_languages,
         })
-        url = (self.base_url + self._UrlPaths.ADD_TARGET_LANG).format(
-            project_id,
+
+        url = urlparse.urljoin(
+            self.base_url, self._UrlPaths.ADD_TARGET_LANG.format(project_id),
         )
+
         headers = {'content-type': 'application/json'}
 
         response = self._execute(url, data=data, headers=headers)
 
         if response.status_code != self._SuccessCodes.ADD_TARGET_LANGS:
             raise XTMCloudException(response.status_code, response.content,
                                     'adding target languages to project')
 
     def get_workflows_by_name(self, name):
         """Get workflows with a specific name.
 
         Parameters
         ----------
         name: str
             The name of the workflow we're looking for.
 
         Returns
         -------
         iterable
             Of workflow ids that match the name provided.
 
         """
-        url = self.base_url + self._UrlPaths.GET_WORKFLOW_IDS
+        url = urlparse.urljoin(self.base_url, self._UrlPaths.GET_WORKFLOW_IDS)
 
         response = self._execute(url, params={'name': name})
 
         if response.status_code != self._SuccessCodes.GET_WORKFLOW_IDS:
             raise XTMCloudException(response.status_code, response.content,
                                     'extracting workflow ids')
 
         valid_ids = []
         for item in json.loads(response.content.encode('utf-8')):
             if name.lower().replace(' ', '') == \
                     item['name'].lower().replace(' ', ''):
                 valid_ids.append(item['id'])
         return valid_ids
 
 
-def get_token(username, password, user_id):
+def get_token(username, password, user_id, base_url):
     """Generate an API token from username and password.
 
     Parameters
     ----------
     username: str
         The username used to generate the token.
     password: str
         The password used to generate the token.
     user_id: int
         The user ID used to generate the token.
+    base_url: str
+        The url used to connect to the API
 
     Returns
     -------
     str
         The resulting token generated by the API.
 
     Raises
     ------
     XTMCloudException
         If the credentials provided were invalid.
 
     """
     request_body = json.dumps({
         'client': username,
         'password': password,
         'userId': user_id,
     })
 
-    url = _BASE_URL + 'auth/token'
+    url = urlparse.urljoin(base_url, 'auth/token')
 
     headers = {'content-type': 'application/json'}
 
     response = requests.post(url, data=request_body, headers=headers)
 
     if response.status_code == 200:
         return json.loads(response.text)['token'].encode()
 
     raise XTMCloudException(response.status_code,
                             response.text.encode('utf-8'),
                             'generating token')