improvements

Author: root

dynamics365crm/client.py

@@ -7,7 +7,7 @@ class Client:
     header = {"Accept": "application/json, */*", "content-type": "application/json; charset=utf-8",
               'OData-MaxVersion': '4.0', 'OData-Version': '4.0'}
 
-    def __init__(self, client_id, client_secret, token=None):
+    def __init__(self, client_id=None, client_secret=None, token=None):
         self.client_id = client_id
         self.client_secret = client_secret
         self.token = token
@@ -116,32 +116,42 @@ def parse_response(self, response):
         return response.json()
 
     def url_petition(self, redirect_uri, resource):
-        url = "https://login.microsoftonline.com/{0}/oauth2/authorize?client_id={1}&response_type={2}&redirect_uri={3}&response_mode={4}&resource={5}".format(
-            "common", self.client_id, "code", redirect_uri, "query", resource)
+        if self.client_id is not None and redirect_uri is not None and resource is not None:
+            url = "https://login.microsoftonline.com/{0}/oauth2/authorize?client_id={1}&response_type={2}&redirect_uri={3}&response_mode={4}&resource={5}".format(
+                "common", self.client_id, "code", redirect_uri, "query", resource)
+
+            # this part needs an administrator autorization
+            # url = "https://login.microsoftonline.com/common/adminconsent?client_id={0}&redirect_uri={1}".format(
+            #     client_id, redirect_uri)
+            return url
+        else:
+            raise Exception("The attributes necessary to get the url were not obtained.")
 
-        # this part needs an administrator autorization
-        # url = "https://login.microsoftonline.com/common/adminconsent?client_id={0}&redirect_uri={1}".format(
-        #     client_id, redirect_uri)
-        return url
 
     def exchange_code(self, redirect_uri, code):
-        url = 'https://login.microsoftonline.com/common/oauth2/v2.0/token'
-        args = {
-            'client_id': self.client_id,
-            'redirect_uri': redirect_uri,
-            'client_secret': self.client_secret,
-            'code': code,
-            'grant_type': 'authorization_code',
-        }
-        response = requests.post(url, data=args)
-        return self.parse_response(response)
+        if self.client_id is not None and self.client_secret is not None and redirect_uri is not None and code is not None:
+            url = 'https://login.microsoftonline.com/common/oauth2/v2.0/token'
+            args = {
+                'client_id': self.client_id,
+                'redirect_uri': redirect_uri,
+                'client_secret': self.client_secret,
+                'code': code,
+                'grant_type': 'authorization_code',
+            }
+            response = requests.post(url, data=args)
+            return self.parse_response(response)
+        else:
+            raise Exception("The attributes necessary to exchange the code were not obtained.")
 
     def refresh_token(self, refresh_token, redirect_uri, resource):
-        url = "https://login.microsoftonline.com/common/oauth2/token"
-        args = {"client_id": self.client_id, "grant_type": "refresh_token", "refresh_token": refresh_token,
-                "redirect_uri": redirect_uri, "client_secret": self.client_secret, "resource": resource}
-        response = requests.post(url, data=args)
-        return self.parse_response(response)
+        if self.client_id is not None and self.client_secret is not None and refresh_token is not None and redirect_uri is not None and resource is not None:
+            url = "https://login.microsoftonline.com/common/oauth2/token"
+            args = {"client_id": self.client_id, "grant_type": "refresh_token", "refresh_token": refresh_token,
+                    "redirect_uri": redirect_uri, "client_secret": self.client_secret, "resource": resource}
+            response = requests.post(url, data=args)
+            return self.parse_response(response)
+        else:
+            raise Exception("The attributes necessary to refresh the token were not obtained.")
 
     def set_token(self, token):
         """

dynamics365crm/test.py

@@ -3,15 +3,15 @@
 
 """
 MAIN INSTANCE (petition)
-obligatory send the access_token
 """
-petition = Client("")
 tenant_id = ""
 client_id = ""
 client_secret = ""
 dynamics_resource = ""
 CRM_resource = ""
 refresh_token = ""
+token = ""
+petition = Client(client_id=client_id, client_secret=client_secret, token=token)
 
 """
 API ENDPOINTS EXAMPLES
@@ -22,7 +22,7 @@
 REFRESH TOKEN
 to refresh the token you have to send the client_id, the client_secret, the refresh_token, the redirect_uri, and the resource
 Example:
-    refresh = petition.refresh_token(client_id, client_id, refresh_token, "REDIRECT URI", CRM_resource)
+    refresh = petition.refresh_token(refresh_token, redirect_uri, resource)
     pprint.pprint(refresh)
 """
 

test/test_client.py

@@ -0,0 +1,34 @@
+import os
+from unittest import TestCase
+from urllib.parse import urlparse, parse_qs
+from dynamics365crm.client import Client
+
+
+class Dynamics365CRMTestCases(TestCase):
+
+    def setUp(self):
+        self.client_id = os.environ.get('CLIENT_ID')
+        self.client_secret = os.environ.get('CLIENT_SECRET')
+        self.token = os.environ.get('ACCESS_TOKEN')
+        self.redirect_url = os.environ.get('REDIRECT_URL')
+        self.resource = os.environ.get('RESOURCE')
+        self.client = Client(client_id=self.client_id, client_secret=self.client_secret, token=self.token)
+
+    def test_oauth_access(self):
+        url = self.client.url_petition(self.redirect_url, self.resource)
+        self.assertIsInstance(url, str)
+        o = urlparse(url)
+        query = parse_qs(o.query)
+        self.assertIn('client_id', query)
+        self.assertEqual(query['client_id'][0], self.client_id)
+        self.assertIn('redirect_uri', query)
+        self.assertEqual(query['redirect_uri'][0], self.redirect_url)
+
+    def test_get_data(self):
+        response = self.client.get_data(type="contacts", select="fullname, emailaddress1, createdon", orderby="createdon desc", top="1")
+        self.assertIsInstance(response, dict)
+
+    def test_create_data(self):
+        response = self.client.create_data(type='contacts', firstname="NAME", lastname="LASTNAME", middlename="MIDDLENAME", emailaddress1="[email protected]")
+        print(response)
+        self.assertIsInstance(response, bool)