Merge pull request #1 from akopkesheshyan/v1.1.0

V1.1.0

Author: Akop Kesheshyan

README.md

@@ -56,6 +56,12 @@ except BitrixError as message:
 List methods return all available items at once. For large collections
 of data use limits.
 
+## Tests
+
+```
+python -m unittest discover
+```
+
 ## Author
 
 Akop Kesheshyan - <akop.kesheshyan@icloud.com>

bitrix24/bitrix24.py

@@ -11,6 +11,7 @@
 """
 import requests
 from time import sleep
+from urllib.parse import urlparse
 from .exceptions import BitrixError
 
 
@@ -27,20 +28,48 @@ class Bitrix24(object):
     """
 
     def __init__(self, domain, timeout=60):
-        self.domain = domain
+        """Create Bitrix24 API object
+        :param domain: str Bitrix24 webhook domain
+        :param timeout: int Timeout for API request in seconds
+        """
+        self.domain = self._prepare_domain(domain)
         self.timeout = timeout
 
-    def _prepare_params(self, params):
-        """Transforms list of params to a valid bitrix array."""
+    def _prepare_domain(self, domain):
+        """Normalize user passed domain to a valid one."""
+        if domain == '' or not isinstance(domain, str):
+            raise Exception('Empty domain')
 
-        new_params = {}
-        for index, value in params.items():
-            if type(value) is dict:
-                for i, v in value.items():
-                    new_params['%s[%s]' % (index, i)] = v
-            else:
-                new_params[index] = value
-        return new_params
+        o = urlparse(domain)
+        user_id, code = o.path.split('/')[2:4]
+        return "{0}://{1}/rest/{2}/{3}".format(o.scheme, o.netloc, user_id, code)
+
+    def _prepare_params(self, params, prev=''):
+        """Transforms list of params to a valid bitrix array."""
+        ret = ''
+        if isinstance(params, dict):
+            for key, value in params.items():
+                if isinstance(value, dict):
+                    if prev:
+                        key = "{0}[{1}]".format(prev, key)
+                    ret += self._prepare_params(value, key)
+                elif (isinstance(value, list) or isinstance(value, tuple)) and len(value) > 0:
+                    for offset, val in enumerate(value):
+                        if isinstance(val, dict):
+                            ret += self._prepare_params(
+                                val, "{0}[{1}][{2}]".format(prev, key, offset))
+                        else:
+                            if prev:
+                                ret += "{0}[{1}][{2}]={3}&".format(
+                                    prev, key, offset, val)
+                            else:
+                                ret += "{0}[{1}]={2}&".format(key, offset, val)
+                else:
+                    if prev:
+                        ret += "{0}[{1}]={2}&".format(prev, key, value)
+                    else:
+                        ret += "{0}={1}&".format(key, value)
+        return ret
 
     def callMethod(self, method, **params):
         """Calls a REST method with specified parameters.
@@ -65,12 +94,12 @@ def callMethod(self, method, **params):
             # Looks like we need to wait until expires limitation time by Bitrix24 API
             sleep(2)
             return self.callMethod(method, **params)
-        
+
         if 'error' in r:
             raise BitrixError(r)
-        if 'page' not in params:
-            params['page'] = 1
-        if 'next' in r and r['total'] > (r['next']*params['page']):
-            params['page'] += 1
+        if 'start' not in params:
+            params['start'] = 0
+        if 'next' in r and r['total'] > params['start']:
+            params['start'] += 50
             return r['result'] + self.callMethod(method, **params)
         return r['result']

tests/test_bitrix24.py

@@ -6,17 +6,11 @@
 class Bitrix24Test(unittest.TestCase):
 
     def setUp(self):
-        self.b24 = Bitrix24(os.environ.get('TEST_DOMAIN'))
+        self.b24 = Bitrix24('https://example.bitrix24.com/rest/1/123456789')
 
-    def test_call_post_method(self):
-        r = self.b24.callMethod('crm.deal.add', fields={
-                                'TITLE': 'Hello World'})
-        self.assertIs(type(r), int)
-
-    def test_call_get_method(self):
-        r = self.b24.callMethod('crm.deal.list', filter={
-                                'TITLE': 'Hello World'})
-        self.assertIs(type(r), list)
+    def test_init_with_empty_domain(self):
+        with self.assertRaises(Exception):
+            Bitrix24('')
 
     def test_call_with_empty_method(self):
         with self.assertRaises(BitrixError):
@@ -26,6 +20,63 @@ def test_call_non_exists_method(self):
         with self.assertRaises(BitrixError):
             self.b24.callMethod('hello.world')
 
+    def test_call_wrong_method(self):
+        with self.assertRaises(BitrixError):
+            self.b24.callMethod('helloworld')
+
+class ParamsPreparationTest(unittest.TestCase):
+
+    def setUp(self):
+        self.b24 = Bitrix24('https://example.bitrix24.com/rest/1/123456789')
+
+    def test_one_level(self):
+        params = {"fruit": "apple"}
+        param_string = self.b24._prepare_params(params)
+        self.assertEqual(param_string, "fruit=apple&")
+
+    def test_one_level_several_items(self):
+        params = {"fruit": "apple", "vegetable": "broccoli"}
+        param_string = self.b24._prepare_params(params)
+        self.assertEqual(param_string, "fruit=apple&vegetable=broccoli&")
+
+    def test_multi_level(self):
+        params = {"fruit": {"citrus": "lemon"}}
+        param_string = self.b24._prepare_params(params)
+        self.assertEqual(param_string, "fruit[citrus]=lemon&")
+
+    def test_multi_level_deep(self):
+        params = {"root": {"level 1": {"level 2": {"level 3": "value"}}}}
+        param_string = self.b24._prepare_params(params)
+        self.assertEqual(
+            param_string, "root[level 1][level 2][level 3]=value&")
+
+    def test_list_dict_mixed(self):
+        params = {"root": {"level 1": [
+            {"list_dict 1": "value 1"}, {"list_dict 2": "value 2"}]}}
+        param_string = self.b24._prepare_params(params)
+        self.assertEqual(
+            param_string, "root[level 1][0][list_dict 1]=value 1&root[level 1][1][list_dict 2]=value 2&")
+
+    def test_multi_level_several_items(self):
+        params = {"fruit": {"citrus": "lemon", "sweet": "apple"}}
+        param_string = self.b24._prepare_params(params)
+        self.assertEqual(
+            param_string, "fruit[citrus]=lemon&fruit[sweet]=apple&")
+
+    def test_list(self):
+        params = {"fruit": ["lemon", "apple"]}
+        param_string = self.b24._prepare_params(params)
+        self.assertEqual(param_string, "fruit[0]=lemon&fruit[1]=apple&")
+
+    def test_tuple(self):
+        params = {"fruit": ("lemon", "apple")}
+        param_string = self.b24._prepare_params(params)
+        self.assertEqual(param_string, "fruit[0]=lemon&fruit[1]=apple&")
+
+    def test_string(self):
+        param_string = self.b24._prepare_params('')
+        self.assertEqual(param_string, "")
+
 
 if __name__ == '__main__':
     unittest.main()