diff --git a/.gitignore b/.gitignore index 283d4a09..0aaeb9a6 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,5 @@ htmlcov build dist shotgun_api3.egg-info +/%1 + diff --git a/azure-pipelines-templates/run-tests.yml b/azure-pipelines-templates/run-tests.yml index 4da86608..dcf8926b 100644 --- a/azure-pipelines-templates/run-tests.yml +++ b/azure-pipelines-templates/run-tests.yml @@ -48,6 +48,8 @@ jobs: python.version: '2.7' Python37: python.version: '3.7' + Python39: + python.version: '3.9' # These are the steps that will be executed inside each job. steps: diff --git a/shotgun_api3/lib/six.py b/shotgun_api3/lib/six.py index 357e624a..b22d2e57 100644 --- a/shotgun_api3/lib/six.py +++ b/shotgun_api3/lib/six.py @@ -36,6 +36,7 @@ PY2 = sys.version_info[0] == 2 PY3 = sys.version_info[0] == 3 PY34 = sys.version_info[0:2] >= (3, 4) +PY38 = sys.version_info[0:2] >= (3, 8) if PY3: string_types = str, diff --git a/shotgun_api3/shotgun.py b/shotgun_api3/shotgun.py index acaeb06f..7e48573d 100644 --- a/shotgun_api3/shotgun.py +++ b/shotgun_api3/shotgun.py @@ -35,7 +35,6 @@ from .lib.six import BytesIO # used for attachment upload from .lib.six.moves import map -import base64 from .lib.six.moves import http_cookiejar # used for attachment upload import datetime import logging @@ -57,6 +56,12 @@ # to be exposed as part of the API. from .lib.six.moves.xmlrpc_client import Error, ProtocolError, ResponseError # noqa +if six.PY3: + from base64 import encodebytes as base64encode +else: + from base64 import encodestring as base64encode + + LOG = logging.getLogger("shotgun_api3") """ Logging instance for shotgun_api3 @@ -651,18 +656,20 @@ def __init__(self, # if the service contains user information strip it out # copied from the xmlrpclib which turned the user:password into # and auth header - # Do NOT urlsplit(self.base_url) here, as it contains the lower case version - # of the base_url argument. Doing so would base64-encode the lowercase - # version of the credentials. - auth, self.config.server = urllib.parse.splituser(urllib.parse.urlsplit(base_url).netloc) + + # Do NOT self._split_url(self.base_url) here, as it contains the lower + # case version of the base_url argument. Doing so would base64encode + # the lowercase version of the credentials. + auth, self.config.server = self._split_url(base_url) if auth: - auth = base64.encodestring(six.ensure_binary(urllib.parse.unquote(auth))).decode("utf-8") + auth = base64encode(six.ensure_binary( + urllib.parse.unquote(auth))).decode("utf-8") self.config.authorization = "Basic " + auth.strip() # foo:bar@123.456.789.012:3456 if http_proxy: - # check if we're using authentication. Start from the end since there might be - # @ in the user's password. + # check if we're using authentication. Start from the end since + # there might be @ in the user's password. p = http_proxy.rsplit("@", 1) if len(p) > 1: self.config.proxy_user, self.config.proxy_pass = \ @@ -710,6 +717,33 @@ def __init__(self, self.config.user_password = None self.config.auth_token = None + def _split_url(self, base_url): + """ + Extract the hostname:port and username/password/token from base_url + sent when connect to the API. + + In python 3.8 `urllib.parse.splituser` was deprecated warning devs to + use `urllib.parse.urlparse`. + """ + if six.PY38: + auth = None + results = urllib.parse.urlparse(base_url) + server = results.hostname + if results.port: + server = "{}:{}".format(server, results.port) + + if results.username: + auth = results.username + + if results.password: + auth = "{}:{}".format(auth, results.password) + + else: + auth, server = urllib.parse.splituser( + urllib.parse.urlsplit(base_url).netloc) + + return auth, server + # ======================================================================== # API Functions diff --git a/tests/base.py b/tests/base.py index f1179c06..8d0e3a5b 100644 --- a/tests/base.py +++ b/tests/base.py @@ -2,8 +2,6 @@ import os import re import unittest -from shotgun_api3.lib.six.moves.configparser import ConfigParser - from . import mock @@ -12,6 +10,12 @@ from shotgun_api3.shotgun import ServerCapabilities from shotgun_api3.lib import six +if six.PY2: + from shotgun_api3.lib.six.moves.configparser import SafeConfigParser as ConfigParser +else: + from shotgun_api3.lib.six.moves.configparser import ConfigParser + + try: # Attempt to import skip from unittest. Since this was added in Python 2.7 # in the case that we're running on Python 2.6 we'll need a decorator to diff --git a/tests/test_api.py b/tests/test_api.py index 50632216..0e341838 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1661,7 +1661,7 @@ def test_zero_is_not_none(self): # Should be filtered out result = self.sg.find('Asset', [['id', 'is', self.asset['id']], [num_field, 'is_not', None]], [num_field]) - self.assertEquals([], result) + self.assertEqual([], result) # Set it to zero self.sg.update('Asset', self.asset['id'], {num_field: 0}) @@ -1681,17 +1681,17 @@ def test_include_archived_projects(self): if self.sg.server_caps.version > (5, 3, 13): # Ticket #25082 result = self.sg.find_one('Shot', [['id', 'is', self.shot['id']]]) - self.assertEquals(self.shot['id'], result['id']) + self.assertEqual(self.shot['id'], result['id']) # archive project self.sg.update('Project', self.project['id'], {'archived': True}) # setting defaults to True, so we should get result result = self.sg.find_one('Shot', [['id', 'is', self.shot['id']]]) - self.assertEquals(self.shot['id'], result['id']) + self.assertEqual(self.shot['id'], result['id']) result = self.sg.find_one('Shot', [['id', 'is', self.shot['id']]], include_archived_projects=False) - self.assertEquals(None, result) + self.assertEqual(None, result) # unarchive project self.sg.update('Project', self.project['id'], {'archived': False}) @@ -2810,7 +2810,7 @@ def test_import_httplib(self): # right one.) httplib2_compat_version = httplib2.Http.__module__.split(".")[-1] if six.PY2: - self.assertEquals(httplib2_compat_version, "python2") + self.assertEqual(httplib2_compat_version, "python2") elif six.PY3: self.assertTrue(httplib2_compat_version, "python3") diff --git a/tests/test_client.py b/tests/test_client.py index 00130c5d..db4b93bd 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -12,11 +12,11 @@ CRUD functions. These tests always use a mock http connection so not not need a live server to run against.""" -import base64 import datetime -from shotgun_api3.lib.six.moves import urllib import os import re + +from shotgun_api3.lib.six.moves import urllib from shotgun_api3.lib import six try: import simplejson as json @@ -38,8 +38,14 @@ from . import base +if six.PY3: + from base64 import encodebytes as base64encode +else: + from base64 import encodestring as base64encode + + def b64encode(val): - return base64.encodestring(six.ensure_binary(val)).decode("utf-8") + return base64encode(six.ensure_binary(val)).decode("utf-8") class TestShotgunClient(base.MockTestBase): @@ -164,6 +170,61 @@ def test_url(self): expected = "Basic " + b64encode(urllib.parse.unquote(login_password)).strip() self.assertEqual(expected, sg.config.authorization) + def test_b64encode(self): + """Parse value using the proper encoder.""" + login = "thelogin" + password = "%thepassw0r#$" + login_password = "%s:%s" % (login, password) + expected = 'dGhlbG9naW46JXRoZXBhc3N3MHIjJA==' + result = b64encode(urllib.parse.unquote(login_password)).strip() + self.assertEqual(expected, result) + + def test_read_config(self): + """Validate that config values are properly coerced.""" + this_dir = os.path.dirname(os.path.realpath(__file__)) + config_path = os.path.join(this_dir, "test_config_file") + config = base.ConfigParser() + config.read(config_path) + result = config.get("SERVER_INFO", "api_key") + expected = "%abce" + + self.assertEqual(expected, result) + + def test_split_url(self): + """Validate that url parts are properly extracted.""" + + sg = api.Shotgun("https://ci.shotgunstudio.com", + "foo", "bar", connect=False) + + + base_url = "https://ci.shotgunstudio.com" + expected_server = "ci.shotgunstudio.com" + expected_auth = None + auth, server = sg._split_url(base_url) + self.assertEqual(auth, expected_auth) + self.assertEqual(server, expected_server) + + base_url = "https://ci.shotgunstudio.com:9500" + expected_server = "ci.shotgunstudio.com:9500" + expected_auth = None + auth, server = sg._split_url(base_url) + self.assertEqual(auth, expected_auth) + self.assertEqual(server, expected_server) + + base_url = "https://x:y@ci.shotgunstudio.com:9500" + expected_server = "ci.shotgunstudio.com:9500" + expected_auth = "x:y" + auth, server = sg._split_url(base_url) + self.assertEqual(auth, expected_auth) + self.assertEqual(server, expected_server) + + base_url = "https://12345XYZ@ci.shotgunstudio.com:9500" + expected_server = "ci.shotgunstudio.com:9500" + expected_auth = "12345XYZ" + auth, server = sg._split_url(base_url) + self.assertEqual(auth, expected_auth) + self.assertEqual(server, expected_server) + def test_authorization(self): """Authorization passed to server""" login = self.human_user['login'] diff --git a/tests/test_config_file b/tests/test_config_file new file mode 100644 index 00000000..8215eecd --- /dev/null +++ b/tests/test_config_file @@ -0,0 +1,7 @@ +[SERVER_INFO] +server_url : https://url +script_name : xyz +api_key : %%abce + +[TEST_DATA] +project_name : hjkl \ No newline at end of file diff --git a/tests/test_unit.py b/tests/test_unit.py index 23ff67e0..1755b51c 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -35,8 +35,8 @@ def test_http_proxy_server(self): self.api_key, http_proxy=http_proxy, connect=False) - self.assertEquals(sg.config.proxy_server, proxy_server) - self.assertEquals(sg.config.proxy_port, 8080) + self.assertEqual(sg.config.proxy_server, proxy_server) + self.assertEqual(sg.config.proxy_port, 8080) proxy_server = "123.456.789.012" http_proxy = proxy_server sg = api.Shotgun(self.server_path, @@ -44,8 +44,8 @@ def test_http_proxy_server(self): self.api_key, http_proxy=http_proxy, connect=False) - self.assertEquals(sg.config.proxy_server, proxy_server) - self.assertEquals(sg.config.proxy_port, 8080) + self.assertEqual(sg.config.proxy_server, proxy_server) + self.assertEqual(sg.config.proxy_port, 8080) def test_http_proxy_server_and_port(self): proxy_server = "someserver.com" @@ -56,8 +56,8 @@ def test_http_proxy_server_and_port(self): self.api_key, http_proxy=http_proxy, connect=False) - self.assertEquals(sg.config.proxy_server, proxy_server) - self.assertEquals(sg.config.proxy_port, proxy_port) + self.assertEqual(sg.config.proxy_server, proxy_server) + self.assertEqual(sg.config.proxy_port, proxy_port) proxy_server = "123.456.789.012" proxy_port = 1234 http_proxy = "%s:%d" % (proxy_server, proxy_port) @@ -66,8 +66,8 @@ def test_http_proxy_server_and_port(self): self.api_key, http_proxy=http_proxy, connect=False) - self.assertEquals(sg.config.proxy_server, proxy_server) - self.assertEquals(sg.config.proxy_port, proxy_port) + self.assertEqual(sg.config.proxy_server, proxy_server) + self.assertEqual(sg.config.proxy_port, proxy_port) def test_http_proxy_server_and_port_with_authentication(self): proxy_server = "someserver.com" @@ -81,10 +81,10 @@ def test_http_proxy_server_and_port_with_authentication(self): self.api_key, http_proxy=http_proxy, connect=False) - self.assertEquals(sg.config.proxy_server, proxy_server) - self.assertEquals(sg.config.proxy_port, proxy_port) - self.assertEquals(sg.config.proxy_user, proxy_user) - self.assertEquals(sg.config.proxy_pass, proxy_pass) + self.assertEqual(sg.config.proxy_server, proxy_server) + self.assertEqual(sg.config.proxy_port, proxy_port) + self.assertEqual(sg.config.proxy_user, proxy_user) + self.assertEqual(sg.config.proxy_pass, proxy_pass) proxy_server = "123.456.789.012" proxy_port = 1234 proxy_user = "user" @@ -96,10 +96,10 @@ def test_http_proxy_server_and_port_with_authentication(self): self.api_key, http_proxy=http_proxy, connect=False) - self.assertEquals(sg.config.proxy_server, proxy_server) - self.assertEquals(sg.config.proxy_port, proxy_port) - self.assertEquals(sg.config.proxy_user, proxy_user) - self.assertEquals(sg.config.proxy_pass, proxy_pass) + self.assertEqual(sg.config.proxy_server, proxy_server) + self.assertEqual(sg.config.proxy_port, proxy_port) + self.assertEqual(sg.config.proxy_user, proxy_user) + self.assertEqual(sg.config.proxy_pass, proxy_pass) def test_http_proxy_with_at_in_password(self): proxy_server = "someserver.com" @@ -113,10 +113,10 @@ def test_http_proxy_with_at_in_password(self): self.api_key, http_proxy=http_proxy, connect=False) - self.assertEquals(sg.config.proxy_server, proxy_server) - self.assertEquals(sg.config.proxy_port, proxy_port) - self.assertEquals(sg.config.proxy_user, proxy_user) - self.assertEquals(sg.config.proxy_pass, proxy_pass) + self.assertEqual(sg.config.proxy_server, proxy_server) + self.assertEqual(sg.config.proxy_port, proxy_port) + self.assertEqual(sg.config.proxy_user, proxy_user) + self.assertEqual(sg.config.proxy_pass, proxy_pass) def test_malformatted_proxy_info(self): conn_info = { @@ -172,7 +172,7 @@ def test_filters(self): args = ['', [[path, relation, value]], None] result = self.get_call_rpc_params(args, {}) actual_condition = result['filters']['conditions'][0] - self.assertEquals(expected_condition, actual_condition) + self.assertEqual(expected_condition, actual_condition) @patch('shotgun_api3.Shotgun._call_rpc') def get_call_rpc_params(self, args, kws, call_rpc): @@ -262,8 +262,8 @@ def assert_platform(self, sys_ret_val, expected): expected_local_path_field = "local_path_%s" % expected client_caps = api.shotgun.ClientCapabilities() - self.assertEquals(client_caps.platform, expected) - self.assertEquals(client_caps.local_path_field, expected_local_path_field) + self.assertEqual(client_caps.platform, expected) + self.assertEqual(client_caps.local_path_field, expected_local_path_field) finally: api.shotgun.sys.platform = platform @@ -272,8 +272,8 @@ def test_no_platform(self): try: api.shotgun.sys.platform = "unsupported" client_caps = api.shotgun.ClientCapabilities() - self.assertEquals(client_caps.platform, None) - self.assertEquals(client_caps.local_path_field, None) + self.assertEqual(client_caps.platform, None) + self.assertEqual(client_caps.local_path_field, None) finally: api.shotgun.sys.platform = platform @@ -285,7 +285,7 @@ def test_py_version(self, mock_sys): mock_sys.version_info = (major, minor, micro, 'final', 0) expected_py_version = "%s.%s" % (major, minor) client_caps = api.shotgun.ClientCapabilities() - self.assertEquals(client_caps.py_version, expected_py_version) + self.assertEqual(client_caps.py_version, expected_py_version) class TestFilters(unittest.TestCase): @@ -296,7 +296,7 @@ def test_empty(self): } result = api.shotgun._translate_filters([], None) - self.assertEquals(result, expected) + self.assertEqual(result, expected) def test_simple(self): filters = [ @@ -313,7 +313,7 @@ def test_simple(self): } result = api.shotgun._translate_filters(filters, "any") - self.assertEquals(result, expected) + self.assertEqual(result, expected) # Test both styles of passing arrays def test_arrays(self): @@ -329,14 +329,14 @@ def test_arrays(self): ] result = api.shotgun._translate_filters(filters, "all") - self.assertEquals(result, expected) + self.assertEqual(result, expected) filters = [ ["code", "in", ["test1", "test2", "test3"]] ] result = api.shotgun._translate_filters(filters, "all") - self.assertEquals(result, expected) + self.assertEqual(result, expected) def test_nested(self): filters = [ @@ -379,7 +379,7 @@ def test_nested(self): } result = api.shotgun._translate_filters(filters, "all") - self.assertEquals(result, expected) + self.assertEqual(result, expected) def test_invalid(self): self.assertRaises(api.ShotgunError, api.shotgun._translate_filters, [], "bogus") @@ -461,7 +461,7 @@ def test_found_correct_cert(self): os.path.join(os.path.dirname(api.__file__), "lib", "certifi", "cacert.pem") ) # Now ensure that the path the SG API has found is correct. - self.assertEquals(cert_path, self.certs) + self.assertEqual(cert_path, self.certs) self.assertTrue(os.path.isfile(self.certs)) def test_httplib(self): @@ -475,7 +475,7 @@ def test_httplib(self): # Now check that the good urls connect properly using the certs for url in self.test_urls: response, message = self._check_url_with_sg_api_httplib2(url, self.certs) - self.assertEquals(response["status"], "200") + self.assertEqual(response["status"], "200") def test_urlib(self): """