adding all the user groups endpoints

Author: Naor Livne

NebulaPythonSDK/sdk.py

@@ -240,3 +240,45 @@ def create_user(self, user, config):
         response = requests.request("POST", url, data=payload, headers=headers, timeout=self.request_timeout)
         filtered_response = {"status_code": response.status_code, "reply": response.json()}
         return filtered_response
+
+    # list all user groups
+    def list_user_groups(self):
+        url = self.host + "/api/" + self.API_VERSION + "/user_groups"
+        headers = self.headers
+        response = requests.request("GET", url, headers=headers, timeout=self.request_timeout)
+        filtered_response = {"status_code": response.status_code, "reply": response.json()}
+        return filtered_response
+
+    # list a user group configuration
+    def list_user_group(self, user):
+        url = self.host + "/api/" + self.API_VERSION + "/user_groups/" + user
+        headers = self.headers
+        response = requests.request("GET", url, headers=headers, timeout=self.request_timeout)
+        filtered_response = {"status_code": response.status_code, "reply": response.json()}
+        return filtered_response
+
+    # delete an existing nebula user group, no confirmation required in SDK so be careful
+    def delete_user_group(self, user):
+        url = self.host + "/api/" + self.API_VERSION + "/user_groups/" + user
+        headers = self.headers
+        response = requests.request("DELETE", url, headers=headers, timeout=self.request_timeout)
+        filtered_response = {"status_code": response.status_code, "reply": response.json()}
+        return filtered_response
+
+    # update a user group
+    def update_user_group(self, user, config):
+        url = self.host + "/api/" + self.API_VERSION + "/user_groups/" + user + "/update"
+        payload = json.dumps(config)
+        headers = self.headers
+        response = requests.request("PUT", url, data=payload, headers=headers, timeout=self.request_timeout)
+        filtered_response = {"status_code": response.status_code, "reply": response.json()}
+        return filtered_response
+
+    # create a new nebula user group, requires the group name to create and a complete dict of the config values for it
+    def create_user_group(self, user, config):
+        url = self.host + "/api/" + self.API_VERSION + "/user_groups/" + user
+        payload = json.dumps(config)
+        headers = self.headers
+        response = requests.request("POST", url, data=payload, headers=headers, timeout=self.request_timeout)
+        filtered_response = {"status_code": response.status_code, "reply": response.json()}
+        return filtered_response

README.md

@@ -119,4 +119,32 @@ connection.refresh_user_token("user_name")
 # delete a user
 connection.delete_user("user_name")
 
+# create a user group
+user_group_config = {
+                        "group_members": [
+                            "user_name"
+                            ], 
+                        "pruning_allowed": True, 
+                        "apps": {
+                            "app_name": "rw"
+                            }, 
+                        "device_groups": {
+                            "device_group_name": "ro"
+                            }, 
+                        "admin": False
+                    }
+connection.create_user_group("user_group_name", user_group_config)
+
+# list user group info
+connection.list_user_group("user_group_name")
+
+# list all users group
+connection.list_user_groups()
+
+# update a user group
+user_group_config = {"admin": False}
+connection.update_user_group("user_group_name", user_group_config)
+
+# delete a user group
+connection.delete_user_group("user_group_name")
 ```

test/test_nebula_python_sdk.py

@@ -285,3 +285,63 @@ def test_user_workflow(self, user="unit_test_user"):
         reply = nebula_connection_object.delete_user(user)
         self.assertEqual(reply["status_code"], 200)
         self.assertEqual(reply["reply"], {})
+
+    def test_user_group_workflow(self, user="unit_test_user", user_group="unit_test_user_group",
+                                 app="unit_test_device_group_app"):
+        nebula_connection_object = nebula_connection()
+
+        # make sure no user & user group exists prior to the run
+        nebula_connection_object.delete_user(user)
+        nebula_connection_object.delete_user_group(user_group)
+        nebula_connection_object.delete_app(app)
+
+        # creating a user that will be used in the tests
+        user_config = {"password": "unit_test_password", "token": "unit_test_token"}
+        nebula_connection_object.create_user(user, user_config)
+
+        # creating a app that will be used in the tests
+        create_temp_app(nebula_connection_object, app)
+
+        # check creating a user group
+        user_group_config = {
+            "group_members": [user],
+            "pruning_allowed": True,
+            "apps": {app: "rw"},
+            "device_groups": {},
+            "admin": False
+        }
+
+        reply = nebula_connection_object.create_user_group(user_group, user_group_config)
+        self.assertEqual(reply["status_code"], 200)
+        self.assertEqual(reply["reply"]["user_group"], user_group)
+        self.assertEqual(reply["reply"]["group_members"], [user])
+        self.assertTrue(reply["reply"]["pruning_allowed"])
+        self.assertEqual(reply["reply"]["apps"], {app: "rw"})
+        self.assertEqual(reply["reply"]["device_groups"], {})
+        self.assertFalse(reply["reply"]["admin"])
+
+        # check getting user group info
+        reply = nebula_connection_object.list_user_group(user_group)
+        self.assertEqual(reply["status_code"], 200)
+        self.assertEqual(reply["reply"]["user_group"], user_group)
+        self.assertEqual(reply["reply"]["group_members"], [user])
+        self.assertTrue(reply["reply"]["pruning_allowed"])
+        self.assertEqual(reply["reply"]["apps"], {app: "rw"})
+        self.assertEqual(reply["reply"]["device_groups"], {})
+        self.assertFalse(reply["reply"]["admin"])
+
+        # check updating a user group
+        user_group_config = {"admin": True}
+        reply = nebula_connection_object.update_user_group(user_group, user_group_config)
+        self.assertEqual(reply["status_code"], 200)
+        self.assertEqual(reply["reply"]["user_group"], user_group)
+        self.assertEqual(reply["reply"]["group_members"], [user])
+        self.assertTrue(reply["reply"]["pruning_allowed"])
+        self.assertEqual(reply["reply"]["apps"], {app: "rw"})
+        self.assertEqual(reply["reply"]["device_groups"], {})
+        self.assertTrue(reply["reply"]["admin"])
+
+        # check deleting a user group
+        reply = nebula_connection_object.delete_user_group(user_group)
+        self.assertEqual(reply["status_code"], 200)
+        self.assertEqual(reply["reply"], {})