Postgres Support

test/apiTest.py

@@ -0,0 +1,21 @@
+import requests
+import json
+
+from phpTest import PhpTest
+
+class ApiTestCase(PhpTest):
+
+    def __init__(self):
+        super().__init__("test_api")
+        self.session = requests.Session()
+
+    def api(self, method):
+        return "%s/api/%s" % (self.url, method)
+
+    def test_api(self):
+
+        res = self.session.post(self.api("login"), data={ "username": PhpTest.ADMIN_USERNAME, "password": PhpTest.ADMIN_PASSWORD })
+        self.assertEquals(200, res.status_code, self.httpError(res))
+        self.assertEquals([], self.getPhpErrors(res))
+        obj = json.loads(res.text)
+        self.assertEquals(True, obj["success"], obj["msg"])

test/installTest.py

@@ -1,30 +1,13 @@
-import unittest
 import requests
-import json
-import re
-import string
-import random
 
-class InstallTestCase(unittest.TestCase):
+from phpTest import PhpTest
+
+class InstallTestCase(PhpTest):
 
     def __init__(self, args):
         super().__init__("test_install")
         self.args = args
         self.session = requests.Session()
-        self.url = "http://localhost/"
-
-        keywords = ["Fatal error", "Warning", "Notice", "Parse error", "Deprecated"]
-        self.phpPattern = re.compile("<b>(%s)</b>:" % ("|".join(keywords)))
-
-    def randomString(self, length):
-       letters = string.ascii_lowercase + string.ascii_uppercase + string.digits
-       return ''.join(random.choice(letters) for i in range(length))
-
-    def httpError(self, res):
-        return "Server returned: %d %s" % (res.status_code, res.reason)
-
-    def getPhpErrors(self, res):
-        return [line for line in res.text.split("\n") if self.phpPattern.search(line)]
 
     def test_install(self):
 
@@ -39,46 +22,44 @@ class InstallTestCase(unittest.TestCase):
         self.assertEquals([], self.getPhpErrors(res))
 
         # Create User
-        valid_username = self.randomString(16)
-        valid_password = self.randomString(16)
 
         # 1. Invalid username
         for username in ["a", "a"*33]:
             res = self.session.post(self.url, data={ "username": username, "password": "123456", "confirmPassword": "123456" })
             self.assertEquals(200, res.status_code, self.httpError(res))
             self.assertEquals([], self.getPhpErrors(res))
-            obj = json.loads(res.text)
+            obj = self.getJson(res)
             self.assertEquals(False, obj["success"])
             self.assertEquals("The username should be between 5 and 32 characters long", obj["msg"])
 
         # 2. Invalid password
-        res = self.session.post(self.url, data={ "username": valid_username, "password": "1", "confirmPassword": "1" })
+        res = self.session.post(self.url, data={ "username": PhpTest.ADMIN_USERNAME, "password": "1", "confirmPassword": "1" })
         self.assertEquals(200, res.status_code, self.httpError(res))
         self.assertEquals([], self.getPhpErrors(res))
-        obj = json.loads(res.text)
+        obj = self.getJson(res)
         self.assertEquals(False, obj["success"])
         self.assertEquals("The password should be at least 6 characters long", obj["msg"])
 
         # 3. Passwords do not match
-        res = self.session.post(self.url, data={ "username": valid_username, "password": "1", "confirmPassword": "2" })
+        res = self.session.post(self.url, data={ "username": PhpTest.ADMIN_USERNAME, "password": "1", "confirmPassword": "2" })
         self.assertEquals(200, res.status_code, self.httpError(res))
         self.assertEquals([], self.getPhpErrors(res))
-        obj = json.loads(res.text)
+        obj = self.getJson(res)
         self.assertEquals(False, obj["success"])
         self.assertEquals("The given passwords do not match", obj["msg"])
 
         # 4. User creation OK
-        res = self.session.post(self.url, data={ "username": valid_username, "password": valid_password, "confirmPassword": valid_password })
+        res = self.session.post(self.url, data={ "username": PhpTest.ADMIN_USERNAME, "password": PhpTest.ADMIN_PASSWORD, "confirmPassword": PhpTest.ADMIN_PASSWORD })
         self.assertEquals(200, res.status_code, self.httpError(res))
         self.assertEquals([], self.getPhpErrors(res))
-        obj = json.loads(res.text)
+        obj = self.getJson(res)
         self.assertEquals(True, obj["success"])
 
         # Mail: SKIP
         res = self.session.post(self.url, data={ "skip": "true" })
         self.assertEquals(200, res.status_code, self.httpError(res))
         self.assertEquals([], self.getPhpErrors(res))
-        obj = json.loads(res.text)
+        obj = self.getJson(res)
         self.assertEquals(True, obj["success"])
 
         # Creation successful:

test/phpTest.py

@@ -0,0 +1,36 @@
+import unittest
+import string
+import random
+import re
+import json
+
+class PhpTest(unittest.TestCase):
+
+    def randomString(length):
+       letters = string.ascii_lowercase + string.ascii_uppercase + string.digits
+       return ''.join(random.choice(letters) for i in range(length))
+
+    ADMIN_USERNAME = "Administrator"
+    ADMIN_PASSWORD = randomString(16)
+
+    def __init__(self, test_method):
+        super().__init__(test_method)
+        keywords = ["Fatal error", "Warning", "Notice", "Parse error", "Deprecated"]
+        self.phpPattern = re.compile("<b>(%s)</b>:" % ("|".join(keywords)))
+        self.url = "http://localhost/"
+
+    def httpError(self, res):
+        return "Server returned: %d %s" % (res.status_code, res.reason)
+
+    def getPhpErrors(self, res):
+        return [line for line in res.text.split("\n") if self.phpPattern.search(line)]
+
+    def getJson(self, res):
+        obj = None
+        try:
+            obj = json.loads(res.text)
+        except:
+            pass
+        finally:
+            self.assertTrue(isinstance(obj, dict), res.text)
+            return obj

test/requirements.txt

@@ -1,2 +1,3 @@
 requests==2.23.0
+psycopg2==2.8.4
 mysql_connector_repackaged==0.3.1

test/test.py

@@ -6,9 +6,12 @@ import argparse
 import random
 import string
 import unittest
+
 import mysql.connector
+import psycopg2
 
 from installTest import InstallTestCase
+from apiTest import ApiTestCase
 
 CONFIG_FILES = ["../core/Configuration/Database.class.php","../core/Configuration/JWT.class.php","../core/Configuration/Mail.class.php"]
 
@@ -19,15 +22,13 @@ def randomName(length):
 def performTest(args):
     suite = unittest.TestSuite()
     suite.addTest(InstallTestCase(args))
+    suite.addTest(ApiTestCase())
     runner = unittest.TextTestRunner()
     runner.run(suite)
 
 def testMysql(args):
 
     # Create a temporary database
-    cursor = None
-    database = None
-    connection = None
     if args.database is None:
         args.database = "webbase_test_%s" % randomName(6)
         config = {
@@ -43,6 +44,7 @@ def testMysql(args):
         cursor = connection.cursor()
         print("[ ] Creating temporary databse %s" % args.database)
         cursor.execute("CREATE DATABASE %s" % args.database)
+        cursor.commit()
         print("[+] Success")
 
     # perform test
@@ -60,6 +62,37 @@ def testMysql(args):
         print("[ ] Closing connection…")
         connection.close()
 
+def testPostgres(args):
+
+    # Create a temporary database
+    if args.database is None:
+        args.database = "webbase_test_%s" % randomName(6)
+        connection_string = "host=%s port=%d user=%s password=%s" % (args.host, args.port, args.username, args.password)
+
+        print("[ ] Connecting to dbms…")
+        connection = psycopg2.connect(connection_string)
+        connection.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
+        print("[+] Success")
+        cursor = connection.cursor()
+        print("[ ] Creating temporary databse %s" % args.database)
+        cursor.execute("CREATE DATABASE %s" % args.database)
+        print("[+] Success")
+
+    # perform test
+    try:
+        args.type = "postgres"
+        performTest(args)
+    finally:
+        if cursor is not None:
+            print("[ ] Deleting temporary database")
+            cursor.execute("DROP DATABASE %s" % args.database)
+            cursor.close()
+            print("[+] Success")
+
+    if connection is not None:
+        print("[ ] Closing connection…")
+        connection.close()
+
 if __name__ == "__main__":
 
     supportedDbms = {
@@ -95,3 +128,9 @@ if __name__ == "__main__":
 
     if args.dbms == "mysql":
         testMysql(args)
+    elif args.dbms == "postgres":
+        testPostgres(args)
+
+    for f in CONFIG_FILES:
+        if os.path.isfile(f):
+            os.remove(f)