IntenseWebs/pgadmin4
3
0
Fork 0
mirror of https://github.com/pgadmin-org/pgadmin4.git synced 2025-02-25 18:55:31 -06:00
Code Issues Packages Projects Releases Wiki Activity

Improve code coverage and API test cases for FTS Configuration, FTS Parser, FTS Dictionaries, and FTS Template. Fixes #5329.

Browse Source
This commit is contained in:
Nikhil Mohite 2020-07-10 11:40:08 +05:30 committed by Akshay Joshi
parent 357a9179b1
commit 7420673a0f
37 changed files with 3579 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/en_US/release_notes_4_24.rst
View File

@ -18,6 +18,7 @@ Housekeeping
************
| `Issue #5326 <https://redmine.postgresql.org/issues/5326>`_ - Improve code coverage and API test cases for Domain and Domain Constraints.
| `Issue #5329 <https://redmine.postgresql.org/issues/5329>`_ - Improve code coverage and API test cases for FTS Configuration, FTS Parser, FTS Dictionaries, and FTS Template.
Bug fixes
*********

291
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/fts_configurations_test_data.json Normal file
View File

@ -0,0 +1,291 @@
{
"dependency_dependent_fts_configurations": [
{
"name": "Get FTS Configuration dependency",
"url": "/browser/fts_configuration/dependency/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get FTS Configuration dependent",
"url": "/browser/fts_configuration/dependent/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"get_fts_configuration_nodes_and_node": [
{
"name": "Get FTS Configuration nodes",
"url": "/browser/fts_configuration/nodes/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching FTS Configuration nodes",
"url": "/browser/fts_configuration/nodes/",
"error_fetching_fts_configuration": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts configuration nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get FTS Configuration node",
"url": "/browser/fts_configuration/nodes/",
"is_positive_test": true,
"node": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching FTS Configuration node",
"url": "/browser/fts_configuration/nodes/",
"error_fetching_fts_configuration": true,
"is_positive_test": false,
"node": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts configuration nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while fetching FTS Configuration node",
"url": "/browser/fts_configuration/nodes/",
"error_fetching_fts_configuration": true,
"set_wrong_fts_configuration_value": true,
"is_positive_test": false,
"node": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(True, {'rows': []})"
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get FTS Configuration nodes list",
"url": "/browser/fts_configuration/obj/",
"is_positive_test": true,
"node": false,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Configuration nodes list",
"url": "/browser/fts_configuration/obj/",
"is_positive_test": false,
"error_fetching_fts_configuration": true,
"node": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts configuration nodes list')"
},
"expected_data": {
"status_code": 500
}
}
],
"fts_configuration_get_sql": [
{
"name": "Get fts configuration SQL",
"url": "/browser/fts_configuration/sql/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a created fts configuration sql",
"url": "/browser/fts_configuration/sql/",
"error_fetching_fts_configuration": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error while fetching a fts configuration sql')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while fetching a created fts configuration sql",
"url": "/browser/fts_configuration/sql/",
"error_fetching_fts_configuration": true,
"is_positive_test": false,
"msql": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(True, None)"
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get fts configuration msql: With existing fts configuration id.",
"url": "/browser/fts_configuration/msql/",
"is_positive_test": true,
"msql": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Error while fetching a created fts configuration sql",
"url": "/browser/fts_configuration/msql/",
"error_fetching_fts_configuration": true,
"is_positive_test": false,
"msql": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching a fts configuration sql')"
},
"expected_data": {
"status_code": 500
}
}
],
"get_fts_configuration_get_dictionaries": [
{
"name": "Get fts configuration dictionaries",
"url": "/browser/fts_configuration/dictionaries/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get fts configuration dictionaries",
"url": "/browser/fts_configuration/dictionaries/",
"is_positive_test": false,
"error_fetching_fts_configuration": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching a fts configuration dictionaries')"
},
"expected_data": {
"status_code": 500
}
}
],
"get_fts_configuration_get_tokens": [
{
"name": "Get fts configuration tokens",
"url": "/browser/fts_configuration/tokens/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get fts configuration tokens",
"url": "/browser/fts_configuration/tokens/",
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 0)"
},
"expected_data": {
"status_code": 500
}
}
],
"get_fts_configuration_copy_config": [
{
"name": "Get fts configuration copy",
"url": "/browser/fts_configuration/copyConfig/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get fts configuration copy negative",
"url": "/browser/fts_configuration/copyConfig/",
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching a fts configuration dictionaries')"
},
"expected_data": {
"status_code": 500
}
}
],
"get_fts_configuration_parsers": [
{
"name": "Get fts configuration parsers",
"url": "/browser/fts_configuration/parsers/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get fts configuration parsers",
"url": "/browser/fts_configuration/parsers/",
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching a fts configuration dictionaries')"
},
"expected_data": {
"status_code": 500
}
}
],
"delete_fts_configuration": [
]
}

121
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/test_fts_configuration_copyConfig.py Normal file
View File

@ -0,0 +1,121 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_configurations.tests import utils as fts_config_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_configurations_utils
class FTSConfigurationDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS configuration
under test schema. """
scenarios = utils.generate_scenarios(
'get_fts_configuration_copy_config',
fts_configurations_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_configuration_func_%s" % str(uuid.uuid4())[1:8]
self.fts_configuration_name = "fts_configuration_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Configuration are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_configuration_id = fts_configurations_utils. \
create_fts_configuration(
self.server, self.db_name, self.schema_name,
self.fts_configuration_name)
def _fts_configuration_copy_config(self):
"""
This functions returns the fts configuration copy config
:return: fts configuration copy config
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def runTest(self):
""" This function will add new FTS configuration under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_configurations_utils.verify_fts_configuration(
self.server, self.db_name, self.fts_configuration_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Configuration.")
if self.is_positive_test:
response = self._fts_configuration_copy_config()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self._fts_configuration_copy_config()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_config and disconnect the test
database."""
fts_config_utils.delete_fts_configurations(self.server, self.db_name,
self.schema_name,
self.fts_configuration_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

9
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/test_fts_configuration_delete.py
View File

@ -78,6 +78,15 @@ class FTSConfDeleteTestCase(BaseTestGenerator):
follow_redirects=True)
self.assertEquals(delete_response.status_code, 200)
negative_delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(0),
follow_redirects=True)
self.assertEquals(negative_delete_response.status_code, 500)
def tearDown(self):
"""This function disconnect the test database."""

118
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/test_fts_configuration_dependency_dependent.py Normal file
View File

@ -0,0 +1,118 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import json
import uuid
from pgadmin.utils import server_utils as server_utils
from regression import trigger_funcs_utils as fts_config_funcs_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_configurations.tests import utils as fts_config_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as fts_configurations_utils
class FTSConfigurationDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS configuration
under test schema. """
scenarios = utils.generate_scenarios(
'dependency_dependent_fts_configurations',
fts_configurations_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_configuration_func_%s" % str(uuid.uuid4())[1:8]
self.fts_configuration_name = "fts_configuration_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Configuration are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_configuration_id = fts_configurations_utils. \
create_fts_configuration(
self.server, self.db_name, self.schema_name,
self.fts_configuration_name)
def runTest(self):
""" This function will add new FTS configuration under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_configurations_utils.verify_fts_configuration(
self.server, self.db_name, self.fts_configuration_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Configuration.")
if self.is_positive_test:
response = self.get_dependency_dependent()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def get_dependency_dependent(self):
"""
This function returns the fts configuration dependency and dependent
:return: fts configuration dependency and dependent
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(self.fts_configuration_id),
follow_redirects=True)
def tearDown(self):
"""This function delete the fts_config and disconnect the test
database."""
fts_config_utils.delete_fts_configurations(self.server, self.db_name,
self.schema_name,
self.fts_configuration_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

122
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/test_fts_configuration_get_dictionaries.py Normal file
View File

@ -0,0 +1,122 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_configurations.tests import utils as fts_config_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_configurations_utils
class FTSConfigurationDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS configuration
under test schema. """
scenarios = utils.generate_scenarios(
'get_fts_configuration_get_dictionaries',
fts_configurations_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_configuration_func_%s" % str(uuid.uuid4())[1:8]
self.fts_configuration_name = "fts_configuration_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Configuration are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_configuration_id = fts_configurations_utils. \
create_fts_configuration(
self.server, self.db_name, self.schema_name,
self.fts_configuration_name)
def get_fts_configuration_dictionaries(self):
"""
This functions returns the fts configuration dictionaries
:return: fts configuration dictionaries
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def runTest(self):
""" This function will add new FTS configuration under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_configurations_utils.verify_fts_configuration(
self.server, self.db_name, self.fts_configuration_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Configuration.")
if self.is_positive_test:
response = self.get_fts_configuration_dictionaries()
else:
if hasattr(self, "error_fetching_fts_configuration"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_fts_configuration_dictionaries()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_config and disconnect the test
database."""
fts_config_utils.delete_fts_configurations(self.server, self.db_name,
self.schema_name,
self.fts_configuration_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

142
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/test_fts_configuration_get_nodes.py Normal file
View File

@ -0,0 +1,142 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_configurations.tests import utils as fts_config_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_configurations_utils
class FTSConfigurationNodesTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS configuration
under test schema. """
scenarios = utils.generate_scenarios(
'get_fts_configuration_nodes_and_node',
fts_configurations_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_configuration_func_%s" % str(uuid.uuid4())[1:8]
self.fts_configuration_name = "fts_configuration_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Configuration are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_configuration_id = fts_configurations_utils. \
create_fts_configuration(
self.server, self.db_name, self.schema_name,
self.fts_configuration_name)
def get_fts_configuration_nodes(self):
"""
This functions returns the fts configuration nodes
:return: fts configuration nodes
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def get_fts_configuration_node(self):
"""
This functions returns the fts configuration node
:return: fts configuration node
"""
if hasattr(self, "set_wrong_fts_configuration_value"):
self.fts_configuration_id = 0
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/' +
str(self.fts_configuration_id), content_type='html/json')
def runTest(self):
""" This function will add new FTS configuration under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_configurations_utils.verify_fts_configuration(
self.server, self.db_name, self.fts_configuration_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Configuration.")
if self.is_positive_test:
if hasattr(self, "node"):
response = self.get_fts_configuration_node()
else:
response = self.get_fts_configuration_nodes()
else:
if hasattr(self, "error_fetching_fts_configuration"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "node"):
response = self.get_fts_configuration_node()
else:
response = self.get_fts_configuration_nodes()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_config and disconnect the test
database."""
fts_config_utils.delete_fts_configurations(
self.server, self.db_name, self.schema_name,
self.fts_configuration_name
)
database_utils.disconnect_database(self, self.server_id, self.db_id)

124
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/test_fts_configuration_get_sql.py Normal file
View File

@ -0,0 +1,124 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_configurations.tests import utils as fts_config_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_configurations_utils
class FTSConfigurationDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS configuration
under test schema. """
scenarios = utils.generate_scenarios(
'fts_configuration_get_sql',
fts_configurations_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_configuration_func_%s" % str(uuid.uuid4())[1:8]
self.fts_configuration_name = "fts_configuration_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Configuration are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_configuration_id = fts_configurations_utils. \
create_fts_configuration(
self.server, self.db_name, self.schema_name,
self.fts_configuration_name)
def get_fts_configuration_sql(self):
"""
this function fetch event trigger sql
:return: event trigger sql details
"""
return self.tester.get(
self.url +
str(utils.SERVER_GROUP) + '/' + str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/' +
str(self.fts_configuration_id),
content_type='html/json'
)
def runTest(self):
""" This function will add new FTS configuration under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_configurations_utils.verify_fts_configuration(
self.server, self.db_name, self.fts_configuration_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Configuration.")
if self.is_positive_test:
response = self.get_fts_configuration_sql()
else:
if hasattr(self, "error_fetching_fts_configuration"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_fts_configuration_sql()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_config and disconnect the test
database."""
fts_config_utils.delete_fts_configurations(self.server, self.db_name,
self.schema_name,
self.fts_configuration_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

121
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/test_fts_configuration_parsers.py Normal file
View File

@ -0,0 +1,121 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_configurations.tests import utils as fts_config_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_configurations_utils
class FTSConfigurationDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS configuration
under test schema. """
scenarios = utils.generate_scenarios(
'get_fts_configuration_parsers',
fts_configurations_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_configuration_func_%s" % str(uuid.uuid4())[1:8]
self.fts_configuration_name = "fts_configuration_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Configuration are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_configuration_id = fts_configurations_utils. \
create_fts_configuration(
self.server, self.db_name, self.schema_name,
self.fts_configuration_name)
def _fts_configuration_parses(self):
"""
This functions returns the fts configuration parsers
:return: fts configuration parsers
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def runTest(self):
""" This function will add new FTS configuration under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_configurations_utils.verify_fts_configuration(
self.server, self.db_name, self.fts_configuration_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Configuration.")
if self.is_positive_test:
response = self._fts_configuration_parses()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self._fts_configuration_parses()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_config and disconnect the test
database."""
fts_config_utils.delete_fts_configurations(self.server, self.db_name,
self.schema_name,
self.fts_configuration_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

10
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/test_fts_configuration_put.py
View File

@ -90,6 +90,16 @@ class FTSConfPutTestCase(BaseTestGenerator):
follow_redirects=True)
self.assertEquals(put_response.status_code, 200)
negative_put_response = self.tester.put(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(0),
data=json.dumps(data),
follow_redirects=True)
self.assertEquals(negative_put_response.status_code, 500)
def tearDown(self):
"""This function delete the fts_config and disconnect the test
database."""

123
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/test_fts_configuration_tokens.py Normal file
View File

@ -0,0 +1,123 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_configurations.tests import utils as fts_config_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_configurations_utils
class FTSConfigurationDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS configuration
under test schema. """
scenarios = utils.generate_scenarios(
'get_fts_configuration_get_tokens',
fts_configurations_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_configuration_func_%s" % str(uuid.uuid4())[1:8]
self.fts_configuration_name = "fts_configuration_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Configuration are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_configuration_id = fts_configurations_utils. \
create_fts_configuration(
self.server, self.db_name, self.schema_name,
self.fts_configuration_name
)
def get_fts_configuration_tokens(self):
"""
This functions returns the fts configuration tokens
:return: fts configuration tokens
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/' +
str(self.fts_configuration_id),
content_type='html/json')
def runTest(self):
""" This function will add new FTS configuration under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_configurations_utils.verify_fts_configuration(
self.server, self.db_name, self.fts_configuration_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Configuration.")
if self.is_positive_test:
response = self.get_fts_configuration_tokens()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_fts_configuration_tokens()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_config and disconnect the test
database."""
fts_config_utils.delete_fts_configurations(self.server, self.db_name,
self.schema_name,
self.fts_configuration_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

4
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_configurations/tests/utils.py
View File

@ -12,10 +12,14 @@ from __future__ import print_function
import os
import sys
import traceback
import json
from regression.python_test_utils.test_utils import get_db_connection
file_name = os.path.basename(__file__)
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/fts_configurations_test_data.json") as data_file:
test_cases = json.load(data_file)
def create_fts_configuration(server, db_name, schema_name, fts_conf_name):

143
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_dictionaries/tests/fts_dictionaries_test_data.json Normal file
View File

@ -0,0 +1,143 @@
{
"get_fts_dictionaries_nodes_and_node": [
{
"name": "Get FTS Dictionary nodes",
"url": "/browser/fts_dictionary/nodes/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching FTS Dictionary nodes",
"url": "/browser/fts_dictionary/nodes/",
"error_fetching_fts_dictionaries": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts dictionary nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get FTS Dictionary node",
"url": "/browser/fts_dictionary/nodes/",
"is_positive_test": true,
"node": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching FTS Dictionary node",
"url": "/browser/fts_dictionary/nodes/",
"error_fetching_fts_dictionaries": true,
"is_positive_test": false,
"node": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts dictionary nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while fetching FTS Dictionary node",
"url": "/browser/fts_dictionary/nodes/",
"error_fetching_fts_dictionaries": true,
"set_wrong_fts_configuration_value": true,
"is_positive_test": false,
"node": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(True, {'rows': []})"
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get FTS Dictionary nodes list",
"url": "/browser/fts_dictionary/obj/",
"is_positive_test": true,
"node": false,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Dictionary nodes list",
"url": "/browser/fts_dictionary/obj/",
"is_positive_test": false,
"error_fetching_fts_dictionaries": true,
"node": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts dictionary nodes list')"
},
"expected_data": {
"status_code": 500
}
}
],
"dependency_dependent_fts_dictionaries": [
{
"name": "Get FTS Dictionaries dependency",
"url": "/browser/fts_dictionary/dependency/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get FTS Dictionaries dependent",
"url": "/browser/fts_dictionary/dependent/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"fetch_templates_fts_dictionaries": [
{
"name": "Get FTS Dictionaries templates",
"url": "/browser/fts_dictionary/fetch_templates/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Dictionaries templates",
"url": "/browser/fts_dictionary/fetch_templates/",
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts dictionary template')"
},
"expected_data": {
"status_code": 500
}
}
]
}

10
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_dictionaries/tests/test_fts_dictionaries_delete.py
View File

@ -81,6 +81,16 @@ class FtsDictionaryDeleteTestCase(BaseTestGenerator):
self.assertEquals(delete_response.status_code, 200)
negative_delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(0),
follow_redirects=True)
self.assertEquals(negative_delete_response.status_code, 500)
def tearDown(self):
"""This function disconnect the test database."""

118
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_dictionaries/tests/test_fts_dictionaries_dependency_dependent.py Normal file
View File

@ -0,0 +1,118 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import json
import uuid
from pgadmin.utils import server_utils as server_utils
from regression import trigger_funcs_utils as fts_dict_funcs_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_dictionaries.tests import utils as fts_dict_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as fts_dictionaries_utils
class FTSDictionariesDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS dictionaries
under test schema. """
scenarios = utils.generate_scenarios(
'dependency_dependent_fts_dictionaries',
fts_dictionaries_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_dictionaries_func_%s" % str(uuid.uuid4())[1:8]
self.fts_dictionaries_name = "fts_dictionaries_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Dictionaries are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_dict_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_dictionaries = fts_dictionaries_utils. \
create_fts_dictionary(
self.server, self.db_name, self.schema_name,
self.fts_dictionaries_name)
def runTest(self):
""" This function will add new FTS dictionaries under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_dict_response = fts_dictionaries_utils.verify_fts_dict(
self.server, self.db_name, self.fts_dictionaries_name
)
if not fts_dict_response:
raise Exception("Could not find the FTS Dictionaries.")
if self.is_positive_test:
response = self.get_dependency_dependent()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def get_dependency_dependent(self):
"""
This function returns the fts dictionaries dependency and dependent
:return: fts dictionaries dependency and dependent
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(self.fts_dictionaries),
follow_redirects=True)
def tearDown(self):
"""This function delete the fts_dict and disconnect the test
database."""
fts_dict_utils.delete_fts_dictionaries(self.server, self.db_name,
self.schema_name,
self.fts_dictionaries_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

122
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_dictionaries/tests/test_fts_dictionaries_fetch_templates.py Normal file
View File

@ -0,0 +1,122 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_dictionaries.tests import utils as fts_dict_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_dict_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_dictionaries_utils
class FTSDictionariesDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the templates in FTS dictionaries
under test schema. """
scenarios = utils.generate_scenarios(
'fetch_templates_fts_dictionaries',
fts_dictionaries_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_dictionaries_func_%s" % str(uuid.uuid4())[1:8]
self.fts_dictionaries_name = "fts_dictionaries_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Dictionaries are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_dict_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_dictionaries_id = fts_dictionaries_utils. \
create_fts_dictionary(
self.server, self.db_name, self.schema_name,
self.fts_dictionaries_name)
def get_fts_dictionaries_fetch_templates(self):
"""
This functions returns the fts dictionaries templates
:return: fts dictionaries templates
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def runTest(self):
""" This function will add new FTS dictionaries under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_dictionaries_utils.verify_fts_dict(
self.server, self.db_name, self.fts_dictionaries_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Dictionaries.")
if self.is_positive_test:
response = self.get_fts_dictionaries_fetch_templates()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_fts_dictionaries_fetch_templates()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_dict and disconnect the test
database."""
fts_dict_utils.delete_fts_dictionaries(self.server, self.db_name,
self.schema_name,
self.fts_dictionaries_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

142
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_dictionaries/tests/test_fts_dictionaries_get_nodes.py Normal file
View File

@ -0,0 +1,142 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_dictionaries.tests import utils as fts_dict_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_dict_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_dictionaries_utils
class FTSDictionariesNodesTestCase(BaseTestGenerator):
""" This class will get the nodes/node FTS dictionaries
under test schema. """
scenarios = utils.generate_scenarios(
'get_fts_dictionaries_nodes_and_node',
fts_dictionaries_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_dictionaries_func_%s" % str(uuid.uuid4())[1:8]
self.fts_configuration_name = "fts_dictionaries_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Dictionaries are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_dict_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_dictionaries_id = fts_dictionaries_utils. \
create_fts_dictionary(
self.server, self.db_name, self.schema_name,
self.fts_configuration_name)
def get_fts_dictionaries_nodes(self):
"""
This functions returns the fts dictionaries nodes
:return: fts dictionaries nodes
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def get_fts_dictionaries_node(self):
"""
This functions returns the fts dictionaries node
:return: fts dictionaries node
"""
if hasattr(self, "set_wrong_fts_dictionaries_value"):
self.fts_dictionaries_id = 0
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/' +
str(self.fts_dictionaries_id), content_type='html/json')
def runTest(self):
""" This function will add new FTS dictionaries under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_dictionaries_utils.verify_fts_dict(
self.server, self.db_name, self.fts_configuration_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Dictionaries.")
if self.is_positive_test:
if hasattr(self, "node"):
response = self.get_fts_dictionaries_node()
else:
response = self.get_fts_dictionaries_nodes()
else:
if hasattr(self, "error_fetching_fts_dictionaries"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "node"):
response = self.get_fts_dictionaries_node()
else:
response = self.get_fts_dictionaries_nodes()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_dict and disconnect the test
database."""
fts_dict_utils.delete_fts_dictionaries(self.server, self.db_name,
self.schema_name,
self.fts_configuration_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

11
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_dictionaries/tests/test_fts_dictionaries_put.py
View File

@ -89,6 +89,17 @@ class FtsDictionaryPutTestCase(BaseTestGenerator):
self.assertEquals(put_response.status_code, 200)
negative_put_response = self.tester.put(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(0),
data=json.dumps(data),
follow_redirects=True)
self.assertEquals(negative_put_response.status_code, 500)
def tearDown(self):
"""This function delete the fts dictionaries and disconnect the test
database."""

4
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_dictionaries/tests/utils.py
View File

@ -12,10 +12,14 @@ from __future__ import print_function
import os
import sys
import traceback
import json
from regression.python_test_utils.test_utils import get_db_connection
file_name = os.path.basename(__file__)
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/fts_dictionaries_test_data.json") as data_file:
test_cases = json.load(data_file)
def create_fts_dictionary(server, db_name, schema_name, fts_dict_name):

248
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/fts_parsers_test_data.json Normal file
View File

@ -0,0 +1,248 @@
{
"get_fts_parsers_nodes_and_node": [
{
"name": "Get FTS Parsers nodes",
"url": "/browser/fts_parser/nodes/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching FTS Parsers nodes",
"url": "/browser/fts_parser/nodes/",
"error_fetching_fts_parser": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts Parsers nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get FTS Dictionary node",
"url": "/browser/fts_parser/nodes/",
"is_positive_test": true,
"node": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching FTS Parsers node",
"url": "/browser/fts_parser/nodes/",
"error_fetching_fts_parser": true,
"is_positive_test": false,
"node": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts Parsers nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while fetching FTS Parsers node",
"url": "/browser/fts_parser/nodes/",
"error_fetching_fts_parser": true,
"set_wrong_fts_parsers_value": true,
"is_positive_test": false,
"node": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(True, {'rows': []})"
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get FTS Parsers nodes list",
"url": "/browser/fts_parser/obj/",
"is_positive_test": true,
"node": false,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Parsers nodes list",
"url": "/browser/fts_parser/obj/",
"is_positive_test": false,
"error_fetching_fts_parser": true,
"node": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts Parsers nodes list')"
},
"expected_data": {
"status_code": 500
}
}
],
"dependency_dependent_fts_parsers": [
{
"name": "Get FTS Parsers dependency",
"url": "/browser/fts_parser/dependency/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get FTS Parsers dependent",
"url": "/browser/fts_parser/dependent/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"head_line_functions_fts_parsers": [
{
"name": "Get FTS Parsers headline function",
"url": "/browser/fts_parser/headline_functions/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Parsers headline function",
"url": "/browser/fts_parser/headline_functions/",
"is_positive_test": false,
"error_fetching_fts_parser": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts parser headline functions')"
},
"expected_data": {
"status_code": 500
}
}
],
"lextype_functions_fts_parsers": [
{
"name": "Get FTS Parsers lectype function",
"url": "/browser/fts_parser/lextype_functions/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Parsers lextype function",
"url": "/browser/fts_parser/lextype_functions/",
"is_positive_test": false,
"error_fetching_fts_parser": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts parser lextype functions')"
},
"expected_data": {
"status_code": 500
}
}
],
"end_functions_fts_parsers": [
{
"name": "Get FTS Parsers end function",
"url": "/browser/fts_parser/end_functions/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Parsers end function",
"url": "/browser/fts_parser/end_functions/",
"is_positive_test": false,
"error_fetching_fts_parser": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts parser end functions')"
},
"expected_data": {
"status_code": 500
}
}
],
"token_functions_fts_parsers": [
{
"name": "Get FTS Parsers token function",
"url": "/browser/fts_parser/token_functions/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Parsers token function",
"url": "/browser/fts_parser/token_functions/",
"is_positive_test": false,
"error_fetching_fts_parser": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts parser token functions')"
},
"expected_data": {
"status_code": 500
}
}
],
"start_functions_fts_parsers": [
{
"name": "Get FTS Parsers start function",
"url": "/browser/fts_parser/start_functions/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Parsers start function",
"url": "/browser/fts_parser/start_functions/",
"is_positive_test": false,
"error_fetching_fts_parser": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts parser start functions')"
},
"expected_data": {
"status_code": 500
}
}
]
}

10
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/test_fts_parser_delete.py
View File

@ -79,6 +79,16 @@ class FtsParserDeleteTestCase(BaseTestGenerator):
self.assertEquals(delete_response.status_code, 200)
negative_delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(0),
follow_redirects=True)
self.assertEquals(negative_delete_response.status_code, 500)
def tearDown(self):
"""This function disconnect the test database."""

118
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/test_fts_parser_dependency_dependent.py Normal file
View File

@ -0,0 +1,118 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import json
import uuid
from pgadmin.utils import server_utils as server_utils
from regression import trigger_funcs_utils as fts_parser_funcs_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_parsers.tests import utils as fts_parser_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as fts_parsers_utils
class FTSParsersDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS parsers
under test schema. """
scenarios = utils.generate_scenarios(
'dependency_dependent_fts_parsers',
fts_parsers_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_parsers_func_%s" % str(uuid.uuid4())[1:8]
self.fts_parsers_name = "fts_parsers_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Parsers are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_parser_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_parsers = fts_parsers_utils. \
create_fts_parser(
self.server, self.db_name, self.schema_name,
self.fts_parsers_name)
def runTest(self):
""" This function will add new FTS dictionaries under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_dict_response = fts_parsers_utils.verify_fts_parser(
self.server, self.db_name, self.fts_parsers_name
)
if not fts_dict_response:
raise Exception("Could not find the FTS Parsers.")
if self.is_positive_test:
response = self.get_dependency_dependent()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def get_dependency_dependent(self):
"""
This function returns the fts parsers dependency and dependent
:return: fts parsers dependency and dependent
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(self.fts_parsers),
follow_redirects=True)
def tearDown(self):
"""This function delete the fts_parser and disconnect the test
database."""
fts_parser_utils.delete_fts_parser(self.server, self.db_name,
self.schema_name,
self.fts_parsers_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

123
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/test_fts_parser_end_functions.py Normal file
View File

@ -0,0 +1,123 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_parsers.tests import utils as fts_parser_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_parsers_utils
class FTSParsersEndFunctionTestCase(BaseTestGenerator):
""" This class will get the end functions FTS parsers
under test schema. """
scenarios = utils.generate_scenarios(
'end_functions_fts_parsers',
fts_parsers_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_parser_func_%s" % str(uuid.uuid4())[1:8]
self.fts_parser_name = "fts_parser_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Parsers are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_parser_id = fts_parsers_utils. \
create_fts_parser(
self.server, self.db_name, self.schema_name,
self.fts_parser_name)
def get_fts_parsers_end_functions(self):
"""
This functions returns the fts parsers end functions
:return: fts end functions
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def runTest(self):
""" This function will add new FTS parsers under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_parsers_utils.verify_fts_parser(
self.server, self.db_name, self.fts_parser_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Parsers.")
if self.is_positive_test:
response = self.get_fts_parsers_end_functions()
else:
if hasattr(self, "error_fetching_fts_parser"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_fts_parsers_end_functions()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_parser and disconnect the test
database."""
fts_parser_utils.delete_fts_parser(self.server, self.db_name,
self.schema_name,
self.fts_parser_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

142
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/test_fts_parser_get_nodes.py Normal file
View File

@ -0,0 +1,142 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_parsers.tests import utils as fts_parser_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_parsers_utils
class FTSParsersNodesTestCase(BaseTestGenerator):
""" This class will get the nodes/node FTS parsers
under test schema. """
scenarios = utils.generate_scenarios(
'get_fts_parsers_nodes_and_node',
fts_parsers_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_parser_func_%s" % str(uuid.uuid4())[1:8]
self.fts_parser_name = "fts_parser_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Parsers are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_parser_id = fts_parsers_utils. \
create_fts_parser(
self.server, self.db_name, self.schema_name,
self.fts_parser_name)
def get_fts_parsers_nodes(self):
"""
This functions returns the fts parsers nodes
:return: fts parsers nodes
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def get_fts_parsers_node(self):
"""
This functions returns the fts parsers node
:return: fts parsers node
"""
if hasattr(self, "set_wrong_fts_parsers_value"):
self.fts_parser_id = 0
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/' +
str(self.fts_parser_id), content_type='html/json')
def runTest(self):
""" This function will add new FTS parsers under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_parsers_utils.verify_fts_parser(
self.server, self.db_name, self.fts_parser_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Parsers.")
if self.is_positive_test:
if hasattr(self, "node"):
response = self.get_fts_parsers_node()
else:
response = self.get_fts_parsers_nodes()
else:
if hasattr(self, "error_fetching_fts_parser"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "node"):
response = self.get_fts_parsers_node()
else:
response = self.get_fts_parsers_nodes()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_parser and disconnect the test
database."""
fts_parser_utils.delete_fts_parser(self.server, self.db_name,
self.schema_name,
self.fts_parser_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

123
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/test_fts_parser_head_line_functions.py Normal file
View File

@ -0,0 +1,123 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_parsers.tests import utils as fts_parser_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_parsers_utils
class FTSParsersNodesTestCase(BaseTestGenerator):
""" This class will get the headline functions FTS parsers
under test schema. """
scenarios = utils.generate_scenarios(
'head_line_functions_fts_parsers',
fts_parsers_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_parser_func_%s" % str(uuid.uuid4())[1:8]
self.fts_parser_name = "fts_parser_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Parsers are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_parser_id = fts_parsers_utils. \
create_fts_parser(
self.server, self.db_name, self.schema_name,
self.fts_parser_name)
def get_fts_parsers_headline_functions(self):
"""
This functions returns the fts parsers headline functions
:return: fts headline functions node
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def runTest(self):
""" This function will add new FTS parsers under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_parsers_utils.verify_fts_parser(
self.server, self.db_name, self.fts_parser_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Parsers.")
if self.is_positive_test:
response = self.get_fts_parsers_headline_functions()
else:
if hasattr(self, "error_fetching_fts_parser"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_fts_parsers_headline_functions()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_parser and disconnect the test
database."""
fts_parser_utils.delete_fts_parser(self.server, self.db_name,
self.schema_name,
self.fts_parser_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

123
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/test_fts_parser_lextype_functions.py Normal file
View File

@ -0,0 +1,123 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_parsers.tests import utils as fts_parser_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_parsers_utils
class FTSParsersLexTypeTestCase(BaseTestGenerator):
""" This class will get the lextype functions FTS parsers
under test schema. """
scenarios = utils.generate_scenarios(
'lextype_functions_fts_parsers',
fts_parsers_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_parser_func_%s" % str(uuid.uuid4())[1:8]
self.fts_parser_name = "fts_parser_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Parsers are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_parser_id = fts_parsers_utils. \
create_fts_parser(
self.server, self.db_name, self.schema_name,
self.fts_parser_name)
def get_fts_parsers_lextype_functions(self):
"""
This functions returns the fts parsers lextype functions
:return: fts lextype functions node
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def runTest(self):
""" This function will add new FTS parsers under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_parsers_utils.verify_fts_parser(
self.server, self.db_name, self.fts_parser_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Parsers.")
if self.is_positive_test:
response = self.get_fts_parsers_lextype_functions()
else:
if hasattr(self, "error_fetching_fts_parser"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_fts_parsers_lextype_functions()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_parser and disconnect the test
database."""
fts_parser_utils.delete_fts_parser(self.server, self.db_name,
self.schema_name,
self.fts_parser_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

123
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/test_fts_parser_start_functions.py Normal file
View File

@ -0,0 +1,123 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_parsers.tests import utils as fts_parser_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_parsers_utils
class FTSParsersEndFunctionTestCase(BaseTestGenerator):
""" This class will get the start functions FTS parsers
under test schema. """
scenarios = utils.generate_scenarios(
'start_functions_fts_parsers',
fts_parsers_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_parser_func_%s" % str(uuid.uuid4())[1:8]
self.fts_parser_name = "fts_parser_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Parsers are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_parser_id = fts_parsers_utils. \
create_fts_parser(
self.server, self.db_name, self.schema_name,
self.fts_parser_name)
def get_fts_parsers_start_functions(self):
"""
This functions returns the fts parsers start functions
:return: fts start functions node
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def runTest(self):
""" This function will add new FTS parsers under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_parsers_utils.verify_fts_parser(
self.server, self.db_name, self.fts_parser_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Parsers.")
if self.is_positive_test:
response = self.get_fts_parsers_start_functions()
else:
if hasattr(self, "error_fetching_fts_parser"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_fts_parsers_start_functions()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_parser and disconnect the test
database."""
fts_parser_utils.delete_fts_parser(self.server, self.db_name,
self.schema_name,
self.fts_parser_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

123
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/test_fts_parser_token_functions.py Normal file
View File

@ -0,0 +1,123 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_parsers.tests import utils as fts_parser_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_config_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_parsers_utils
class FTSParsersTokenFunctionTestCase(BaseTestGenerator):
""" This class will get the token functions FTS parsers
under test schema. """
scenarios = utils.generate_scenarios(
'token_functions_fts_parsers',
fts_parsers_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_parser_func_%s" % str(uuid.uuid4())[1:8]
self.fts_parser_name = "fts_parser_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Parsers are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_config_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_parser_id = fts_parsers_utils. \
create_fts_parser(
self.server, self.db_name, self.schema_name,
self.fts_parser_name)
def get_fts_parsers_token_functions(self):
"""
This functions returns the fts parsers token functions
:return: fts token functions node
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def runTest(self):
""" This function will add new FTS parsers under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_conf_response = fts_parsers_utils.verify_fts_parser(
self.server, self.db_name, self.fts_parser_name
)
if not fts_conf_response:
raise Exception("Could not find the FTS Parsers.")
if self.is_positive_test:
response = self.get_fts_parsers_token_functions()
else:
if hasattr(self, "error_fetching_fts_parser"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_fts_parsers_token_functions()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def tearDown(self):
"""This function delete the fts_parser and disconnect the test
database."""
fts_parser_utils.delete_fts_parser(self.server, self.db_name,
self.schema_name,
self.fts_parser_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

4
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_parsers/tests/utils.py
View File

@ -12,10 +12,14 @@ from __future__ import print_function
import os
import sys
import traceback
import json
from regression.python_test_utils.test_utils import get_db_connection
file_name = os.path.basename(__file__)
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/fts_parsers_test_data.json") as data_file:
test_cases = json.load(data_file)
def create_fts_parser(server, db_name, schema_name, fts_parser_name):

170
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_templates/tests/fts_templates_test_data.json Normal file
View File

@ -0,0 +1,170 @@
{
"get_fts_template_nodes_and_node": [
{
"name": "Get FTS Templates nodes",
"url": "/browser/fts_template/nodes/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching FTS Templates nodes",
"url": "/browser/fts_template/nodes/",
"error_fetching_fts_template": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts templates nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get FTS Templates node",
"url": "/browser/fts_template/nodes/",
"is_positive_test": true,
"node": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching FTS Templates node",
"url": "/browser/fts_template/nodes/",
"error_fetching_fts_template": true,
"is_positive_test": false,
"node": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts templates nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while fetching FTS Templates node",
"url": "/browser/fts_template/nodes/",
"error_fetching_fts_template": true,
"set_wrong_fts_templates_value": true,
"is_positive_test": false,
"node": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(True, {'rows': []})"
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get FTS Templates nodes list",
"url": "/browser/fts_template/obj/",
"is_positive_test": true,
"node": false,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Templates nodes list",
"url": "/browser/fts_template/obj/",
"is_positive_test": false,
"error_fetching_fts_template": true,
"node": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts templates nodes list')"
},
"expected_data": {
"status_code": 500
}
}
],
"dependency_dependent_fts_templates": [
{
"name": "Get FTS Templates dependency",
"url": "/browser/fts_template/dependency/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get FTS Templates dependent",
"url": "/browser/fts_template/dependent/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"get_init_fts_templates": [
{
"name": "Get FTS Templates get init",
"url": "/browser/fts_template/get_init/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Templates get init",
"url": "/browser/fts_template/get_init/",
"is_positive_test": false,
"error_fetching_fts_template": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts template init functions')"
},
"expected_data": {
"status_code": 500
}
}
],
"get_lexize_fts_templates": [
{
"name": "Get FTS Templates get lexize",
"url": "/browser/fts_template/get_lexize/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while Get FTS Templates get lexize",
"url": "/browser/fts_template/get_lexize/",
"is_positive_test": false,
"error_fetching_fts_template": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching fts template lexize functions')"
},
"expected_data": {
"status_code": 500
}
}
]
}

118
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_templates/tests/test_fts_template_dependency_dependent.py Normal file
View File

@ -0,0 +1,118 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import json
import uuid
from pgadmin.utils import server_utils as server_utils
from regression import trigger_funcs_utils as fts_template_funcs_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_templates.tests import utils as fts_template_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as fts_templates_utils
class FTSTemplatesDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the dependency and dependents FTS templates
under test schema. """
scenarios = utils.generate_scenarios(
'dependency_dependent_fts_templates',
fts_templates_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_template_func_%s" % str(uuid.uuid4())[1:8]
self.fts_templates_name = "fts_template_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Templates are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_template_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_templates = fts_templates_utils. \
create_fts_template(
self.server, self.db_name, self.schema_name,
self.fts_templates_name)
def runTest(self):
""" This function will add new FTS templates under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_dict_response = fts_templates_utils.verify_fts_template(
self.server, self.db_name, self.fts_templates_name
)
if not fts_dict_response:
raise Exception("Could not find the FTS Templates.")
if self.is_positive_test:
response = self.get_dependency_dependent()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def get_dependency_dependent(self):
"""
This function returns the fts templates dependency and dependent
:return: fts templates dependency and dependent
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(self.fts_templates),
follow_redirects=True)
def tearDown(self):
"""This function delete the fts_templates and disconnect the test
database."""
fts_template_utils.delete_fts_template(self.server, self.db_name,
self.schema_name,
self.fts_templates_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

125
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_templates/tests/test_fts_template_get_init.py Normal file
View File

@ -0,0 +1,125 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_templates.tests import utils as fts_template_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_template_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_templates_utils
class FTSTemplatesGetInitTestCase(BaseTestGenerator):
""" This class will get the init FTS templates
under test schema. """
scenarios = utils.generate_scenarios(
'get_init_fts_templates',
fts_templates_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_template_func_%s" % str(uuid.uuid4())[1:8]
self.fts_templates_name = "fts_template_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Templates are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_template_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_templates_id = fts_templates_utils. \
create_fts_template(
self.server, self.db_name, self.schema_name,
self.fts_templates_name)
def runTest(self):
""" This function will add new FTS templates under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_dict_response = fts_templates_utils.verify_fts_template(
self.server, self.db_name, self.fts_templates_name
)
if not fts_dict_response:
raise Exception("Could not find the FTS Templates.")
if self.is_positive_test:
response = self.get_init()
else:
if hasattr(self, "error_fetching_fts_template"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_init()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def get_init(self):
"""
This function returns the fts templates init
:return: fts templates init
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/',
follow_redirects=True)
def tearDown(self):
"""This function delete the fts_templates and disconnect the test
database."""
fts_template_utils.delete_fts_template(self.server, self.db_name,
self.schema_name,
self.fts_templates_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

125
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_templates/tests/test_fts_template_get_lexize.py Normal file
View File

@ -0,0 +1,125 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_templates.tests import utils as fts_template_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_template_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_templates_utils
class FTSTemplatesGetLexizeTestCase(BaseTestGenerator):
""" This class will get the lexize FTS templates
under test schema. """
scenarios = utils.generate_scenarios(
'get_lexize_fts_templates',
fts_templates_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_template_func_%s" % str(uuid.uuid4())[1:8]
self.fts_templates_name = "fts_template_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "FTS Templates are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_template_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_templates_id = fts_templates_utils. \
create_fts_template(
self.server, self.db_name, self.schema_name,
self.fts_templates_name)
def runTest(self):
""" This function will add new FTS templates under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_dict_response = fts_templates_utils.verify_fts_template(
self.server, self.db_name, self.fts_templates_name
)
if not fts_dict_response:
raise Exception("Could not find the FTS Templates.")
if self.is_positive_test:
response = self.get_lexzie()
else:
if hasattr(self, "error_fetching_fts_template"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_lexzie()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def get_lexzie(self):
"""
This function returns the fts templates lexize
:return: fts templates lexize
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/',
follow_redirects=True)
def tearDown(self):
"""This function delete the fts_templates and disconnect the test
database."""
fts_template_utils.delete_fts_template(self.server, self.db_name,
self.schema_name,
self.fts_templates_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

143
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_templates/tests/test_fts_template_get_nodes.py Normal file
View File

@ -0,0 +1,143 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.schemas \
.fts_templates.tests import utils as fts_template_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tests import \
utils as schema_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils import server_utils as server_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression import trigger_funcs_utils as fts_template_funcs_utils
from regression.python_test_utils import test_utils as utils
from . import utils as fts_templates_utils
class FTSTemplatesDependencyDependentTestCase(BaseTestGenerator):
""" This class will get the nodes/node FTS templates
under test schema. """
scenarios = utils.generate_scenarios(
'get_fts_template_nodes_and_node',
fts_templates_utils.test_cases
)
def setUp(self):
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
self.extension_name = "postgres_fdw"
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.db_user = self.server["username"]
self.func_name = "fts_template_func_%s" % str(uuid.uuid4())[1:8]
self.fts_templates_name = "fts_template_delete_%s" % (
str(uuid.uuid4())[1:8])
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
server_version = 0
if "type" in server_con["data"]:
if server_con["data"]["version"] < 90500:
message = "Event triggers are not supported by PG9.4 " \
"and PPAS9.4 and below."
self.skipTest(message)
self.function_info = fts_template_funcs_utils.create_trigger_function(
self.server, self.db_name, self.schema_name, self.func_name,
server_version)
self.fts_templates_id = fts_templates_utils. \
create_fts_template(
self.server, self.db_name, self.schema_name,
self.fts_templates_name)
def runTest(self):
""" This function will add new FTS templates under test schema. """
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.")
schema_response = schema_utils.verify_schemas(self.server,
self.db_name,
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema.")
fts_dict_response = fts_templates_utils.verify_fts_template(
self.server, self.db_name, self.fts_templates_name
)
if not fts_dict_response:
raise Exception("Could not find the FTS Templates.")
if self.is_positive_test:
if hasattr(self, "node"):
response = self.get_fts_templates_node()
else:
response = self.get_fts_templates_nodes()
else:
if hasattr(self, "error_fetching_fts_template"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "node"):
response = self.get_fts_templates_node()
else:
response = self.get_fts_templates_nodes()
actual_response_code = response.status_code
expected_response_code = self.expected_data['status_code']
self.assertEquals(actual_response_code, expected_response_code)
def get_fts_templates_nodes(self):
"""
This functions returns the fts templates nodes
:return: fts templates nodes
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/',
content_type='html/json')
def get_fts_templates_node(self):
"""
This functions returns the fts templates node
:return: fts templates node
"""
if hasattr(self, "set_wrong_fts_templates_value"):
self.fts_templates_id = 0
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.schema_id) + '/' +
str(self.fts_templates_id), content_type='html/json')
def tearDown(self):
"""This function delete the fts_templates and disconnect the test
database."""
fts_template_utils.delete_fts_template(self.server, self.db_name,
self.schema_name,
self.fts_templates_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

10
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_templates/tests/test_fts_templates_delete.py
View File

@ -78,6 +78,16 @@ class FtsTemplateDeleteTestCase(BaseTestGenerator):
self.assertEquals(delete_response.status_code, 200)
negative_delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(0),
follow_redirects=True)
self.assertEquals(negative_delete_response.status_code, 500)
def tearDown(self):
"""This function disconnect the test database."""

11
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_templates/tests/test_fts_templates_put.py
View File

@ -87,6 +87,17 @@ class FtsTemplatePutTestCase(BaseTestGenerator):
self.assertEquals(put_response.status_code, 200)
negative_put_response = self.tester.put(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(0),
data=json.dumps(data),
follow_redirects=True)
self.assertEquals(negative_put_response.status_code, 500)
def tearDown(self):
"""This function delete the fts_template and disconnect the test
database."""

4
web/pgadmin/browser/server_groups/servers/databases/schemas/fts_templates/tests/utils.py
View File

@ -12,10 +12,14 @@ from __future__ import print_function
import os
import sys
import traceback
import json
from regression.python_test_utils.test_utils import get_db_connection
file_name = os.path.basename(__file__)
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/fts_templates_test_data.json") as data_file:
test_cases = json.load(data_file)
def create_fts_template(server, db_name, schema_name, fts_temp_name):