adding all of the user endpoints

Author: Naor Livne

NebulaPythonSDK/sdk.py

@@ -148,8 +148,7 @@ def list_device_group(self, device_group):
         filtered_response = {"status_code": response.status_code, "reply": response.json()}
         return filtered_response
 
-    # create a new nebula delete_device_group, requires the app name to create and a complete dict of the config values
-    # for it
+    # create a new nebula device_group, requires the app name to create and a complete dict of the config values for it
     def create_device_group(self, device_group, config):
         url = self.host + "/api/" + self.API_VERSION + "/device_groups/" + device_group
         payload = json.dumps(config)
@@ -158,7 +157,7 @@ def create_device_group(self, device_group, config):
         filtered_response = {"status_code": response.status_code, "reply": response.json()}
         return filtered_response
 
-    # delete an existing nebula delete_device_group, no confirmation required in SDK so be careful
+    # delete an existing nebula device_group, no confirmation required in SDK so be careful
     def delete_device_group(self, device_group):
         url = self.host + "/api/" + self.API_VERSION + "/device_groups/" + device_group
         headers = self.headers
@@ -192,16 +191,52 @@ def list_reports(self, page_size=10, hostname=None, device_group=None, report_cr
         filtered_response = {"status_code": response.status_code, "reply": response.json()}
         return filtered_response
 
-    # TODO - list users
-
-    # TODO - get user info
-
-    # TODO - delete a user
+    # list all users
+    def list_users(self):
+        url = self.host + "/api/" + self.API_VERSION + "/users"
+        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
 
-    # TODO -  update a user
+    # list a user configuration
+    def list_user(self, user):
+        url = self.host + "/api/" + self.API_VERSION + "/users/" + 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
 
-    # TODO - refresh a user token
+    # delete an existing nebula user, no confirmation required in SDK so be careful
+    def delete_user(self, user):
+        url = self.host + "/api/" + self.API_VERSION + "/users/" + 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
 
-    # TODO - create new user
+    # update a user pass and\or bearer token
+    def update_user(self, user, config):
+        url = self.host + "/api/" + self.API_VERSION + "/users/" + 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
 
+    # refresh a token for a user
+    def refresh_user_token(self, user):
+        url = self.host + "/api/" + self.API_VERSION + "/users/" + user + "/refresh"
+        headers = self.headers
+        response = requests.request("POST", url, headers=headers, timeout=self.request_timeout)
+        filtered_response = {"status_code": response.status_code, "reply": response.json()}
+        return filtered_response
 
+    # create a new nebula user, requires the user name to create and a complete dict of the config values for it
+    def create_user(self, user, config):
+        url = self.host + "/api/" + self.API_VERSION + "/users/" + 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

@@ -99,4 +99,24 @@ connection.update_device_group("device_group_name", device_group_config)
 # list paginated reports
 connection.list_reports()
 
+# create a user
+user_config = {"password": "user_password", "token": "user_token"}
+connection.create_user("user_name", user_config)
+
+# list user info
+connection.list_user("user_name")
+
+# list all users
+connection.list_users()
+
+# update a user
+user_config = {"password": "new_password", "token": "new_token"}
+connection.update_user("user_name", user_config)
+
+# refresh a user token
+connection.refresh_user_token("user_name")
+
+# delete a user
+connection.delete_user("user_name")
+
 ```

setup.py

@@ -2,7 +2,7 @@
 
 __author__ = 'Naor Livne'
 __author_email__ = 'naorlivne@gmail.com'
-__version__ = '2.2.1'
+__version__ = '2.3.0'
 
 with open('README.md') as f:
     long_description = f.read()

test/test_nebula_python_sdk.py

@@ -238,17 +238,50 @@ def test_list_reports(self):
         self.assertEqual(reply["reply"]["data"], None)
         self.assertEqual(reply["reply"]["last_id"], None)
 
-    # TODO - create unit tests for the user flow
-
-    # TODO - list users
+    def test_list_users(self):
+        nebula_connection_object = nebula_connection()
+        reply = nebula_connection_object.list_users()
+        self.assertEqual(reply["status_code"], 200)
+        self.assertTrue(isinstance(reply["reply"]["users"], list))
 
-    # TODO - get user info
+    def test_user_workflow(self, user="unit_test_user"):
+        nebula_connection_object = nebula_connection()
 
-    # TODO - delete a user
+        # make sure no user exists prior to the run
+        nebula_connection_object.delete_user(user)
 
-    # TODO -  update a user
+        # check creating a user
+        user_config = {"password": "unit_test_password", "token": "unit_test_token"}
+        reply = nebula_connection_object.create_user(user, user_config)
+        self.assertEqual(reply["status_code"], 200)
+        self.assertTrue(isinstance(reply["reply"]["password"], str))
+        self.assertTrue(isinstance(reply["reply"]["token"], str))
 
-    # TODO - refresh a user token
+        # check getting user info
+        reply = nebula_connection_object.list_user(user)
+        self.assertEqual(reply["status_code"], 200)
+        self.assertEqual(reply["reply"]["user_name"], user)
+        self.assertTrue(isinstance(reply["reply"]["password"], str))
+        self.assertTrue(isinstance(reply["reply"]["token"], str))
 
-    # TODO - create new user
+        #check updating a user
+        user_config = {"password": "unit_test_password_updated", "token": "unit_test_token_updated"}
+        reply = nebula_connection_object.update_user(user, user_config)
+        self.assertEqual(reply["status_code"], 200)
+        self.assertEqual(reply["reply"]["user_name"], user)
+        self.assertTrue(isinstance(reply["reply"]["password"], str))
+        self.assertTrue(isinstance(reply["reply"]["token"], str))
+        user_config = {"password": "unit_test_password", "token": "unit_test_token"}
+        reply_second = nebula_connection_object.update_user(user, user_config)
+        self.assertNotEqual(reply["reply"]["password"], reply_second["reply"]["password"])
+        self.assertNotEqual(reply["reply"]["token"], reply_second["reply"]["token"])
+
+        # check refreshing a user token
+        reply = nebula_connection_object.refresh_user_token(user)
+        self.assertEqual(reply["status_code"], 200)
+        self.assertTrue(isinstance(reply["reply"]["token"], str))
 
+        # check deleting a user
+        reply = nebula_connection_object.delete_user(user)
+        self.assertEqual(reply["status_code"], 200)
+        self.assertEqual(reply["reply"], {})