Add tests for Languages

web/pgadmin/browser/server_groups/servers/databases/languages/tests/__init__.py

@@ -0,0 +1,16 @@
+# #################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2016, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+# ##################################################################
+
+from pgadmin.utils.route import BaseTestGenerator
+
+
+class LanguageTestGenerator(BaseTestGenerator):
+
+    def runTest(self):
+        return []

web/pgadmin/browser/server_groups/servers/databases/languages/tests/test_language_add.py

@@ -0,0 +1,80 @@
+# #################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2016, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+# ##################################################################
+from __future__ import print_function
+import json
+import uuid
+
+from pgadmin.utils.route import BaseTestGenerator
+from regression import parent_node_dict
+from regression import test_utils as utils
+from pgadmin.browser.server_groups.servers.databases.tests import \
+    utils as database_utils
+from . import utils as language_utils
+
+
+class LanguagesAddTestCase(BaseTestGenerator):
+    scenarios = [
+        ('Language add test case', dict(url='/browser/language/obj/'))
+    ]
+
+    def setUp(self):
+        self.server_data = parent_node_dict["database"][-1]
+        self.server_id = self.server_data["server_id"]
+        self.db_id = self.server_data['db_id']
+        self.db_name = self.server_data["db_name"]
+        db_con = database_utils.connect_database(self,
+                                                 utils.SERVER_GROUP,
+                                                 self.server_id,
+                                                 self.db_id)
+        if not db_con["info"] == "Database connected.":
+            raise Exception("Could not connect to database.")
+
+    def runTest(self):
+        """This function will add language under test database."""
+
+        db_user = self.server['username']
+
+        self.data = {
+            "lanacl": [],
+            "laninl": "btendscan",
+            "lanowner": db_user,
+            "lanproc": "plpgsql_call_handler",
+            "lanval": "fmgr_c_validator",
+            "name": "language_%s" % str(uuid.uuid4())[1:4],
+            "seclabels": [],
+            "template_list":
+                [
+                    "plperl",
+                    "plperlu",
+                    "plpython2u",
+                    "plpython3u",
+                    "plpythonu",
+                    "pltcl",
+                    "pltclu"
+                ],
+            "trusted": "true"
+                }
+
+        response = self.tester.post(
+            self.url + str(utils.SERVER_GROUP) + '/' +
+            str(self.server_id) + '/' + str(
+                self.db_id) + '/',
+            data=json.dumps(self.data),
+            content_type='html/json')
+
+        self.assertEquals(response.status_code, 200)
+
+    def tearDown(self):
+        """This function delete added language and
+        disconnect the test database."""
+
+        language_utils.delete_language(self.server, self.db_name,
+                                       self.data['name'])
+        database_utils.disconnect_database(self, self.server_id, self.db_id)
+

web/pgadmin/browser/server_groups/servers/databases/languages/tests/test_language_delete.py

@@ -0,0 +1,54 @@
+# #################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2016, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+# ##################################################################
+
+from __future__ import print_function
+import uuid
+
+from pgadmin.utils.route import BaseTestGenerator
+from regression import parent_node_dict
+from regression import test_utils as utils
+from pgadmin.browser.server_groups.servers.databases.tests import \
+    utils as database_utils
+from . import utils as language_utils
+
+
+class LanguagesDeleteTestCase(BaseTestGenerator):
+    scenarios = [
+        ('Language delete test case', dict(url='/browser/language/obj/'))
+    ]
+
+    def setUp(self):
+        self.server_data = parent_node_dict["database"][-1]
+        self.server_id = self.server_data["server_id"]
+        self.db_id = self.server_data['db_id']
+        self.db_name = self.server_data["db_name"]
+        self.lang_name = "language_%s" % str(uuid.uuid4())[1:4]
+
+        db_con = database_utils.connect_database(self,
+                                                 utils.SERVER_GROUP,
+                                                 self.server_id,
+                                                 self.db_id)
+        if not db_con["info"] == "Database connected.":
+            raise Exception("Could not connect to database.")
+        self.language_id = language_utils.create_language(self.server,
+                                                          self.db_name,
+                                                          self.lang_name)
+
+    def runTest(self):
+        """This function will delete language under test database."""
+
+        response = self.tester.delete("{0}{1}/{2}/{3}/{4}".format(
+            self.url, utils.SERVER_GROUP, self.server_id, self.db_id,
+            self.language_id), follow_redirects=True)
+        self.assertEquals(response.status_code, 200)
+
+    def tearDown(self):
+        """This function disconnect the test database."""
+
+        database_utils.disconnect_database(self, self.server_id, self.db_id)

web/pgadmin/browser/server_groups/servers/databases/languages/tests/test_language_get.py

@@ -0,0 +1,56 @@
+# #################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2016, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+# ##################################################################
+from __future__ import print_function
+import uuid
+
+from pgadmin.utils.route import BaseTestGenerator
+from regression import parent_node_dict
+from regression import test_utils as utils
+from pgadmin.browser.server_groups.servers.databases.tests import \
+    utils as database_utils
+from . import utils as language_utils
+
+
+class LanguagesGetTestCase(BaseTestGenerator):
+    scenarios = [
+        ('Language get test case', dict(url='/browser/language/obj/'))
+    ]
+
+    def setUp(self):
+        self.server_data = parent_node_dict["database"][-1]
+        self.server_id = self.server_data["server_id"]
+        self.db_id = self.server_data['db_id']
+        self.db_name = self.server_data["db_name"]
+        self.lang_name = "language_%s" % str(uuid.uuid4())[1:4]
+        db_con = database_utils.connect_database(self,
+                                                 utils.SERVER_GROUP,
+                                                 self.server_id,
+                                                 self.db_id)
+
+        if not db_con["info"] == "Database connected.":
+            raise Exception("Could not connect to database.")
+        self.language_id = language_utils.create_language(self.server,
+                                                          self.db_name,
+                                                          self.lang_name)
+
+    def runTest(self):
+        """This function will get the language under test database."""
+
+        response = self.tester.get("{0}{1}/{2}/{3}/{4}".format(
+            self.url, utils.SERVER_GROUP, self.server_id, self.db_id,
+            self.language_id), follow_redirects=True)
+        self.assertEquals(response.status_code, 200)
+
+    def tearDown(self):
+        """This function delete added language and
+               disconnect the test database."""
+
+        language_utils.delete_language(self.server, self.db_name,
+                                       self.lang_name)
+        database_utils.disconnect_database(self, self.server_id, self.db_id)

web/pgadmin/browser/server_groups/servers/databases/languages/tests/test_language_put.py

@@ -0,0 +1,62 @@
+# #################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2016, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+# ##################################################################
+from __future__ import print_function
+import json
+import uuid
+
+from pgadmin.utils.route import BaseTestGenerator
+from regression import parent_node_dict
+from regression import test_utils as utils
+from pgadmin.browser.server_groups.servers.databases.tests import \
+    utils as database_utils
+from . import utils as language_utils
+
+
+class LanguagesPutTestCase(BaseTestGenerator):
+    scenarios = [
+        ('Language update test case', dict(url='/browser/language/obj/'))
+    ]
+
+    def setUp(self):
+        self.server_data = parent_node_dict["database"][-1]
+        self.server_id = self.server_data["server_id"]
+        self.db_id = self.server_data['db_id']
+        self.db_name = self.server_data["db_name"]
+        self.lang_name = "language_%s" % str(uuid.uuid4())[1:4]
+        db_con = database_utils.connect_database(self,
+                                                 utils.SERVER_GROUP,
+                                                 self.server_id,
+                                                 self.db_id)
+        if not db_con["info"] == "Database connected.":
+            raise Exception("Could not connect to database.")
+        self.language_id = language_utils.create_language(self.server,
+                                                          self.db_name,
+                                                          self.lang_name)
+
+    def runTest(self):
+        """This function will update the language under test database."""
+
+        data = \
+            {
+                "id": self.language_id,
+                "description": "This is test comment."
+            }
+        response = self.tester.put("{0}{1}/{2}/{3}/{4}".format(
+            self.url, utils.SERVER_GROUP, self.server_id, self.db_id,
+            self.language_id), data=json.dumps(data),
+            follow_redirects=True)
+        self.assertEquals(response.status_code, 200)
+
+    def tearDown(self):
+        """This function delete added language and
+        disconnect the test database."""
+
+        language_utils.delete_language(self.server, self.db_name,
+                                       self.lang_name)
+        database_utils.disconnect_database(self, self.server_id, self.db_id)

web/pgadmin/browser/server_groups/servers/databases/languages/tests/utils.py

@@ -0,0 +1,108 @@
+# #################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2016, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+# ##################################################################
+from __future__ import print_function
+import sys
+import traceback
+
+from regression.test_utils import get_db_connection
+
+
+def create_language(server, db_name, lang_name):
+    """
+    This function add a language into database
+    :param server: server details
+    :type server: dict
+    :param db_name: database name
+    :type db_name: str
+    :param lang_name: language name
+    :type lang_name: str
+    :return cast id
+    :rtype: int
+    """
+    try:
+        connection = get_db_connection(db_name,
+                                       server['username'],
+                                       server['db_password'],
+                                       server['host'],
+                                       server['port'])
+
+        pg_cursor = connection.cursor()
+        query = ("CREATE TRUSTED PROCEDURAL LANGUAGE %s "
+                 "HANDLER plpgsql_call_handler" % lang_name)
+        pg_cursor.execute(query)
+        connection.commit()
+        # Get 'oid' from newly created language
+        pg_cursor.execute("SELECT oid from pg_language where lanname='%s'" %
+                          lang_name)
+        language = pg_cursor.fetchone()
+        language_id = language[0]
+        connection.close()
+        return language_id
+    except Exception:
+        traceback.print_exc(file=sys.stderr)
+
+
+def verify_language(server, db_name, lang_name):
+    """
+    This function verifies the language exist in database or not.
+    :param server: server details
+    :type server: dict
+    :param db_name: database name
+    :type db_name: str
+    :param lang_name: language name
+    :type lang_name: str
+    :return language: language record
+    :rtype: tuple
+    """
+    try:
+        connection = get_db_connection(db_name,
+                                       server['username'],
+                                       server['db_password'],
+                                       server['host'],
+                                       server['port'])
+        pg_cursor = connection.cursor()
+        pg_cursor.execute("SELECT oid from pg_language where lanname='%s'" %
+                          lang_name)
+        language = pg_cursor.fetchall()
+        connection.close()
+        return language
+    except Exception:
+        traceback.print_exc(file=sys.stderr)
+
+
+def delete_language(server, db_name, lang_name):
+    """
+    This function delete the language exist in database.
+    :param server: server details
+    :type server: dict
+    :param db_name: database name
+    :type db_name: str
+    :param lang_name: language name
+    :type lang_name: str
+    :return None
+    """
+    try:
+        connection = get_db_connection(db_name,
+                                       server['username'],
+                                       server['db_password'],
+                                       server['host'],
+                                       server['port'])
+        pg_cursor = connection.cursor()
+        pg_cursor.execute("SELECT * from pg_language where lanname='%s'" %
+                          lang_name)
+        languages = pg_cursor.fetchall()
+        language_count = len(languages)
+        if language_count:
+            pg_cursor.execute(
+                "DELETE FROM pg_language where lanname='%s'" %
+                lang_name)
+            connection.commit()
+        connection.close()
+    except Exception:
+        traceback.print_exc(file=sys.stderr)