added unitTests for utility functions, refactored some code in main file (issue #50)

.travis.yml

@@ -10,6 +10,7 @@ before_script: cd app
 
 script:
   - flake8 --exclude=\.eggs,tests,docs --ignore=E124,E303,W504 --max-line-length 80 .
+  - python3 tests/UnitTest.py
 
 deploy:
   provider: pypi

README.md

@@ -103,9 +103,33 @@ username=username,<br>
 password= hashed_password,<br>
 email=email,<br>
 status=OK<br>
+}<br><br>
+<b>Upload/create credentials</b><br>
+POST /users/username/credentials<br>
+RequestBody:<br>
+{ file : text or attached file<br>
+file_name : name,<br>
+encrypted: boolean,<br>
+vault_pass: ansible vault password<br>
 }<br>
-
-
+return : response with successful credential upload status<br>
+<br>
+<b>List credentials</b><br>
+GET /users/username/credentials/file_name<br>
+return : response with encrypted credentials from file<br>
+
+<b>Delete credentials</b><br>
+DELETE /users/username/credentials/file-name<br>
+return : response with successful delete status<br>
+<br>
+<b>Update credentials</b><br>
+PUT /users/username/credentials/file_name<br>
+RequestBody: { file : updated text or updated attached file<br>
+encrypted: boolean,<br>
+vault_pass: ansible vault password<br>
+}<br>
+return : response with successful credential update status<br>
+<br>
 ## Linchpin Project
 LinchPin is a simple cloud orchestration tool. Its intended purpose is managing cloud resources across multiple infrastructures. These resources can be provisioned, decommissioned, and configured all using declarative data and a simple command-line interface.
 

app/config.yml

@@ -11,9 +11,11 @@ inventory_path: /dummy/inventories/*
 linchpin_latest_file_path: /dummy/resources/linchpin.latest
 # path to Pinfile for dummy workspace
 pinfile_json_path: /dummy/PinFile.json
-#details for admin user login
 #file name for linchpin.latest file in resources
 linchpin_latest_name: linchpin.latest
+#details for admin user login
 admin_username: admin
 admin_password: password
-admin_email: admin@xyz
+admin_email: admin@xyz
+# path to creds folder
+creds_path: /tmp

app/data_access_layer/UserBaseDB.py

@@ -42,3 +42,7 @@ def db_reset_api_key(self, username, new_api_key):
     @abstractmethod
     def db_update(self, username, updated_username, password_hash, email):
         pass
+
+    @abstractmethod
+    def db_update_creds_folder(self, username, creds_folder):
+        pass

app/data_access_layer/UserRestDB.py

@@ -24,9 +24,11 @@ def db_insert(self, username, password_hash, api_key_hash,
              admin user
             :param email: User email
         """
+        creds_folder = None
         self.table.insert({'username': username, 'password': password_hash,
                            'api_key': api_key_hash,
-                           'email': email, 'admin': admin})
+                           'email': email, 'admin': admin,
+                           'creds_folder': creds_folder})
 
     def db_search_name(self, username) -> List[Dict]:
         """
@@ -36,7 +38,7 @@ def db_search_name(self, username) -> List[Dict]:
             param username
         """
         user = Query()
-        return self.table.search(user.username == username)
+        return self.table.search(user.username == username)[0]
 
     def db_list_all(self) -> List[Dict]:
         """
@@ -113,3 +115,8 @@ def db_update(self, username, updated_username, password_hash,
         self.table.update({'username': updated_username,
                            'password': password_hash,
                            'email': email}, user.username == username)
+
+    def db_update_creds_folder(self, username, creds_folder):
+        user = Query()
+        self.table.update({'creds_folder': creds_folder},
+                          user.username == username)

app/response_messages/response.py

@@ -3,6 +3,7 @@
 PROVISION_SUCCESS = "Workspace provisioned successfully"
 DESTROY_SUCCESS = "Workspace/resources destroyed successfully"
 NOT_FOUND = "Workspace does not exist"
+NOT_FOUND_USER = "User account does not exist"
 DUPLICATE_WORKSPACE = "Workspace with same name found," \
                       "try again by renaming"
 EMPTY_WORKSPACE = "Only public repositories can be "\
@@ -19,6 +20,7 @@
 LINCHPIN_LATEST_NOT_FOUND = "linchpin.latest not found. " \
                             "Please check that it exists"\
                             " or specify linchpin_latest_path in request"
+CREDENTIALS_FILE_NOT_FOUND = "No credential file found with this name"
 DESTROY_STATUS_SUCCESS = "Destroyed"
 DESTROY_FAILED = "FAILED destroy"
 STATUS_OK = "200 OK"
@@ -36,3 +38,6 @@
                   "the format route?api_key=value"
 MISSING_USERNAME = "please provide the username in request route in "\
                    "the format route?username=value"
+CREDENTIALS_UPLOADED = "Credentials uploaded successfully"
+CREDENTIALS_UPDATED = "Credentials updated sccessfully"
+CREDENTIALS_DELETED = "Credentials deleted successfully"

app/tests/UnitTest.py

@@ -0,0 +1,55 @@
+import unittest
+import app
+from app.data_access_layer import RestDB
+from app.data_access_layer import UserRestDB
+
+BASE_URL = 'http://localhost:5000/api/v1.0'
+
+
+class UnitTest(unittest.TestCase):
+
+    def setUp(self):
+        self.app = app.app.test_client()
+        self.app.testing = True
+
+    def test_db_create_admin_user(self):
+        app.create_admin_user(app.DB_PATH, app.ADMIN_USERNAME,
+                              app.ADMIN_PASSWORD, app.ADMIN_EMAIL)
+        self.assertEqual("admin", app.get_connection_users(app.DB_PATH).db_search_name("admin")['username'])
+
+    def test_get_connection(self):
+        self.assertIsInstance(app.get_connection(app.DB_PATH), RestDB.RestDB)
+
+    def test_get_connection_users(self):
+        self.assertIsInstance(app.get_connection_users(app.DB_PATH), UserRestDB.UserRestDB)
+
+    def test_create_fetch_cmd(self):
+        data = {"name": "test", "url": "www.github.com/CentOS-PaaS-SIG/linchpin",
+                "rootfolder":"/"}
+        identity = "test123"
+        self.assertEqual(app.create_fetch_cmd(data, identity, app.WORKSPACE_DIR),
+                         ["linchpin", "-w " + app.WORKSPACE_DIR + identity, "fetch", "--root", data['rootfolder'],
+                          data['url']])
+
+    def test_create_cmd_workspace(self):
+        data = {"id": "test123",
+                "provision_type": "workspace",
+                "pinfile_path": "/dummy/"
+                }
+        action = "up"
+        self.assertEqual(app.create_cmd_workspace(data, data['id'], action,
+                         app.WORKSPACE_PATH, app.WORKSPACE_DIR,
+                         app.CREDS_PATH), ["linchpin", "-w " + app.WORKSPACE_DIR + data['id'] + data['pinfile_path'],
+                                           "--creds-path", "/", "up"])
+
+    def test_create_cmd_up_pinfile(self):
+        data = {"id": "UnitTest",
+                "provision_type": "pinfile",
+                "pinfile_content":{"test":"pinfile_data"}}
+        self.assertEqual(app.create_cmd_up_pinfile(data, data['id'], app.WORKSPACE_DIR, app.CREDS_PATH),
+                         ["linchpin", "-w " + app.WORKSPACE_DIR + data['id'] + "/dummy", "-p" +
+                          "PinFile.json", "--creds-path", "/", "up"])
+
+
+if __name__ == '__main__':
+    unittest.main()

app/utils/__init__.py

@@ -1,10 +1,7 @@
 import os
-import json
 import uuid
 from app.data_access_layer import RestDB
 from app.data_access_layer import UserRestDB
-from app.response_messages import response
-from flask import jsonify
 from typing import List
 from werkzeug.security import generate_password_hash
 
@@ -74,13 +71,15 @@ def create_fetch_cmd(data, identity, workspace_dir) -> List[str]:
 
 
 def create_cmd_workspace(data, identity, action,
-                         workspace_path, workspace_dir) -> List[str]:
+                         workspace_path, workspace_dir,
+                         creds_folder_path) -> List[str]:
     """
         Creates a list to feed the subprocess for provisioning/
         destroying existing workspaces
         :param data: JSON data from POST requestBody
         :param identity: unique uuid_name assigned to the workspace
         :param action: up or destroy action
+        :param creds_folder_path: path to the credentials folder
         :return a list for the subprocess to run
     """
     if 'pinfile_path' in data:
@@ -89,14 +88,12 @@ def create_cmd_workspace(data, identity, action,
     else:
         check_path = identity
     cmd = ["linchpin", "-w " + workspace_dir + check_path]
+    if 'creds_path' in data:
+        cmd.extend(("--creds-path", data['creds_path']))
+    else:
+        cmd.extend(("--creds-path", creds_folder_path))
     if 'pinfile_name' in data:
         cmd.extend(("-p", data['pinfile_name']))
-        pinfile_name = data['pinfile_name']
-    else:
-        pinfile_name = "PinFile"
-    if not check_workspace_has_pinfile(check_path, pinfile_name,
-                                       workspace_path):
-        return jsonify(status=response.PINFILE_NOT_FOUND)
     cmd.append(action)
     if 'tx_id' in data:
         cmd.extend(("-t", data['tx_id']))
@@ -109,22 +106,23 @@ def create_cmd_workspace(data, identity, action,
 
 def create_cmd_up_pinfile(data,
                           identity,
-                          workspace_path,
                           workspace_dir,
-                          pinfile_json_path) -> List[str]:
+                          creds_folder_path) -> List[str]:
     """
         Creates a list to feed the subprocess for provisioning
         new workspaces instantiated using a pinfile
         :param data: JSON data from POST requestBody
         :param identity: unique uuid_name assigned to the workspace
+        :param creds_folder_path: path to the credentials folder
         :return a list for the subprocess to run
     """
-    pinfile_content = data['pinfile_content']
-    json_pinfile_path = workspace_path + "/" + identity + pinfile_json_path
-    with open(json_pinfile_path, 'w') as json_data:
-        json.dump(pinfile_content, json_data)
     cmd = ["linchpin", "-w " + workspace_dir + identity + "/dummy", "-p" +
-           "PinFile.json", "up"]
+           "PinFile.json"]
+    if 'creds_path' in data:
+        cmd.extend(("--creds-path", data['creds_path']))
+    else:
+        cmd.extend(("--creds-path", creds_folder_path))
+    cmd.append("up")
     if 'inventory_format' in data:
         cmd.extend(("--if", data['inventory_format']))
     return cmd
@@ -140,7 +138,7 @@ def check_workspace_has_pinfile(name, pinfile_name, workspace_path) -> bool:
     return os.listdir(workspace_path + "/" + name).__contains__(pinfile_name)
 
 
-def check_workspace_empty(name, workspace_path) -> bool:
+def check_workspace_empty(name, workspace_path, working_dir) -> bool:
     """
         Verifies if a workspace fetched/created is empty
         :param name: name of the workspace to be verified

requirements.txt

@@ -2,3 +2,4 @@ Flask>=1.0.3
 linchpin<=1.7.5
 flask-swagger-ui>=3.20.9
 flake8>=3.7.7
+ ansible-vault>=1.2.0