self signed certificate - ssl disabling support w/ testing

.gitignore

@@ -1,6 +1,7 @@
 #Eclipse
 .project
-
+venv
+venv_win
 TestFiles
 test.txt
 

Algorithmia/client.py

@@ -41,6 +41,7 @@ def __init__(self, apiKey=None, apiAddress=None, caCert=None, bearerToken=None):
             self.apiAddress = Algorithmia.getApiAddress()
         if caCert == False:
             self.requestSession.verify = False
+            self.requestSession.trust_env = False
             config = Configuration(use_ssl=False)
         elif caCert is None and 'REQUESTS_CA_BUNDLE' in os.environ:
             caCert = os.environ.get('REQUESTS_CA_BUNDLE')

Test/conftest.py

@@ -1,12 +1,15 @@
 import sys
 from time import sleep
+import os, signal
 if sys.version_info.major >= 3:
-    from Test.api import start_webserver
+    from Test.api import start_webserver_reg, start_webserver_self_signed
     import pytest
 
     @pytest.fixture(scope='package', autouse=True)
     def fastapi_start():
-        p = start_webserver()
+        p_reg = start_webserver_reg()
+        p_self_signed = start_webserver_self_signed()
         sleep(2)
-        yield p
-        p.terminate()
+        yield p_reg, p_self_signed
+        os.kill(p_reg.pid, signal.SIGKILL)
+        os.kill(p_self_signed.pid, signal.SIGKILL)

Test/CLI_test.py → Test/regular/CLI_test.py

@@ -93,8 +93,8 @@ def setUp(self):
         # create a directory to use in testing the cp command
         self.client = Algorithmia.client()
         CLI().mkdir("data://.my/moredata", self.client)
-        if not os.path.exists("./TestFiles/"):
-            os.mkdir("./TestFiles/")
+        if not os.path.exists("../TestFiles/"):
+            os.mkdir("../TestFiles/")
 
     def test_ls(self):
         parentDir = "data://.my/"
@@ -132,7 +132,7 @@ def test_rmdir(self):
 
     def test_cat(self):
         file = "data://.my/moredata/test.txt"
-        localfile = "./TestFiles/test.txt"
+        localfile = "./../TestFiles/test.txt"
         fileContents = "some text in test file"
 
         CLI().rm(file, self.client)
@@ -156,7 +156,7 @@ def test_get_build_logs(self):
 
     # local to remote
     def test_cp_L2R(self):
-        localfile = "./TestFiles/test.txt"
+        localfile = "./../TestFiles/test.txt"
         testfile = open(localfile, "w")
         testfile.write("some text")
         testfile.close()
@@ -199,7 +199,7 @@ def test_auth(self):
 
     def test_auth_cert(self):
 
-        localfile = "./TestFiles/fakecert.pem"
+        localfile = "./../TestFiles/fakecert.pem"
 
         testfile = open(localfile, "w")
         testfile.write("")
@@ -244,7 +244,7 @@ def test_list_languages(self):
         self.assertTrue(result is not None and "anaconda3" in result[1])
 
     def test_rm(self):
-        localfile = "./TestFiles/testRM.txt"
+        localfile = "./../TestFiles/testRM.txt"
 
         testfile = open(localfile, "w")
         testfile.write("some text")
@@ -263,7 +263,7 @@ def test_rm(self):
         self.assertTrue("testRM.txt" in result1 and "testRM.txt" not in result2)
 
     def test_get_template(self):
-        filename = "./temptest"
+        filename = "./../temptest"
         envid = "36fd467e-fbfe-4ea6-aa66-df3f403b7132"
         response = CLI().get_template(envid, filename, self.client)
         print(response)

Test/algo_failure_test.py → Test/regular/algo_failure_test.py

@@ -11,8 +11,6 @@
     # you will load the version installed on the computer.
     sys.path = ['../'] + sys.path
     from requests import Response
-    from Test.api import app
-
 
     class AlgoTest(unittest.TestCase):
         error_500 = Response()

Test/client_test.py → Test/regular/client_test.py

@@ -20,7 +20,6 @@ class ClientDummyTest(unittest.TestCase):
         @classmethod
         def setUpClass(cls):
             cls.client = Algorithmia.client(api_address="http://localhost:8080", api_key="simabcd123")
-
         admin_username = "a_Mrtest"
         admin_org_name = "a_myOrg"
         environment_name = "Python 3.9"
@@ -274,16 +273,6 @@ def test_get_build_logs(self):
 
             self.assertTrue(u'error' not in result)
 
-        def test_get_build_logs_no_ssl(self):
-            client = Algorithmia.client(api_address='https://api.algorithmia.com',
-                                        api_key=self.regular_api_key, ca_cert=False)
-            user = unicode(os.environ.get('ALGO_USER_NAME'))
-            algo = u'Echo'
-            result = client.algo(user + '/' + algo).build_logs()
-            if u'error' in result:
-                print(result)
-            self.assertTrue("error" not in result)
-
         def test_edit_org(self):
             org_name = "a_myOrg84"
 

Test/resources/cert.cert

@@ -0,0 +1,21 @@
+-----BEGIN CERTIFICATE-----
+MIIDazCCAlOgAwIBAgIUTikiwxFBpLW4pC+5VfOis1xCYKcwDQYJKoZIhvcNAQEL
+BQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDAeFw0yMjA1MDMxNzE2MjZaFw0yMjA2
+MDIxNzE2MjZaMEUxCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEw
+HwYDVQQKDBhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQwggEiMA0GCSqGSIb3DQEB
+AQUAA4IBDwAwggEKAoIBAQDBUMZkg/bCJurQIB9znskTjv8URtIK6qqvZpYGTbfI
+AzY6HiI0o1gPxjINZW7cBky/9MeEV5zyJghC4WoK099cIUNq2TmAWAjlRgIE8iEy
+9z7QVfbSMainuw0RTlD5/8FRWtRe5v8qwbqLICMn3qv/KsG6bRezyS7UVihwFJua
+E4dki+y6KSha4RrCtC43inbPlncB4om7PfJQyt5nI7N4KxbY2L3BUa5/+x1ux/ni
+C/3y808vLJVQ6nLYgTEg/6K6lFrig0mUIMnCuOiBsrms3NmBPuDdRri/z1ulFHJB
+WVQVQ5DgWher0f/dMzHwyRj3ffC8bAPlhrvLHwPQtNeRAgMBAAGjUzBRMB0GA1Ud
+DgQWBBRoC77Hql6kEzk7WC6BeaPBu82K/jAfBgNVHSMEGDAWgBRoC77Hql6kEzk7
+WC6BeaPBu82K/jAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3DQEBCwUAA4IBAQCn
+W9acM3+rxsiBBClTYEm2tOiukcXEkI7IzvW/4r7P24SmUiDD3vxVVbZ6nevVkg+P
+4/QH+YYE3JUeXaN+xnHYjSy4NKxjd3EHT7BFxLMe0DQaodMj0klHqBtULNzojv8+
+/5tpQsjDLeeeDyOIJNz8r6CU9Gzh7j1EBF8BRdLA1z2UVmt6l6d4o3xOTYpOlZs3
+gI+ASxF9ODQzCCOeMYO2qiuMV3RD0oNdIEHUiMD+iHeC1jFGlxZzaWNeuUzP7Yj/
+MOwbBo8l6Hk2BUuUayLxZFLd0wN28IRkLEU5/SOh3mKz79nfPk6pD9rHUO1a53lI
+Ua5xJ5tSwG6bMtNnHYYX
+-----END CERTIFICATE-----

Test/resources/cert.key

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQDBUMZkg/bCJurQ
+IB9znskTjv8URtIK6qqvZpYGTbfIAzY6HiI0o1gPxjINZW7cBky/9MeEV5zyJghC
+4WoK099cIUNq2TmAWAjlRgIE8iEy9z7QVfbSMainuw0RTlD5/8FRWtRe5v8qwbqL
+ICMn3qv/KsG6bRezyS7UVihwFJuaE4dki+y6KSha4RrCtC43inbPlncB4om7PfJQ
+yt5nI7N4KxbY2L3BUa5/+x1ux/niC/3y808vLJVQ6nLYgTEg/6K6lFrig0mUIMnC
+uOiBsrms3NmBPuDdRri/z1ulFHJBWVQVQ5DgWher0f/dMzHwyRj3ffC8bAPlhrvL
+HwPQtNeRAgMBAAECggEAPr2OhhTmQ0EKOJ4UVxwTuotQci5CAVUELIUo78bNfNa+
+BMK+60KQVB5JJFvlTPemdS5miqc8wsJhMAOkvPripS4OiWES7nqj+HVuNli3OalQ
+86DSyIlhaX6l0RYP5fOBtHu8LUjfS+swNfMqNchpHhmsYmsBpFIJJtUHrsihb7GR
+4LpNOZ5go4+LG7FX9KaUE4FvAlS7hi6KLSMua10+3+NAlXggbcVikHr3Uq6eQIvk
+z09cs+q2FHaESdTjXSIitmYOfJU5KK3QSfXAr/vaqakjnMvfp8MzQ5dHFsy03HRZ
+Sy+LjRKOEOCMCT4DmGIPO4V89i3prbVH4JxixCOaeQKBgQDzuwERWE04JEtvfjxS
+OAciQKLIxhfa4t2VB65d3115wxfDPIBYU5Mx5YV4aQyOddNxBwpmX/wYwstx2JDZ
+2JM0OjOKLnSvlQfr5UmsY9jUO7CdmgC5HpgbHNhc8uJFw4pd+XypWSjytmVxBSdb
+m0+in/iUUQuFNH/+BNLVVgWSiwKBgQDLDBCTEpKQvx2kAc8TEtwrWNhacZILab5D
+StQBEL62VfGMdXYaA5dXreo5nqioHfBR3BfAmDecmq3iEFE8/yHJs6pLdcmj0Z1L
+034UQedYLCmL9zuAgC6p4SKIMPubnYtMrNJOL3lq0ibogz3rfOhdN2B6S88IYoSL
+M6asdoQN0wKBgCd1VPzr4MSAC75nH3joHS+Ma045087Z/6mK7s2/xbBax1QSTWz/
+Sss/L1aJG0FNDgg0bZiZXYTctHcf6oN6Loq8CXALiVSLuhaUrlK8b3QcncFGF2vg
+6hspllWl9L/6okIIjAgWqSxyHwYnIXIRONlJMMNCQ60zDK2hNkjXflt1AoGAX0w3
+Tz/NSGBaogozTUlxymp1iOV63R5xLRYmsKVSTTPDHeBXYNhEpOM8ZnS/xb/fdhwt
+jbgjib3TVKHB7zXzfr5zc91BmUCdaeRGbW2NDgYULdwIskP3IsZGtdL/lEb6BS+r
+uQRxISCnIEPQwQCr8mw2PM/tyIqsmMTSOmmZiv8CgYBAfIC/cNwJyID6AVauZHQo
+S3Bii9CPmPnuklBuS7ikX0bmZ93dzv537nqwGr0j9ksxabLWZRcoxx6MhgmXzXVT
+dy48TWpqpHiMNorYskB9tcZSrBCl70bu5qKp2owqWHW0d4hqH3lkBNFhfwNWm+qC
+54x3T/1fqyaqeapCiE5FGA==
+-----END PRIVATE KEY-----

Test/self_signed/acl_test.py

@@ -0,0 +1,38 @@
+import sys
+# look in ../ BEFORE trying to import Algorithmia.  If you append to the
+# you will load the version installed on the computer.
+sys.path = ['../'] + sys.path
+
+import unittest
+import Algorithmia
+from Algorithmia.acl import AclType, Acl, ReadAcl
+from Algorithmia.datadirectory import DataDirectory
+
+class AclTypeTest(unittest.TestCase):
+    def test_types(self):
+        self.assertTrue(AclType.private.acl_string is None)
+        self.assertEquals(AclType.my_algos.acl_string, 'algo://.my/*')
+        self.assertEquals(AclType.public.acl_string, 'user://*')
+        self.assertEquals(AclType.default, AclType.my_algos)
+
+    def test_from_acl_response(self):
+        self.assertEquals(AclType.from_acl_response([]), AclType.private)
+        self.assertEquals(AclType.from_acl_response(['algo://.my/*']), AclType.my_algos)
+        self.assertEquals(AclType.from_acl_response(['user://*']), AclType.public)
+
+    def test_create_acl(self):
+        c = Algorithmia.client()
+        dd = DataDirectory(c, 'data://.my/privatePermissions')
+        if dd.exists():
+            dd.delete(True)
+        dd.create(ReadAcl.private)
+
+        dd_perms = DataDirectory(c, 'data://.my/privatePermissions').get_permissions()
+        self.assertEquals(dd_perms.read_acl, AclType.private)
+
+        dd.update_permissions(ReadAcl.public)
+        dd_perms = DataDirectory(c, 'data://.my/privatePermissions').get_permissions()
+        self.assertEquals(dd_perms.read_acl, AclType.public)
+
+if __name__ == '__main__':
+    unittest.main()

Test/self_signed/algo_failure_test.py

@@ -0,0 +1,30 @@
+import sys
+
+if sys.version_info[0] >= 3:
+    import unittest
+    import Algorithmia
+    import uvicorn
+    import time
+    from multiprocessing import Process
+
+    # look in ../ BEFORE trying to import Algorithmia.  If you append to the
+    # you will load the version installed on the computer.
+    sys.path = ['../'] + sys.path
+    from requests import Response
+
+    class AlgoTest(unittest.TestCase):
+        error_500 = Response()
+        error_500.status_code = 500
+        error_message = "Non-Algorithm related Failure: " + str(error_500)
+
+        @classmethod
+        def setUpClass(cls):
+            cls.client = Algorithmia.client(api_address="https://localhost:8090", api_key="simabcd123", ca_cert=False)
+
+        def test_throw_500_error_HTTP_response_on_algo_request(self):
+            try:
+                result = self.client.algo('util/500').pipe(bytearray('foo', 'utf-8'))
+            except Exception as e:
+                result = e
+                pass
+            self.assertEqual(str(self.error_message), str(result))

Test/self_signed/algo_test.py

@@ -0,0 +1,136 @@
+import sys
+import os
+from Algorithmia.errors import AlgorithmException
+from Algorithmia.algorithm import OutputType
+import Algorithmia
+# look in ../ BEFORE trying to import Algorithmia.  If you append to the
+# you will load the version installed on the computer.
+sys.path = ['../'] + sys.path
+
+import unittest
+
+if sys.version_info.major >= 3:
+
+
+    class AlgoDummyTest(unittest.TestCase):
+        @classmethod
+        def setUpClass(cls):
+            cls.client = Algorithmia.client(api_address="https://localhost:8090", api_key="simabcd123", ca_cert=False)
+
+        def test_call_customCert(self):
+            result = self.client.algo('util/echo').pipe(bytearray('foo', 'utf-8'))
+            self.assertEquals('binary', result.metadata.content_type)
+            self.assertEquals(bytearray('foo', 'utf-8'), result.result)
+
+        def test_normal_call(self):
+            result = self.client.algo('util/echo').pipe("foo")
+            self.assertEquals("text", result.metadata.content_type)
+            self.assertEquals("foo", result.result)
+
+        def test_async_call(self):
+            result = self.client.algo('util/echo').set_options(output=OutputType.void).pipe("foo")
+            self.assertTrue(hasattr(result, "async_protocol"))
+            self.assertTrue(hasattr(result, "request_id"))
+
+        def test_raw_call(self):
+            result = self.client.algo('util/echo').set_options(output=OutputType.raw).pipe("foo")
+            self.assertEquals("foo", result)
+
+        def test_dict_call(self):
+            result = self.client.algo('util/echo').pipe({"foo": "bar"})
+            self.assertEquals("json", result.metadata.content_type)
+            self.assertEquals({"foo": "bar"}, result.result)
+
+        def test_text_unicode(self):
+            telephone = u"\u260E"
+            # Unicode input to pipe()
+            result1 = self.client.algo('util/Echo').pipe(telephone)
+            self.assertEquals('text', result1.metadata.content_type)
+            self.assertEquals(telephone, result1.result)
+
+            # Unicode return in .result
+            result2 = self.client.algo('util/Echo').pipe(result1.result)
+            self.assertEquals('text', result2.metadata.content_type)
+            self.assertEquals(telephone, result2.result)
+
+        def test_get_build_by_id(self):
+            result = self.client.algo("J_bragg/Echo").get_build("1a392e2c-b09f-4bae-a616-56c0830ac8e5")
+            self.assertTrue(result.build_id is not None)
+
+        def test_get_build_logs(self):
+            result = self.client.algo("J_bragg/Echo").get_build_logs("1a392e2c-b09f-4bae-a616-56c0830ac8e5")
+            self.assertTrue(result.logs is not None)
+
+        def test_get_scm_status(self):
+            result = self.client.algo("J_bragg/Echo").get_scm_status()
+            self.assertTrue(result.scm_connection_status is not None)
+
+        def test_exception_ipa_algo(self):
+            try:
+                result = self.client.algo('zeryx/raise_exception').pipe("")
+            except AlgorithmException as e:
+                self.assertEqual(e.message, "This is an exception")
+
+else:
+    class AlgoTest(unittest.TestCase):
+        def setUp(self):
+            self.client = Algorithmia.client()
+
+        def test_call_customCert(self):
+            open("./test.pem", 'w')
+            c = Algorithmia.client(ca_cert="./test.pem")
+            result = c.algo('util/Echo').pipe(bytearray('foo', 'utf-8'))
+            self.assertEquals('binary', result.metadata.content_type)
+            self.assertEquals(bytearray('foo', 'utf-8'), result.result)
+            try:
+                os.remove("./test.pem")
+            except OSError as e:
+                print(e)
+
+        def test_call_binary(self):
+            result = self.client.algo('util/Echo').pipe(bytearray('foo', 'utf-8'))
+            self.assertEquals('binary', result.metadata.content_type)
+            self.assertEquals(bytearray('foo', 'utf-8'), result.result)
+
+        def test_async_call(self):
+            result = self.client.algo('util/echo').set_options(output=OutputType.void).pipe("foo")
+            self.assertTrue(hasattr(result, "async_protocol"))
+            self.assertTrue(hasattr(result, "request_id"))
+
+        def test_raw_call(self):
+            result = self.client.algo('util/echo').set_options(output=OutputType.raw).pipe("foo")
+            self.assertEquals("foo", result)
+
+        def test_text_unicode(self):
+            telephone = u"\u260E"
+
+            # Unicode input to pipe()
+            result1 = self.client.algo('util/Echo').pipe(telephone)
+            self.assertEquals('text', result1.metadata.content_type)
+            self.assertEquals(telephone, result1.result)
+
+            # Unicode return in .result
+            result2 = self.client.algo('util/Echo').pipe(result1.result)
+            self.assertEquals('text', result2.metadata.content_type)
+            self.assertEquals(telephone, result2.result)
+
+        def test_get_build_by_id(self):
+            result = self.client.algo("J_bragg/Echo").get_build("1a392e2c-b09f-4bae-a616-56c0830ac8e5")
+            self.assertTrue(result.build_id is not None)
+
+        def test_get_build_logs(self):
+            result = self.client.algo("J_bragg/Echo").get_build_logs("1a392e2c-b09f-4bae-a616-56c0830ac8e5")
+            self.assertTrue(result.logs is not None)
+
+        def test_get_scm_status(self):
+            result = self.client.algo("J_bragg/Echo").get_scm_status()
+            self.assertTrue(result.scm_connection_status is not None)
+
+        def test_exception_ipa_algo(self):
+            try:
+                result = self.client.algo('zeryx/raise_exception').pipe("")
+            except AlgorithmException as e:
+                self.assertEqual(e.message, "This is an exception")
+
+if __name__ == '__main__':
+    unittest.main()