adding functional tests

Author: mflynn-lanl

.envs/.local/.email_template

@@ -0,0 +1,3 @@
+API_KEY=fake-api-key
+EMAIL_URL=https:example.com
+[email protected]

.envs/.local/.phame_template

@@ -0,0 +1,9 @@
+USER=fake_user
+PASSWORD=fake_password
+[email protected]
+ADMIN=fake_admin
+ADMIN_PASSWORD=fake_admin_password
+[email protected]
+PUBLIC=public
+PUBLIC_PASSWORD=fake_public_password
+[email protected]

docker-compose-dev.yml

@@ -41,6 +41,8 @@ services:
       - api_data:/usr/src/app/static/uploads
     expose:
       - 5000
+    env_file:
+      - ./.envs/.local/.phame
     environment:
       - FLASK_ENV=development
       - FLASK_APP=project/__init__.py

docker-compose-prod.yml

@@ -40,8 +40,8 @@ services:
     expose:
       - 5000
     env_file:
-      - ./.envs/.local/.postgres
-      - ./.envs/.local/.email
+      - ./.envs/.prod/.phame
+      - ./.envs/.prod/.email
     environment:
       - FLASK_ENV=development
       - FLASK_APP=project/__init__.py
@@ -61,6 +61,8 @@ services:
       - postgres_backup_local:/backups
     ports:
     - 5435:5432
+    env_file:
+    - ./.envs/.prod/.postgres
     environment:
     - POSTGRES_USER=${POSTGRES_USER}
     - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}

services/phame/project/api/users.py

@@ -167,6 +167,26 @@ def profile():
         return render_template('profile.html', user=current_user)
 
 
+@users_blueprint.route('/delete', methods=['POST'])
+def delete_user():
+    response_object = {'status': 'fail', 'message': 'Cannot delete user'}
+    try:
+        if not current_user.is_admin:
+            response_object['message'] = 'You must be an admin to delete users!'
+            return jsonify(response_object), 422
+        post_data = request.get_json()
+        if not post_data:
+            return jsonify(response_object), 400
+        username = post_data.get('username')
+        user = User.query.filter_by(username=username).first()
+        logging.debug(f'deleting user {username}')
+        db.session.delete(user)
+        db.session.commit()
+    except IntegrityError:
+        db.session.rollback()
+        return jsonify(response_object), 400
+
+
 @users_blueprint.route('/ping', methods=['GET'])
 def ping_pong():
     return jsonify({

services/phame/project/tests/fixtures/example_operator_data.yaml

@@ -1,4 +1,4 @@
 admin_credentials = {'PHAME_ADMIN_USERNAME': 'fake_admin', 'PHAME_ADMIN_PASSWORD': 'fake_password', 'PHAME_ADMIN_NAME': 'fake_person'}
-operator_credentials = {'PHAME_OPERATOR_USERNAME': 'test_user', 'PHAME_OPERATOR_PASSWORD': 'NOT_A_PASSWORD',
-                        'PHAME_OPERATOR_NAME': 'User Name',
-                        'PHAME_OPERATOR_EMAIL': '[email protected]'}
+user_credentials = {'PHAME_USER_USERNAME': 'test_user', 'PHAME_USER_PASSWORD': 'NOT_A_PASSWORD',
+                        'PHAME_USER_NAME': 'User Name',
+                        'PHAME_USER_EMAIL': '[email protected]'}

services/phame/project/tests/fixtures/site_data.yaml

@@ -1,6 +1,8 @@
 site_params:
   PHAME_TEST_URL:
-    'https://localhost/phame'
+    'http://localhost/phame'
+  PHAME_USERS_URL:
+    'http://localhost/users'
 
 create_report_widget_list :
   [{'widget_type':'combo', 'widget_id': 'location','widget_value': 'Ascension M1'},

services/phame/project/tests/functional_tests.py

@@ -7,11 +7,9 @@
 import time
 import os
 from selenium.webdriver.common.by import By
-from urlparse import *
 import argparse
 from selenium.webdriver.support import expected_conditions as EC
 import sys
-print sys.path
 import yaml
 
 sys.path.append(os.path.dirname(__file__))
@@ -25,46 +23,51 @@ class SiteTest(unittest.TestCase):
 
     def setUp(self):
 
-        base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+        base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
         chromeOptions = webdriver.ChromeOptions()
-        prefs = {"download.default_directory": os.path.join(base_dir,'cm','tests')}
+        prefs = {"download.default_directory": os.path.join(base_dir, 'project', 'tests', 'fixtures')}
         chromeOptions.add_experimental_option("prefs", prefs)
-        chrome_driver = os.path.join(base_dir, 'extra', 'chromedriver')
-        self.driver = webdriver.Chrome('/Devel/cmdb/extra/chromedriver', chrome_options=chromeOptions)
-        operator_data = os.path.join(base_dir, 'extra', 'fixtures', 'operator_data.yaml')
+        chrome_driver = os.path.join(base_dir, 'tests', 'extra', 'chromedriver')
+        self.driver = webdriver.Chrome(chrome_driver, options=chromeOptions)
+        operator_data = os.path.join(base_dir, 'tests', 'fixtures', 'operator_data.yaml')
         if not os.path.exists(operator_data):
-            print('operator_data.yaml does not exist, copy example_operator_data.yaml and add your credentials')
+            print('site_data.yaml does not exist, copy example_site_data.yaml and add your credentials')
             sys.exit(0)
         with open(operator_data) as fp:
-            creds = yaml.safe_load(fp)
+            creds = yaml.load(fp, Loader=yaml.FullLoader)
 
         if len(sys.argv) > 1:
             with open(os.path.join(base_dir, 'extra', 'fixtures', sys.argv[1])) as fp:
                 site_data = yaml.safe_load(fp)
         else:
-            with open(os.path.join(base_dir, 'extra', 'fixtures', 'test_sgp_data.yaml')) as fp:
+            with open(os.path.join(base_dir, 'tests', 'fixtures', 'site_data.yaml')) as fp:
                 site_data = yaml.safe_load(fp)
 
         self.url = site_data['site_params']['PHAME_TEST_URL']
-        self.operator_credentials = creds['operator_credentials']
+        self.users_url = site_data['site_params']['PHAME_USERS_URL']
+        self.operator_credentials = creds['user_credentials']
         self.admin_credentials = creds['admin_credentials']
 
+    def tearDown(self):
+        self.driver.get(self.url + '/users/logout')
+        self.driver.close()
+
+
     def admin_login(self):
-        self.admin_logout()
-        self.driver.get(self.url)
+        self.logout()
+        self.driver.get(f"{self.users_url}/login")
         element = WebDriverWait(self.driver, 10).until(
             lambda driver: self.driver.find_element_by_tag_name('a'))
         username = self.driver.find_element_by_name("username")
         password = self.driver.find_element_by_name("password")
         username.clear()
         username.send_keys(self.admin_credentials['PHAME_ADMIN_USERNAME'])
         password.send_keys(self.admin_credentials['PHAME_ADMIN_PASSWORD'])
-        self.driver.find_element_by_name("submitbutton").click()
-
+        self.driver.find_element_by_name("submit").click()
 
-    def admin_logout(self):
-        self.driver.get(self.url + '/accounts/logout')
+    def logout(self):
+        self.driver.get(self.users_url + '/logout')
 
     def login(self):
         username = self.driver.find_element_by_name("username")
@@ -73,4 +76,39 @@ def login(self):
         username.send_keys(self.operator_credentials['PHAME_OPERATOR_USERNAME'])
         password.send_keys(self.operator_credentials['PHAME_OPERATOR_PASSWORD'])
 
-        self.driver.find_element_by_name("submitbutton").click()
+        self.driver.find_element_by_name("submit").click()
+
+    def test_admin_login(self):
+        self.logout()
+        self.driver.get(f"{self.users_url}/login")
+        username = self.driver.find_element_by_name("username")
+        password = self.driver.find_element_by_name("password")
+        username.clear()
+        username.send_keys(self.admin_credentials['PHAME_ADMIN_USERNAME'])
+        password.send_keys(self.admin_credentials['PHAME_ADMIN_PASSWORD'])
+        self.driver.find_element_by_name("submit").click()
+        header = self.driver.find_element_by_xpath('// *[ @ id = "content"] / h1')
+        self.assertIn('PhaME Input', header.text)
+
+    def test_login(self):
+        self.logout()
+        self.driver.get(f"{self.users_url}/login")
+        username = self.driver.find_element_by_name("username")
+        password = self.driver.find_element_by_name("password")
+        username.clear()
+        username.send_keys(self.operator_credentials['PHAME_USER_USERNAME'])
+        password.send_keys(self.operator_credentials['PHAME_USER_PASSWORD'])
+        self.driver.find_element_by_name("submit").click()
+        header = self.driver.find_element_by_xpath('// *[ @ id = "content"] / h1')
+        self.assertIn('PhaME Input', header.text)
+
+    def test_create_user(self):
+        # delete operator user
+        self.driver.get(self.users_url + "/register")
+        username = self.driver.find_element_by_name("username")
+        password = self.driver.find_element_by_name("password")
+        password2 = self.driver.find_element_by_name("password2")
+        email = self.driver.find_element_by_name("email")
+        username.send_keys('test_user')
+        password.send_keys('test_password')
+

services/phame/project/tests/test_users.py

@@ -24,17 +24,6 @@ def add_user(self, username=None, email=None, password=None, is_admin=False):
         user.set_password(pssword)
         db.session.add(user)
         db.session.commit()
-        # with self.client:
-        #     response = self.client.post(url_for('users.register'),
-        #                                 data=json.dumps(
-        #                                     {'username': user,
-        #                                      'email': eml,
-        #                                      'password': pssword,
-        #                                      'password2': pssword2,
-        #                                      'is_admin': is_admin
-        #                                      }),
-        #                                 content_type='application/json', )
-        #     print(response.status_code)
         return user
 
     def create_app(self):
@@ -217,5 +206,58 @@ def test_no_login_redirect(self):
             self.assertEquals(response.status_code, 302)
             self.assertIn('login', response.headers['Location'])
 
+    def test_delete_user(self):
+        admin = self.add_user(username='admin', email='[email protected]',
+                      password='test', is_admin=True)
+        user = self.add_user()
+        response = self.client.get(url_for('users.get_all_users'))
+        data = json.loads(response.data.decode())
+        self.assertEqual(len(data['data']['users']), 2)
+        self.assertEqual(data['data']['users'][1]['username'], 'mark')
+        with self.client:
+            login_user(admin)
+            response = self.client.post(url_for('users.delete_user'),
+                             data=json.dumps(
+                                 {'username': user.username
+                                  }),
+                             content_type='application/json',
+                             )
+            self.assertEqual(response.status_code, 200)
+            response = self.client.get(url_for('users.get_all_users'))
+            data = json.loads(response.data.decode())
+            self.assertEqual(len(data['data']['users']), 1)
+            deleted = True
+            for user in data['data']['users']:
+                if user['username'] == 'mark':
+                    deleted = False
+            self.assertTrue(deleted)
+
+    def test_delete_user_not_admin(self):
+        admin = self.add_user(username='admin', email='[email protected]',
+                      password='test', is_admin=False)
+        user = self.add_user()
+        response = self.client.get(url_for('users.get_all_users'))
+        data = json.loads(response.data.decode())
+        self.assertEqual(len(data['data']['users']), 2)
+        self.assertEqual(data['data']['users'][1]['username'], 'mark')
+        with self.client:
+            login_user(admin)
+            response = self.client.post(url_for('users.delete_user'),
+                             data=json.dumps(
+                                 {'username': user.username
+                                  }),
+                             content_type='application/json',
+                             )
+            self.assertEqual(response.status_code, 422)
+            response = self.client.get(url_for('users.get_all_users'))
+            data = json.loads(response.data.decode())
+            self.assertEqual(len(data['data']['users']), 1)
+            deleted = True
+            for user in data['data']['users']:
+                if user['username'] == 'mark':
+                    deleted = False
+            self.assertTrue(deleted)
+
+
 if __name__ == '__main__':
     unittest.main()

services/phame/requirements.txt

@@ -24,6 +24,7 @@ pandas==0.23.1
 psutil==5.6.2
 psycopg2==2.7.4
 pytz==2018.3
+pyyaml
 redis==2.10.6
 requests
 selenium
@@ -33,4 +34,3 @@ Werkzeug==0.15.3
 requests
 IPython
 Werkzeug==0.15.3
-pyyaml