objectMask support

Author: allmightyspiff

SoftLayer/transports/soap.py

@@ -11,6 +11,7 @@
 from zeep import Client, Settings, Transport, xsd
 from zeep.helpers import serialize_object
 from zeep.cache import SqliteCache
+from zeep.plugins import HistoryPlugin
 
 import requests
 
@@ -31,43 +32,51 @@ def __init__(self, endpoint_url=None, timeout=None, proxy=None, user_agent=None,
 
         # Throw an error for py < 3.6 because of f-strings
         logging.getLogger('zeep').setLevel(logging.ERROR)
+        logging.getLogger('zeep.transports').setLevel(logging.DEBUG)
         self.endpoint_url = (endpoint_url or consts.API_PUBLIC_ENDPOINT).rstrip('/')
         self.timeout = timeout or None
         self.proxy = proxy
         self.user_agent = user_agent or consts.USER_AGENT
         self.verify = verify
         self._client = None
+        self.history = HistoryPlugin()
 
     def __call__(self, request):
         """Makes a SoftLayer API call against the SOAP endpoint.
 
         :param request request: Request object
         """
-        print("Making a SOAP API CALL...")
-        # client = Client(f"{self.endpoint_url}/{request.service}?wsdl")
+
         zeep_settings = Settings(strict=False, xml_huge_tree=True)
         zeep_transport = Transport(cache=SqliteCache(timeout=86400))
         client = Client(f"{self.endpoint_url}/{request.service}?wsdl",
-                        settings=zeep_settings, transport=zeep_transport)
-        # authXsd = xsd.Element(
-        #     f"{self.endpoint_url}/authenticate",
-        #     xsd.ComplexType([
-        #         xsd.Element(f"{self.endpoint_url}/username", xsd.String()),
-        #         xsd.Element(f"{self.endpoint_url}/apiKey", xsd.String())
-        #     ])
-        # )
+                        settings=zeep_settings, transport=zeep_transport, plugins=[self.history])
+
+        # MUST define headers like this because otherwise the objectMask header doesn't work
+        # because it isn't sent in with a namespace.
         xsdUserAuth = xsd.Element(
-            '{http://api.softlayer.com/soap/v3.1/}authenticate',
+            '{http://api.softlayer.com/soap/v3/}authenticate',
             xsd.ComplexType([
-                xsd.Element('{http://api.softlayer.com/soap/v3.1/}username', xsd.String()),
-                xsd.Element('{http://api.softlayer.com/soap/v3.1/}apiKey', xsd.String())
+                xsd.Element('{http://api.service.softlayer.com/soap/v3/}username', xsd.String()),
+                xsd.Element('{http://api.service.softlayer.com/soap/v3/}apiKey', xsd.String())
             ])
         )
-        # transport = Transport(session=get_session())
-        
-        authHeader = xsdUserAuth(username=request.transport_user, apiKey=request.transport_password)
+        xsdMask = xsd.Element(
+            '{http://api.service.softlayer.com/soap/v3.1/}SoftLayer_ObjectMask',
+            xsd.ComplexType([
+                xsd.Element('mask', xsd.String()),
+            ])
+        )
+
+        headers = [
+            xsdMask(mask=request.mask or ''),
+            xsdUserAuth(username=request.transport_user, apiKey=request.transport_password)
+        ]
+
+        pp(headers)
+        print("HEADERS ^^^^^")
         method = getattr(client.service, request.method)
-        result = client.service.getObject(_soapheaders=[authHeader])
+        result = client.service.getObject(_soapheaders=headers)
         return serialize_object(result)
         # result = transport.post(f"{self.endpoint_url}/{request.service}")
 
@@ -80,4 +89,4 @@ def print_reproduceable(self, request):
         :param request request: Request object
         """
 
-        return "THE SOAP API CALL..."
+        return self.history.last_sent

tests/transports/soap_tests.py

@@ -39,21 +39,42 @@ def get_soap_response():
     return response
 
 
-class TestXmlRpcAPICall(testing.TestCase):
+class TestSoapAPICall(testing.TestCase):
 
     def set_up(self):
         self.transport = SoapTransport(endpoint_url='https://api.softlayer.com/soap/v3.1/')
         self.response = get_soap_response()
         self.user = os.getenv('SL_USER')
         self.password = os.environ.get('SL_APIKEY')
-
-    def test_call(self):
         request = Request()
         request.service = 'SoftLayer_Account'
         request.method = 'getObject'
         request.transport_user = self.user
         request.transport_password = self.password
-        data = self.transport(request)
+        self.request = request
+
+    def test_call(self):
+
+        data = self.transport(self.request)
+        pp(data)
+        self.assertEqual(data.get('id'), 307608)
+        self.assertEqual(data.get('companyName'), "SoftLayer Internal - Development Community")
+
+    # def test_debug_call(self):
+
+    #     self.request.mask = "mask[id,accountName,companyName]"
+    #     data = self.transport(self.request)
+        
+    #     self.assertEqual(data.get('id'), 307608)
+    #     debug_data = self.transport.print_reproduceable(self.request)
+    #     print(debug_data['envelope'])
+    #     self.assertEqual(":sdfsdf", debug_data)
+
+    def test_objectMask(self):
+        self.request.mask = "mask[id,companyName]"
+        data = self.transport(self.request)
         pp(data)
+        self.assertEqual(data.get('companyName'), "SoftLayer Internal - Development Community")
+        self.assertIsNone(data.get('address1'))
         self.assertEqual(data.get('id'), 307608)