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 Triggers and Compound Triggers. Fixes #5335

Browse Source
This commit is contained in:
Pradip Parkale 2020-05-15 12:50:50 +05:30 committed by Akshay Joshi
parent c0451cd454
commit 6d6cd19b37
15 changed files with 2004 additions and 247 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/en_US/release_notes_4_22.rst
View File

@ -16,6 +16,7 @@ Housekeeping
************
| `Issue #5255 <https://redmine.postgresql.org/issues/5255>`_ - Implement Selenium Grid to run multiple tests across different browsers, operating systems, and machines in parallel.
| `Issue #5334 <https://redmine.postgresql.org/issues/5334>`_ - Improve code coverage and API test cases for the Rules module.
| `Issue #5335 <https://redmine.postgresql.org/issues/5335>`_ - Improve code coverage and API test cases for Triggers and Compound Triggers.
| `Issue #5443 <https://redmine.postgresql.org/issues/5443>`_ - Remove support for Python 2.
| `Issue #5444 <https://redmine.postgresql.org/issues/5444>`_ - Cleanup Python detection in the runtime project file.
| `Issue #5455 <https://redmine.postgresql.org/issues/5455>`_ - Refactor pgAdmin4.py so it can be imported and is a lot more readable.

800
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/compound_triggers/tests/compound_trigger_test_data.json Normal file
View File

@ -0,0 +1,800 @@
{
"add_compound_trigger": [
{
"name": "Create compound compound trigger for all events",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\n BEFORE STATEMENT IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var);\n var := 'BEFORE STATEMENT';\nEND;",
"evnt_insert": true,
"evnt_update": true,
"evnt_delete": true,
"evnt_truncate": true
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Create compound compound trigger for insert and delete",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\nBEFORE STATEMENT IS\n BEGIN \n DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var);\n var := 'BEFORE STATEMENT';\nEND;\n\nBEFORE EACH ROW IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before each row: ' || var);\n var := 'BEFORE EACH ROW';\nEND;",
"evnt_insert": true,
"evnt_update": false,
"evnt_delete": true
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Create compound compound trigger for insert with when condition",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\n BEFORE EACH ROW IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before each row: ' || var);\n var := 'EACH ROW';\nEND;",
"evnt_insert": true,
"evnt_update": false,
"evnt_delete": false,
"whenclause": "NEW.id < 100"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Create compound compound trigger for insert or update on columns",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\n BEFORE STATEMENT IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var);\n var := 'BEFORE STATEMENT';\nEND;\n\nBEFORE EACH ROW IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before each row: ' || var);\n var := 'BEFORE EACH ROW';\nEND;",
"evnt_insert": true,
"evnt_update": true,
"columns": [
"id",
"name"
]
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Create compound compound trigger for truncate",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\n BEFORE STATEMENT IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var);\n var := 'BEFORE STATEMENT';\nEND;",
"evnt_truncate": true
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Create compound trigger for insert delete and update on view",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\n BEFORE STATEMENT IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var);\n var := 'BEFORE STATEMENT';\nEND;",
"evnt_insert": true,
"evnt_update": true,
"evnt_delete": true,
"evnt_truncate": false
},
"on_view": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Create compound trigger for instead of each row",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\n INSTEAD OF EACH ROW IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Instead of: ' || var);\n var := 'INSTEAD OF EACH ROW';\nEND;",
"evnt_insert": true,
"evnt_update": true,
"evnt_delete": true,
"evnt_truncate": false
},
"on_view": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while adding a compound trigger using wrong table id",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": false,
"wrong_table_id": true,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\n BEFORE STATEMENT IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var);\n var := 'BEFORE STATEMENT';\nEND;",
"evnt_insert": true,
"evnt_update": true,
"evnt_delete": true,
"evnt_truncate": true
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error while adding a compound trigger",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": false,
"test_data": {
"name": "PLACE_HOLDER",
"is_row_compound_trigger": true,
"fires": "BEFORE",
"columns": [],
"tfunction": "PLACE_HOLDER",
"evnt_insert": true
},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error ')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while while fetching the compound trigger id using compound trigger name",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": false,
"internal_server_error": true,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\n BEFORE STATEMENT IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var);\n var := 'BEFORE STATEMENT';\nEND;",
"evnt_insert": true,
"evnt_update": true,
"evnt_delete": true,
"evnt_truncate": true
},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(True, True),(False, 'Mocked Internal Server Error ')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Exception while adding a compound trigger",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": false,
"test_data": {
"prosrc": "var varchar2(20) := 'Global_var';\n\n BEFORE STATEMENT IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var);\n var := 'BEFORE STATEMENT';\nEND;",
"evnt_insert": true,
"evnt_update": true,
"evnt_delete": true,
"evnt_truncate": true
},
"mocking_required": true,
"error_creating_compound_trigger": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error ')"
},
"expected_data": {
"status_code": 500
}
}
],
"get_compound_trigger": [
{
"name": "Get a selected compound trigger URL",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get a selected compound trigger URL using wrong compound trigger id",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"incorrect_compound_trigger_id": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error while fetching a selected compound trigger node properties",
"url": "/browser/compound_trigger/obj/",
"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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get a properties of all compound triggers under table nodes",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"table_nodes": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a compound trigger properties under table nodes",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": false,
"table_nodes": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get a compound trigger Node",
"url": "/browser/compound_trigger/nodes/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get a compound trigger Node using wrong compound trigger id",
"url": "/browser/compound_trigger/nodes/",
"is_positive_test": true,
"incorrect_compound_trigger_id": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get a compound trigger Node dependants",
"url": "/browser/compound_trigger/dependent/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get a compound trigger Node dependency",
"url": "/browser/compound_trigger/dependency/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching the compound trigger under the table nodes using wrong table id",
"url": "/browser/compound_trigger/nodes/",
"is_positive_test": false,
"mocking_required": true,
"dummy_dict": {
"rows": [
{
"schema": "test_schema_88886",
"table": "table_compound_trigger_94da770"
}
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500,
"message": "Internal Server Error"
}
},
{
"name": "Get all the compound trigger under the table nodes",
"url": "/browser/compound_trigger/nodes/",
"is_positive_test": true,
"mocking_required": false,
"table_nodes": true,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get all the compound trigger under the table nodes using wrong table id",
"url": "/browser/compound_trigger/nodes/",
"is_positive_test": true,
"incorrect_table_id": true,
"table_nodes": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching all the compound trigger under the table nodes using wrong table id",
"url": "/browser/compound_trigger/nodes/",
"is_positive_test": false,
"table_nodes": true,
"mocking_required": true,
"dummy_data": false,
"dummy_dict": {
"rows": [
{
"schema": "test_schema_88886",
"table": "table_compound_trigger_94da770"
}
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500,
"message": "Server error"
}
},
{
"name": "Get a compound trigger SQL using wrong compound trigger id",
"url": "/browser/compound_trigger/sql/",
"is_positive_test": true,
"incorrect_compound_trigger_id": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 500
}
},
{
"name": "Fetch MSQL of compound trigger",
"url": "/browser/compound_trigger/msql/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Fetch msql of compound trigger using wrong compound trigger id",
"url": "/browser/compound_trigger/msql/",
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.browser.server_groups.servers.databases.schemas.tables.compound_triggers.utils.get_sql",
"return_value": "('', 'Mocked response')"
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while generating msql of compound trigger",
"url": "/browser/compound_trigger/msql/",
"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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get the modified SQL by passing an argument",
"url": "/browser/compound_trigger/msql/",
"is_positive_test": true,
"mocking_required": false,
"pass_argument": true,
"mock_data": {
},
"expected_data": {
"status_code": 200
}
}
],
"delete_compound_trigger": [
{
"name": "Delete a compound trigger URL",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a compound trigger to delete",
"url": "/browser/compound_trigger/obj/",
"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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while deleting the compound trigger",
"url": "/browser/compound_trigger/obj/",
"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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "compound trigger not found while deleting a compound trigger",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"invalid_compound_trigger_id": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"update_compound_trigger": [
{
"name": "Exception while fetching a compound trigger to update",
"url": "/browser/compound_trigger/obj/",
"new_compound_trigger_id": true,
"is_positive_test": false,
"test_data": {
"description": "This is test comment."
},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.browser.server_groups.servers.databases.schemas.tables.compound_triggers.utils.get_sql",
"return_value": "(False, 'Mocked response')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while getting sql of Updated trigger",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": false,
"mocking_required": true,
"dummy_data": false,
"test_data": {
"description": "This is test comment."
},
"dummy_dict": {
"rows": [
{
"oid": 60299,
"name": "test_trigger_update_0680105",
"xmin": "11787",
"is_enable_trigger": "O",
"tgrelid": 60289,
"tgname": "test_trigger_update_0680105",
"tgfoid": 60298,
"tgtype": 7,
"tgenabled": "O",
"tgisinternal": false,
"tgconstrrelid": 0,
"tgconstrindid": 0,
"tgconstraint": 0,
"tgdeferrable": false,
"tginitdeferred": false,
"tgnargs": 0,
"tgattr": "",
"tgargs": "",
"tgqual": "",
"tgoldtable": "",
"tgnewtable": "",
"relname": "table_trigger_2cadf23",
"parentistable": true,
"nspname": "test_schema_b0a5c",
"description": "",
"lanname": "plpgsql",
"prosrc": " BEGIN NULL; END; ",
"tfunction": "test_schema_b0a5c.trigger_func_add_3d79e4e",
"whenclause": "",
"custom_tgargs": [],
"is_sys_trigger": false,
"is_constraint_trigger": false,
"tgoldtable-2": "",
"tgnewtable-2": ""
}
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "PlACE_HOLDER"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Updated compound trigger not found",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": false,
"mocking_required": true,
"updated_trigger": true,
"dummy_data": {
"rows": []
},
"test_data": {
"description": "This is test comment."
},
"dummy_dict": {
"rows": [
{
"oid": 60299,
"name": "test_trigger_update_0680105",
"xmin": "11787",
"is_enable_trigger": "O",
"tgrelid": 60289,
"tgname": "test_trigger_update_0680105",
"tgfoid": 60298,
"tgtype": 7,
"tgenabled": "O",
"tgisinternal": false,
"tgconstrrelid": 0,
"tgconstrindid": 0,
"tgconstraint": 0,
"tgdeferrable": false,
"tginitdeferred": false,
"tgnargs": 0,
"tgattr": "",
"tgargs": "",
"tgqual": "",
"tgoldtable": "",
"tgnewtable": "",
"relname": "table_trigger_2cadf23",
"parentistable": true,
"nspname": "test_schema_b0a5c",
"description": "",
"lanname": "plpgsql",
"prosrc": " BEGIN NULL; END; ",
"tfunction": "test_schema_b0a5c.trigger_func_add_3d79e4e",
"whenclause": "",
"custom_tgargs": [],
"is_sys_trigger": false,
"is_constraint_trigger": false,
"tgoldtable-2": "",
"tgnewtable-2": ""
}
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "PlACE_HOLDER"
},
"expected_data": {
"status_code": 500,
"message": "Error while fetching a trigger to update"
}
},
{
"name": "update a compound trigger URL",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"test_data": {
"description": "This is test comment."
},
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Update event and code",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"test_data": {
"evnt_update": true,
"prosrc": "var varchar2(20) := 'Global_var';\n\n AFTER STATEMENT IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('After Statement: ' || var);\n var := 'AFTER STATEMENT';\nEND;\n\nAFTER EACH ROW IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('After each row: ' || var);\n var := 'AFTER EACH ROW';\nEND;"
},
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Enable Replica compound compound trigger",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"test_data": {
"is_enable_trigger": "A"
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Disable compound compound trigger",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"test_data": {
"is_enable_trigger": "D"
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Enable a compound trigger",
"url": "/browser/compound_trigger/enable/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"test_data": {
"is_enable_trigger": "O"
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Enable a compound trigger using wrong compound trigger id",
"url": "/browser/compound_trigger/enable/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"wrong_compound_trigger_id": true,
"test_data": {
"is_enable_trigger": "O"
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error while fetching a compound trigger to enable/disable",
"url": "/browser/compound_trigger/enable/",
"is_positive_test": false,
"mocking_required": true,
"test_data": {
"is_enable_trigger": "O"
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while enabling a compound trigger",
"url": "/browser/compound_trigger/enable/",
"is_positive_test": false,
"mocking_required": true,
"test_data": {
"is_enable_trigger": "O"
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Exception while enabling a compound trigger",
"url": "/browser/compound_trigger/enable/",
"is_positive_test": false,
"mocking_required": true,
"dummy_data": false,
"test_data": {
"is_enable_trigger": "O"
},
"dummy_dict": {
"rows": [
1
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "PLACE_HOLDER"
},
"expected_data": {
"status_code": 500,
"message": ""
}
},
{
"name": "Error while updating the compound trigger",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": false,
"mocking_required": true,
"test_data": {
"description": "This is test comment."
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while updating the new compound trigger id in browser tree",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": false,
"mocking_required": true,
"new_compound_trigger_id": true,
"test_data": {
"description": "This is test comment."
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(True, True),(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
}
],
"delete_multiple_compound_trigger": [
{
"name": "Delete multiple compound trigger",
"url": "/browser/compound_trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
]
}

152
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/compound_triggers/tests/test_compound_triggers_add.py
View File

@ -22,114 +22,21 @@ from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.databases.schemas.views.tests \
import utils as view_utils
from . import utils as compound_trigger_utils
import sys
if sys.version_info < (3, 3):
from mock import patch
else:
from unittest.mock import patch
class CompoundTriggersAddTestCase(BaseTestGenerator):
"""This class will add new compound trigger under table node."""
skip_on_database = ['gpdb']
scenarios = [
('Create compound trigger for all events',
dict(
url='/browser/compound_trigger/obj/',
data={
"prosrc": "var varchar2(20) := 'Global_var';\n\n"
"BEFORE STATEMENT IS\nBEGIN\n "
"DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var)"
";\n var := 'BEFORE STATEMENT';\nEND;",
"evnt_insert": True,
"evnt_update": True,
"evnt_delete": True,
"evnt_truncate": True
}
)),
('Create compound trigger for insert and delete',
dict(
url='/browser/compound_trigger/obj/',
data={
"prosrc": "var varchar2(20) := 'Global_var';\n\n"
"BEFORE STATEMENT IS\nBEGIN\n "
"DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var)"
";\n var := 'BEFORE STATEMENT';\nEND;\n\nBEFORE "
"EACH ROW IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('"
"Before each row: ' || var);\n var := 'BEFORE "
"EACH ROW';\nEND;",
"evnt_insert": True,
"evnt_update": False,
"evnt_delete": True
}
)),
('Create compound trigger for insert with when condition',
dict(
url='/browser/compound_trigger/obj/',
data={
"prosrc": "var varchar2(20) := 'Global_var';\n\n"
"BEFORE EACH ROW IS\nBEGIN\n "
"DBMS_OUTPUT.PUT_LINE('Before each row: ' || var)"
";\n var := 'EACH ROW';\nEND;",
"evnt_insert": True,
"evnt_update": False,
"evnt_delete": False,
"whenclause": "NEW.id < 100"
}
)),
('Create compound trigger for insert or update on columns',
dict(
url='/browser/compound_trigger/obj/',
data={
"prosrc": "var varchar2(20) := 'Global_var';\n\n"
"BEFORE STATEMENT IS\nBEGIN\n "
"DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var)"
";\n var := 'BEFORE STATEMENT';\nEND;\n\nBEFORE "
"EACH ROW IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('"
"Before each row: ' || var);\n var := 'BEFORE "
"EACH ROW';\nEND;",
"evnt_insert": True,
"evnt_update": True,
"columns": ["id", "name"]
}
)),
('Create compound trigger for truncate',
dict(
url='/browser/compound_trigger/obj/',
data={
"prosrc": "var varchar2(20) := 'Global_var';\n\n"
"BEFORE STATEMENT IS\nBEGIN\n "
"DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var)"
";\n var := 'BEFORE STATEMENT';\nEND;",
"evnt_truncate": True
}
)),
('Create compound trigger for insert delete and update on view',
dict(
url='/browser/compound_trigger/obj/',
data={
"prosrc": "var varchar2(20) := 'Global_var';\n\n"
"BEFORE STATEMENT IS\nBEGIN\n "
"DBMS_OUTPUT.PUT_LINE('Before Statement: ' || var)"
";\n var := 'BEFORE STATEMENT';\nEND;",
"evnt_insert": True,
"evnt_update": True,
"evnt_delete": True,
"evnt_truncate": False
},
on_view=True
)),
('Create compound trigger for instead of each row',
dict(
url='/browser/compound_trigger/obj/',
data={
"prosrc": "var varchar2(20) := 'Global_var';\n\n"
"INSTEAD OF EACH ROW IS\nBEGIN\n "
"DBMS_OUTPUT.PUT_LINE('Instead of: ' || var)"
";\n var := 'INSTEAD OF EACH ROW';\nEND;",
"evnt_insert": True,
"evnt_update": True,
"evnt_delete": True,
"evnt_truncate": False
},
on_view=True
)),
]
scenarios = utils.generate_scenarios('add_compound_trigger',
compound_trigger_utils.test_cases)
def setUp(self):
super(CompoundTriggersAddTestCase, self).setUp()
@ -175,25 +82,46 @@ class CompoundTriggersAddTestCase(BaseTestGenerator):
view_sql,
self.view_name)
def create_compound_trigger(self, object_id):
return self.tester.post(
"{0}{1}/{2}/{3}/{4}/{5}/".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, object_id),
data=json.dumps(self.test_data),
content_type='html/json'
)
def runTest(self):
"""This function will create compound trigger under table node."""
trigger_name = \
"test_compound_trigger_add_%s" % (str(uuid.uuid4())[1:8])
self.data.update({"name": trigger_name})
self.test_data.update({"name": trigger_name})
object_id = self.table_id
if hasattr(self, 'on_view'):
object_id = self.view_id
response = self.tester.post(
"{0}{1}/{2}/{3}/{4}/{5}/".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, object_id),
data=json.dumps(self.data),
content_type='html/json'
)
self.assertEquals(response.status_code, 200)
if self.is_positive_test:
response = self.create_compound_trigger(object_id)
else:
if hasattr(self, 'wrong_table_id'):
del self.test_data["name"]
response = self.create_compound_trigger(object_id)
elif hasattr(self, 'internal_server_error'):
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = self.create_compound_trigger(object_id)
elif hasattr(self, 'error_creating_compound_trigger'):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.create_compound_trigger(object_id)
else:
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = self.create_compound_trigger(object_id)
self.assertEquals(response.status_code,
self.expected_data["status_code"])
def tearDown(self):
# Disconnect the database

40
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/compound_triggers/tests/test_compound_triggers_delete.py
View File

@ -20,15 +20,19 @@ 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 compound_triggers_utils
import sys
if sys.version_info < (3, 3):
from mock import patch
else:
from unittest.mock import patch
class CompoundTriggersDeleteTestCase(BaseTestGenerator):
"""This class will delete compound trigger under table node."""
skip_on_database = ['gpdb']
scenarios = [
('Delete compound trigger',
dict(url='/browser/compound_trigger/obj/'))
]
scenarios = utils.generate_scenarios('delete_compound_trigger',
compound_triggers_utils.test_cases)
def setUp(self):
super(CompoundTriggersDeleteTestCase, self).setUp()
@ -76,6 +80,15 @@ class CompoundTriggersDeleteTestCase(BaseTestGenerator):
self.table_name,
self.trigger_name)
def delete_compound_trigger(self):
return self.tester.delete(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
follow_redirects=True
)
def runTest(self):
"""This function will update trigger under table node."""
trigger_response = \
@ -84,14 +97,17 @@ class CompoundTriggersDeleteTestCase(BaseTestGenerator):
self.trigger_name)
if not trigger_response:
raise Exception("Could not find the compound trigger to delete.")
response = self.tester.delete(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
follow_redirects=True
)
self.assertEquals(response.status_code, 200)
if self.is_positive_test:
if hasattr(self, "invalid_compound_trigger_id"):
self.trigger_id = 9999
response = self.delete_compound_trigger()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.delete_compound_trigger()
self.assertEquals(response.status_code,
self.expected_data["status_code"])
def tearDown(self):
# Disconnect the database

38
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/compound_triggers/tests/test_compound_triggers_delete_multiple.py
View File

@ -21,15 +21,19 @@ 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 compound_triggers_utils
import sys
if sys.version_info < (3, 3):
from mock import patch
else:
from unittest.mock import patch
class CompoundTriggersDeleteMultipleTestCase(BaseTestGenerator):
"""This class will delete multiple compound triggers under table node."""
skip_on_database = ['gpdb']
scenarios = [
('Delete multiple compound trigger',
dict(url='/browser/compound_trigger/obj/'))
]
scenarios = utils.generate_scenarios('delete_multiple_compound_trigger',
compound_triggers_utils.test_cases)
def setUp(self):
super(CompoundTriggersDeleteMultipleTestCase, self).setUp()
@ -80,6 +84,17 @@ class CompoundTriggersDeleteMultipleTestCase(BaseTestGenerator):
self.trigger_name_1)
]
def delete_multiple_compound_triggers(self, data):
return self.tester.delete(
"{0}{1}/{2}/{3}/{4}/{5}/".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id
),
follow_redirects=True,
data=json.dumps(data),
content_type='html/json'
)
def runTest(self):
"""This function will delete multiple compound trigger under
table node."""
@ -96,16 +111,11 @@ class CompoundTriggersDeleteMultipleTestCase(BaseTestGenerator):
raise Exception("Could not find the compound trigger to delete.")
data = {'ids': self.trigger_ids}
response = self.tester.delete(
"{0}{1}/{2}/{3}/{4}/{5}/".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id
),
follow_redirects=True,
data=json.dumps(data),
content_type='html/json'
)
self.assertEquals(response.status_code, 200)
if self.is_positive_test:
response = self.delete_multiple_compound_triggers(data)
self.assertEquals(response.status_code,
self.expected_data['status_code'])
def tearDown(self):
# Disconnect the database

149
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/compound_triggers/tests/test_compound_triggers_get.py Normal file
View File

@ -0,0 +1,149 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import json
import uuid
from pgadmin.utils import server_utils as server_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tables.tests \
import utils as tables_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.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as compound_triggers_utils
import sys
if sys.version_info < (3, 3):
from mock import patch
else:
from unittest.mock import patch
class CompoundTriggersGetTestCase(BaseTestGenerator):
"""This class will get the compound trigger under table node."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('get_compound_trigger',
compound_triggers_utils.test_cases)
def setUp(self):
super(CompoundTriggersGetTestCase, self).setUp()
self.db_name = parent_node_dict["database"][-1]["db_name"]
schema_info = parent_node_dict["schema"][-1]
self.server_id = schema_info["server_id"]
self.db_id = schema_info["db_id"]
server_con = server_utils.connect_server(self, self.server_id)
if server_con:
if "type" in server_con["data"]:
if server_con["data"]["type"] == "pg":
message = "Compound Triggers are not supported by PG."
self.skipTest(message)
elif server_con["data"]["type"] == "ppas" and \
server_con["data"]["version"] < 120000:
message = "Compound Triggers are not supported by " \
"EPAS server less than 12"
self.skipTest(message)
db_con = database_utils.connect_database(self, utils.SERVER_GROUP,
self.server_id, self.db_id)
if not db_con['data']["connected"]:
raise Exception(
"Could not connect to database to update a compound trigger.")
self.schema_id = schema_info["schema_id"]
self.schema_name = schema_info["schema_name"]
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 to update a trigger.")
self.table_name = \
"table_compound_trigger_%s" % (str(uuid.uuid4())[1:8])
self.table_id = tables_utils.create_table(self.server, self.db_name,
self.schema_name,
self.table_name)
self.trigger_name = \
"test_compound_trigger_update_%s" % (str(uuid.uuid4())[1:8])
self.trigger_id = \
compound_triggers_utils.create_compound_trigger(self.server,
self.db_name,
self.schema_name,
self.table_name,
self.trigger_name)
def get_compound_trigger(self):
return self.tester.get(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
follow_redirects=True
)
def runTest(self):
"""This function will get trigger under table node."""
trigger_response = \
compound_triggers_utils.verify_compound_trigger(self.server,
self.db_name,
self.trigger_name)
if not trigger_response:
raise Exception("Could not find the compound trigger to update.")
if self.is_positive_test:
if hasattr(self, "incorrect_compound_trigger_id"):
self.trigger_id = 9999
if hasattr(self, "pass_argument"):
url = "{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url,
utils.SERVER_GROUP,
self.server_id,
self.db_id,
self.schema_id,
self.table_id,
self.trigger_id)
url = \
url + \
"?oid=17312&description=commaa&name=code&_=1589522392579"
response = self.tester.get(url, follow_redirects=True)
self.assertEquals(response.status_code,
self.expected_data["status_code"])
if hasattr(self, "table_nodes"):
self.trigger_id = ''
response = self.get_compound_trigger()
else:
response = self.get_compound_trigger()
else:
if hasattr(self, "dummy_dict"):
if hasattr(self, "table_nodes"):
self.trigger_id = ''
self.mock_data['return_value'] = [(True, self.dummy_dict), (
False, self.expected_data["message"])]
with patch(self.mock_data["function_name"],
side_effect=self.mock_data["return_value"]):
response = self.get_compound_trigger()
elif hasattr(self, "table_nodes"):
self.trigger_id = ''
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_compound_trigger()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_compound_trigger()
self.assertEquals(response.status_code,
self.expected_data["status_code"])
def tearDown(self):
# Disconnect the database
database_utils.disconnect_database(self, self.server_id, self.db_id)

97
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/compound_triggers/tests/test_compound_triggers_put.py
View File

@ -22,45 +22,19 @@ from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as compound_triggers_utils
import sys
if sys.version_info < (3, 3):
from mock import patch
else:
from unittest.mock import patch
class CompoundTriggersUpdateTestCase(BaseTestGenerator):
"""This class will update compound trigger under table node."""
skip_on_database = ['gpdb']
scenarios = [
('Update comment',
dict(url='/browser/compound_trigger/obj/',
data={"description": "This is test comment."}
)),
('Update event and code',
dict(url='/browser/compound_trigger/obj/',
data={
"evnt_update": True,
"prosrc": "var varchar2(20) := 'Global_var';\n\n"
"AFTER STATEMENT IS\nBEGIN\n "
"DBMS_OUTPUT.PUT_LINE('After Statement: ' || var)"
";\n var := 'AFTER STATEMENT';\nEND;\n\nAFTER "
"EACH ROW IS\nBEGIN\n DBMS_OUTPUT.PUT_LINE('"
"After each row: ' || var);\n var := 'AFTER "
"EACH ROW';\nEND;",
})),
('Enable compound trigger',
dict(url='/browser/compound_trigger/obj/',
data={"is_enable_trigger": 'O'},
disable_trigger=True
)),
('Enable always compound trigger',
dict(url='/browser/compound_trigger/obj/',
data={"is_enable_trigger": 'A'}
)),
('Enable replica compound trigger',
dict(url='/browser/compound_trigger/obj/',
data={"is_enable_trigger": 'R'}
)),
('Disable compound trigger',
dict(url='/browser/compound_trigger/obj/',
data={"is_enable_trigger": 'D'}
)),
]
scenarios = utils.generate_scenarios('update_compound_trigger',
compound_triggers_utils.test_cases)
def setUp(self):
super(CompoundTriggersUpdateTestCase, self).setUp()
@ -107,8 +81,18 @@ class CompoundTriggersUpdateTestCase(BaseTestGenerator):
self.table_name,
self.trigger_name)
def update_compound_trigger(self):
return self.tester.put(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
data=json.dumps(self.test_data),
follow_redirects=True
)
def runTest(self):
"""This function will update trigger under table node."""
"""This function will get trigger under table node."""
trigger_response = \
compound_triggers_utils.verify_compound_trigger(self.server,
self.db_name,
@ -126,16 +110,37 @@ class CompoundTriggersUpdateTestCase(BaseTestGenerator):
False
)
self.data.update({"id": self.trigger_id})
response = self.tester.put(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
data=json.dumps(self.data),
follow_redirects=True
)
self.assertEquals(response.status_code, 200)
self.test_data.update({"id": self.trigger_id})
if self.is_positive_test:
if hasattr(self, "wrong_compound_trigger_id"):
self.trigger_id = 9999
response = self.update_compound_trigger()
elif hasattr(self, "new_compound_trigger_id"):
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = self.update_compound_trigger()
elif hasattr(self, "dummy_dict"):
self.mock_data['return_value'] = [(True, self.dummy_dict), (
False, self.dummy_data)]
with patch(self.mock_data["function_name"],
side_effect=self.mock_data["return_value"]), patch(
'pgadmin.utils.driver.psycopg2.connection.Connection.'
'execute_scalar',
side_effect=[(True, True),
(True, "Mocked response")]):
response = self.update_compound_trigger()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "wrong_compound_trigger_id"):
self.trigger_id = 9999
response = self.update_compound_trigger()
self.assertEquals(response.status_code,
self.expected_data["status_code"])
def tearDown(self):
# Disconnect the database

8
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/compound_triggers/tests/utils.py
View File

@ -11,9 +11,15 @@ from __future__ import print_function
import sys
import traceback
import os
import json
from regression.python_test_utils import test_utils as utils
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/compound_trigger_test_data.json") as data_file:
test_cases = json.load(data_file)
def create_compound_trigger(server, db_name, schema_name, table_name,
trigger_name):
@ -50,7 +56,7 @@ def create_compound_trigger(server, db_name, schema_name, table_name,
old_isolation_level = connection.isolation_level
connection.set_isolation_level(0)
pg_cursor = connection.cursor()
query = "CREATE OR REPLACE TRIGGER %s FOR INSERT ON %s.%s " \
query = "CREATE OR REPLACE TRIGGER %s FOR INSERT OR UPDATE ON %s.%s " \
"COMPOUND TRIGGER %s END;" % (trigger_name, schema_name,
table_name, code)
pg_cursor.execute(query)

57
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/triggers/tests/test_triggers_add.py
View File

@ -21,14 +21,20 @@ from pgadmin.browser.server_groups.servers.databases.tests import utils as \
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 triggers_utils
import sys
if sys.version_info < (3, 3):
from mock import patch
else:
from unittest.mock import patch
class TriggersAddTestCase(BaseTestGenerator):
"""This class will add new trigger under table node."""
skip_on_database = ['gpdb']
scenarios = [
('Add trigger Node URL', dict(url='/browser/trigger/obj/'))
]
scenarios = utils.generate_scenarios('add_trigger',
triggers_utils.test_cases)
def setUp(self):
super(TriggersAddTestCase, self).setUp()
@ -56,25 +62,42 @@ class TriggersAddTestCase(BaseTestGenerator):
trigger_funcs_utils.create_trigger_function_with_trigger(
self.server, self.db_name, self.schema_name, self.func_name)
def runTest(self):
"""This function will trigger under table node."""
trigger_name = "test_trigger_add_%s" % (str(uuid.uuid4())[1:8])
data = {"name": trigger_name,
"is_row_trigger": True,
"fires": "BEFORE",
"columns": [],
"tfunction": "{0}.{1}".format(self.schema_name,
self.func_name),
"evnt_insert": True
}
response = self.tester.post(
def create_trigger(self):
return self.tester.post(
"{0}{1}/{2}/{3}/{4}/{5}/".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id),
data=json.dumps(data),
data=json.dumps(self.test_data),
content_type='html/json'
)
self.assertEquals(response.status_code, 200)
def runTest(self):
"""This function will trigger under table node."""
trigger_name = "test_trigger_add_%s" % (str(uuid.uuid4())[1:8])
self.test_data['name'] = trigger_name
self.test_data['tfunction'] = "{0}.{1}".format(self.schema_name,
self.func_name)
if self.is_positive_test:
response = self.create_trigger()
else:
if hasattr(self, 'wrong_table_id'):
del self.test_data["name"]
response = self.create_trigger()
elif hasattr(self, 'internal_server_error'):
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = self.create_trigger()
elif hasattr(self, 'error_creating_rule'):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.create_trigger()
else:
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = self.create_trigger()
self.assertEquals(response.status_code,
self.expected_data["status_code"])
def tearDown(self):
# Disconnect the database

40
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/triggers/tests/test_triggers_delete.py
View File

@ -21,14 +21,19 @@ 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 triggers_utils
import sys
if sys.version_info < (3, 3):
from mock import patch
else:
from unittest.mock import patch
class TriggersDeleteTestCase(BaseTestGenerator):
"""This class will delete trigger under table node."""
skip_on_database = ['gpdb']
scenarios = [
('Delete trigger Node URL', dict(url='/browser/trigger/obj/'))
]
scenarios = utils.generate_scenarios('delete_trigger',
triggers_utils.test_cases)
def setUp(self):
super(TriggersDeleteTestCase, self).setUp()
@ -63,6 +68,15 @@ class TriggersDeleteTestCase(BaseTestGenerator):
self.trigger_name,
self.func_name)
def delete_trigger(self):
return self.tester.delete(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
follow_redirects=True
)
def runTest(self):
"""This function will delete trigger under table node."""
trigger_response = triggers_utils.verify_trigger(self.server,
@ -70,14 +84,18 @@ class TriggersDeleteTestCase(BaseTestGenerator):
self.trigger_name)
if not trigger_response:
raise Exception("Could not find the trigger to delete.")
response = self.tester.delete(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
follow_redirects=True
)
self.assertEquals(response.status_code, 200)
if self.is_positive_test:
if hasattr(self, "invalid_trigger_id"):
self.trigger_id = 9999
response = self.delete_trigger()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.delete_trigger()
self.assertEquals(response.status_code,
self.expected_data["status_code"])
def tearDown(self):
# Disconnect the database

5
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/triggers/tests/test_triggers_delete_multiple.py
View File

@ -27,9 +27,8 @@ from . import utils as triggers_utils
class TriggersDeleteMultipleTestCase(BaseTestGenerator):
"""This class will delete trigger under table node."""
skip_on_database = ['gpdb']
scenarios = [
('Delete multiple triggers', dict(url='/browser/trigger/obj/'))
]
scenarios = utils.generate_scenarios('delete_multiple_trigger',
triggers_utils.test_cases)
def setUp(self):
super(TriggersDeleteMultipleTestCase, self).setUp()

66
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/triggers/tests/test_triggers_get.py
View File

@ -21,14 +21,19 @@ 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 triggers_utils
import sys
if sys.version_info < (3, 3):
from mock import patch
else:
from unittest.mock import patch
class TriggersGetTestCase(BaseTestGenerator):
"""This class will fetch trigger under table node."""
skip_on_database = ['gpdb']
scenarios = [
('Fetch trigger Node URL', dict(url='/browser/trigger/obj/'))
]
scenarios = utils.generate_scenarios('get_trigger',
triggers_utils.test_cases)
def setUp(self):
super(TriggersGetTestCase, self).setUp()
@ -63,16 +68,63 @@ class TriggersGetTestCase(BaseTestGenerator):
self.trigger_name,
self.func_name)
def runTest(self):
"""This function will fetch trigger under table node."""
response = self.tester.get(
def get_trigger(self):
return self.tester.get(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
follow_redirects=True
)
self.assertEquals(response.status_code, 200)
def runTest(self):
"""This function will fetch trigger under table node."""
if self.is_positive_test:
if hasattr(self, "incorrect_trigger_id"):
self.trigger_id = 9999
if hasattr(self, "pass_argument"):
url = "{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url,
utils.SERVER_GROUP,
self.server_id,
self.db_id,
self.schema_id,
self.table_id,
self.trigger_id)
url = \
url + \
"?oid=17312&description=commaa&name=code&_=1589522392579"
response = self.tester.get(url, follow_redirects=True)
self.assertEquals(response.status_code,
self.expected_data["status_code"])
if hasattr(self, "table_nodes"):
self.trigger_id = ''
response = self.get_trigger()
else:
response = self.get_trigger()
else:
if hasattr(self, "dummy_dict"):
if hasattr(self, "table_nodes"):
self.trigger_id = ''
self.mock_data['return_value'] = [(True, self.dummy_dict), (
False, self.expected_data["message"])]
with patch(self.mock_data["function_name"],
side_effect=self.mock_data["return_value"]):
response = self.get_trigger()
elif hasattr(self, "table_nodes"):
self.trigger_id = ''
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_trigger()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
response = self.get_trigger()
self.assertEquals(response.status_code,
self.expected_data["status_code"])
def tearDown(self):
# Disconnect the database

72
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/triggers/tests/test_triggers_put.py
View File

@ -22,25 +22,19 @@ 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 triggers_utils
import sys
if sys.version_info < (3, 3):
from mock import patch
else:
from unittest.mock import patch
class TriggersUpdateTestCase(BaseTestGenerator):
"""This class will update trigger under table node."""
skip_on_database = ['gpdb']
scenarios = [
('Put trigger Node URL', dict(
url='/browser/trigger/obj/',
data={"description": "This is test comment."})),
('Enable Always compound trigger',
dict(url='/browser/trigger/obj/',
data={"is_enable_trigger": 'A'})),
('Enable Replica compound trigger',
dict(url='/browser/trigger/obj/',
data={"is_enable_trigger": 'R'})),
('Disable compound trigger',
dict(url='/browser/trigger/obj/',
data={"is_enable_trigger": 'D'})),
]
scenarios = utils.generate_scenarios('update_trigger',
triggers_utils.test_cases)
def setUp(self):
super(TriggersUpdateTestCase, self).setUp()
@ -76,6 +70,20 @@ class TriggersUpdateTestCase(BaseTestGenerator):
self.trigger_name,
self.func_name)
def update_trigger(self):
"""
This function update the trigger under table node
:return: response of updated trigger
"""
return self.tester.put(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
data=json.dumps(self.test_data),
follow_redirects=True
)
def runTest(self):
"""This function will update trigger under table node."""
trigger_response = triggers_utils.verify_trigger(self.server,
@ -84,16 +92,32 @@ class TriggersUpdateTestCase(BaseTestGenerator):
if not trigger_response:
raise Exception("Could not find the trigger to update.")
self.data.update({"id": self.trigger_id})
response = self.tester.put(
"{0}{1}/{2}/{3}/{4}/{5}/{6}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, self.table_id,
self.trigger_id),
data=json.dumps(self.data),
follow_redirects=True
)
self.assertEquals(response.status_code, 200)
self.test_data.update({"id": self.trigger_id})
if self.is_positive_test:
if hasattr(self, "wrong_trigger_id"):
self.trigger_id = 9999
response = self.update_trigger()
elif hasattr(self, "new_trigger_id"):
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = self.update_trigger()
elif hasattr(self, "dummy_dict"):
self.mock_data['return_value'] = [(True, self.dummy_dict), (
False, self.expected_data["message"])]
with patch(self.mock_data["function_name"],
side_effect=self.mock_data["return_value"]):
response = self.update_trigger()
else:
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "wrong_trigger_id"):
self.trigger_id = 9999
response = self.update_trigger()
self.assertEquals(response.status_code,
self.expected_data["status_code"])
def tearDown(self):
# Disconnect the database

720
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/triggers/tests/triggers_test_data.json Normal file
View File

@ -0,0 +1,720 @@
{
"add_trigger": [
{
"name": "Add trigger Node",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"test_data": {
"name": "PLACE_HOLDER",
"is_row_trigger": true,
"fires": "BEFORE",
"columns": [],
"tfunction": "PLACE_HOLDER",
"evnt_insert": true
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Create after fires event trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"test_data": {
"name": "PLACE_HOLDER",
"is_row_trigger": true,
"fires": "AFTER",
"columns": [],
"tfunction": "PLACE_HOLDER",
"evnt_update": true
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Check if it throws an error when trigger is created with 'fires after event' and truncate",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"test_data": {
"name": "PLACE_HOLDER",
"is_row_trigger": true,
"fires": "AFTER",
"columns": [],
"tfunction": "PLACE_HOLDER",
"evnt_truncate": true
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 500
}
},
{
"name": "Check if it throws an error when no event is selected while creating a trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"test_data": {
"name": "PLACE_HOLDER",
"is_row_trigger": true,
"fires": "AFTER",
"columns": [],
"tfunction": "PLACE_HOLDER"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while adding a trigger using wrong table",
"url": "/browser/trigger/obj/",
"is_positive_test": false,
"wrong_table_id": true,
"test_data": {
"name": "PLACE_HOLDER",
"is_row_trigger": true,
"fires": "BEFORE",
"columns": [],
"tfunction": "PLACE_HOLDER",
"evnt_insert": true
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error while adding a trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": false,
"error_creating_trigger": true,
"test_data": {
"name": "PLACE_HOLDER",
"is_row_trigger": true,
"fires": "BEFORE",
"columns": [],
"tfunction": "PLACE_HOLDER",
"evnt_insert": true
},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error ')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while while fetching the trigger id using trigger name",
"url": "/browser/trigger/obj/",
"is_positive_test": false,
"internal_server_error": true,
"test_data": {
"name": "PLACE_HOLDER",
"is_row_trigger": true,
"fires": "BEFORE",
"columns": [],
"tfunction": "PLACE_HOLDER",
"evnt_insert": true
},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(True, True),(False, 'Mocked Internal Server Error ')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Exception while adding a trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": false,
"test_data": {
"name": "PLACE_HOLDER",
"is_row_trigger": true,
"fires": "BEFORE",
"columns": [],
"tfunction": "PLACE_HOLDER",
"evnt_insert": true
},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error ')"
},
"expected_data": {
"status_code": 500
}
}
],
"get_trigger": [
{
"name": "Get a trigger URL",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get a trigger URL using wrong trigger id",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"incorrect_trigger_id": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get a trigger properties under table nodes",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"table_nodes": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a trigger properties under table nodes",
"url": "/browser/trigger/obj/",
"is_positive_test": false,
"table_nodes": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get a trigger Node",
"url": "/browser/trigger/nodes/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get a trigger Node using wrong trigger id",
"url": "/browser/trigger/nodes/",
"is_positive_test": true,
"incorrect_trigger_id": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get a trigger Node dependants",
"url": "/browser/trigger/dependent/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get a trigger Node dependency",
"url": "/browser/trigger/dependency/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching the trigger under the table nodes using wrong table id",
"url": "/browser/trigger/nodes/",
"is_positive_test": false,
"mocking_required": true,
"dummy_dict": {
"rows": [
{
"schema": "test_schema_88886",
"table": "table_trigger_94da770"
}
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500,
"message": "Internal Server Error"
}
},
{
"name": "Get all the trigger under the table nodes",
"url": "/browser/trigger/nodes/",
"is_positive_test": true,
"mocking_required": false,
"table_nodes": true,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get all the trigger under the table nodes using wrong table id",
"url": "/browser/trigger/nodes/",
"is_positive_test": true,
"incorrect_table_id": true,
"table_nodes": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching all the trigger under the table nodes using wrong table id",
"url": "/browser/trigger/nodes/",
"is_positive_test": false,
"table_nodes": true,
"mocking_required": true,
"dummy_dict": {
"rows": [
{
"schema": "test_schema_88886",
"table": "table_trigger_94da770"
}
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500,
"message": "Server error"
}
},
{
"name": "Get a trigger SQL using wrong trigger id",
"url": "/browser/trigger/sql/",
"is_positive_test": true,
"incorrect_trigger_id": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Fetch MSQL of trigger",
"url": "/browser/trigger/msql/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Fetch msql of trigger using wrong trigger id",
"url": "/browser/trigger/msql/",
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.browser.server_groups.servers.databases.schemas.tables.triggers.utils.get_sql",
"return_value": "('', 'Mocked response')"
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while generating msql of trigger",
"url": "/browser/trigger/msql/",
"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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get the modified SQL by passing an argument",
"url": "/browser/trigger/msql/",
"is_positive_test": true,
"mocking_required": false,
"pass_argument": true,
"mock_data": {
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get a list of available trigger function",
"url": "/browser/trigger/get_triggerfunctions/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a list of available trigger function",
"url": "/browser/trigger/get_triggerfunctions/",
"is_positive_test": false,
"mocking_required": true,
"dummy_dict": {
"rows": [
{
"schema": "test_schema_88886",
"table": "table_trigger_94da770"
}
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(True, True),(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500,
"message": "Server error"
}
},
{
"name": "Exception while fetching a list of available trigger function",
"url": "/browser/trigger/get_triggerfunctions/",
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(True, True),(True, True)"
},
"expected_data": {
"status_code": 500,
"message": "Server error"
}
}
],
"delete_trigger": [
{
"name": "Delete a trigger URL",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a trigger to delete",
"url": "/browser/trigger/obj/",
"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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while deleting the trigger",
"url": "/browser/trigger/obj/",
"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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while fetching a trigger to delete",
"url": "/browser/trigger/obj/",
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(True, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "trigger not found while deleting a trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"invalid_trigger_id": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"update_trigger": [
{
"name": "update a trigger URL",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"test_data": {
"description": "This is test comment."
},
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Enable Always compound trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"test_data": {
"is_enable_trigger": "A"
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Enable Replica compound trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"test_data": {
"is_enable_trigger": "R"
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Disable compound trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"test_data": {
"is_enable_trigger": "D"
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Enable a trigger",
"url": "/browser/trigger/enable/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"test_data": {
"is_enable_trigger": "O"
},
"expected_data": {
"status_code": 200
}
},
{
"name": "Enable a trigger using wrong trigger id",
"url": "/browser/trigger/enable/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"wrong_trigger_id": true,
"test_data": {
"is_enable_trigger": "O"
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error while fetching a trigger to enable/disable",
"url": "/browser/trigger/enable/",
"is_positive_test": false,
"mocking_required": true,
"test_data": {
"is_enable_trigger": "O"
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while enabling a trigger",
"url": "/browser/trigger/enable/",
"is_positive_test": false,
"mocking_required": true,
"test_data": {
"is_enable_trigger": "O"
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Exception while enabling a trigger",
"url": "/browser/trigger/enable/",
"is_positive_test": false,
"mocking_required": true,
"test_data": {
"is_enable_trigger": "O"
},
"dummy_dict": {
"rows": [
1
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "PLACE_HOLDER"
},
"expected_data": {
"status_code": 500,
"message": ""
}
},
{
"name": "Error while fetching a trigger to update",
"url": "/browser/trigger/obj/",
"is_positive_test": false,
"mocking_required": true,
"test_data": {
"description": "This is test comment."
},
"dummy_dict": {
"rows": [
{
"oid": 60299,
"name": "test_trigger_update_0680105",
"xmin": "11787",
"is_enable_trigger": "O",
"tgrelid": 60289,
"tgname": "test_trigger_update_0680105",
"tgfoid": 60298,
"tgtype": 7,
"tgenabled": "O",
"tgisinternal": false,
"tgconstrrelid": 0,
"tgconstrindid": 0,
"tgconstraint": 0,
"tgdeferrable": false,
"tginitdeferred": false,
"tgnargs": 0,
"tgattr": "",
"tgargs": "",
"tgqual": "",
"tgoldtable": "",
"tgnewtable": "",
"relname": "table_trigger_2cadf23",
"parentistable": true,
"nspname": "test_schema_b0a5c",
"description": "",
"lanname": "plpgsql",
"prosrc": " BEGIN NULL; END; ",
"tfunction": "test_schema_b0a5c.trigger_func_add_3d79e4e",
"whenclause": "",
"custom_tgargs": [],
"is_sys_trigger": false,
"is_constraint_trigger": false,
"tgoldtable-2": "",
"tgnewtable-2": ""
}
]
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500,
"message": "Error while fetching a trigger to update"
}
},
{
"name": "Error while updating the trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": false,
"mocking_required": true,
"test_data": {
"description": "This is test comment."
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while updating the new trigger id in browser tree",
"url": "/browser/trigger/obj/",
"is_positive_test": false,
"mocking_required": true,
"new_trigger_id": true,
"test_data": {
"description": "This is test comment."
},
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(True, True),(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
}
],
"delete_multiple_trigger": [
{
"name": "Delete multiple trigger",
"url": "/browser/trigger/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
]
}

6
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/triggers/tests/utils.py
View File

@ -11,9 +11,15 @@ from __future__ import print_function
import sys
import traceback
import os
import json
from regression.python_test_utils import test_utils as utils
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/triggers_test_data.json") as data_file:
test_cases = json.load(data_file)
def create_trigger(server, db_name, schema_name, table_name, trigger_name,
trigger_func_name):