IntenseWebs/pgadmin4
3
0
Fork 0
mirror of https://github.com/pgadmin-org/pgadmin4.git synced 2024-12-23 07:34:35 -06:00
Code Issues Packages Projects Releases Wiki Activity

Improve code coverage and API test cases for pgAgent. Fixes #5338

Browse Source
This commit is contained in:
Yogesh Mahajan 2020-12-31 13:52:14 +05:30 committed by Akshay Joshi
parent 6fce58f2ad
commit d1f5511566
39 changed files with 3797 additions and 782 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/en_US/release_notes_4_30.rst
View File

@ -13,6 +13,7 @@ New features
Housekeeping
************
| `Issue #5338 <https://redmine.postgresql.org/issues/5338>`_ - Improve code coverage and API test cases for pgAgent.
| `Issue #6052 <https://redmine.postgresql.org/issues/6052>`_ - Added connected pgAdmin user and connection name in the log file.
| `Issue #6079 <https://redmine.postgresql.org/issues/6079>`_ - Updated mimetype from 'text/javascript' to 'application/javascript' as 'text/javascript' is obsolete.

0
web/pgadmin/browser/server_groups/servers/pgagent/schedules/tests/__init__.py Normal file
View File

666
web/pgadmin/browser/server_groups/servers/pgagent/schedules/tests/schedules_test_data.json Normal file
View File

@ -0,0 +1,666 @@
{
"pgagent_create_schedule": [
{
"name": "Create schedule: With exception in pgAgent job.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jscjobid": "jscjobid",
"jscenabled": true,
"jscdesc": "",
"jscname": "pgagent_schedule_name",
"jscexceptions": [
{
"jexdate": "2050-01-01",
"jextime": "14:00"
}
],
"jscstart": "2050-01-01 12:14:21 +05:30",
"jscend": "2050-03-01 12:14:21 +05:30",
"jscminutes": [
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false
],
"jschours": [
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true
],
"jscweekdays": [
false,
false,
false,
false,
false,
false,
false
],
"jscmonthdays": [
false,
false,
true,
false,
true,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false
],
"jscmonths": [
false,
false,
true,
true,
false,
false,
false,
false,
false,
false,
false,
false
]
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Create schedule: With existing pgAgent job while server is down.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": false,
"inventory_data": {},
"test_data": {
"jscjobid": "jscjobid",
"jscenabled": true,
"jscdesc": "",
"jscname": "pgagent_schedule_name",
"jscstart": "2050-01-01 12:14:21 +05:30",
"jscend": "2050-03-01 12:14:21 +05:30",
"jscminutes": [
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false
],
"jschours": [
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true,
true
],
"jscweekdays": [
false,
false,
false,
false,
false,
false,
false
],
"jscmonthdays": [
false,
false,
true,
false,
true,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false
],
"jscmonths": [
false,
false,
true,
true,
false,
false,
false,
false,
false,
false,
false,
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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
}
}
],
"pgagent_delete_schedule": [
{
"name": "Delete schedule: With existing pgAgent job.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Delete multiple schedules: With existing pgAgent job.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": true
}
],
"pgagent_put_schedule": [
{
"name": "Update schedule: With start and end time.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jscdesc": "Test Schedule",
"jscstart": "2050-01-01 12:00:00 +05:30",
"jscend": "2050-01-20 12:00:00 +05:30"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Update schedule: With repeat.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jscmonthdays": [
true,
false,
true,
false,
true,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false
],
"jscweekdays": [
true,
false,
false,
true,
false,
false,
false
],
"jscmonths": [
true,
false,
false,
true,
false,
false,
false,
false,
false,
false,
false,
false
],
"jschours": [
false,
false,
false,
false,
true,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false,
false
]
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Update schedule: Add exception.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jscexceptions": {
"added": [
{
"jexdate": "2050-01-01",
"jextime": "12:00:00"
}
]
}
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Update schedule: Change exception date and time.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jscexceptions": {
"changed": [
{
"jexdate": "2050-01-31",
"jextime": "20:00:00"
}
]
}
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Update schedule: Delete exception.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jscexceptions": {
"deleted": [
{
"jexdate": "2050-01-01",
"jextime": "12:00:00"
}
]
}
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
}
],
"pgagent_get_schedule": [
{
"name": "Get schedule: With existing pgAgent job.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get schedules: With multiple existing pgAgent jobs.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": true
},
{
"name": "Get schedule: With existing pgAgent job while server down.",
"url": "/browser/pga_schedule/obj/",
"is_positive_test": false,
"inventory_data": {},
"test_data": {},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
},
"is_list": false
}
],
"pgagent_get_nodes_schedule": [
{
"name": "Get schedule node: With existing pgAgent job.",
"url": "/browser/pga_schedule/nodes/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get schedules nodes: With multiple existing pgAgent jobs.",
"url": "/browser/pga_schedule/nodes/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": true
}
],
"pgagent_msql_schedule": [
{
"name": "Get schedule msql: For existing pgAgent job.",
"url": "/browser/pga_schedule/msql/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jscdesc": "Msql api Test comment",
"jscstart": "2039-12-24 12:46:41 +05:30"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
}
]
}

81
web/pgadmin/browser/server_groups/servers/pgagent/schedules/tests/test_pgagent_add_schedule.py Normal file
View File

@ -0,0 +1,81 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from unittest.mock import patch
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as \
pgagent_utils
from . import utils as schedules_utils
class PgAgentAddScheduleTestCase(BaseTestGenerator):
"""This class will test the add schedule in the pgAgent job API"""
scenarios = utils.generate_scenarios("pgagent_create_schedule",
schedules_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
# Create a test job
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
# Check and Delete entry for pga_exception table for the above
# date and time as no primary key is defined for pga_exception table
# and there is a unique constraint for date and time. So when we run
# the test cases multiple time then it will fail with unique
# constraint error.
if 'jscexceptions' in self.data:
jexdate = self.data['jscexceptions'][0]["jexdate"]
jextime = self.data['jscexceptions'][0]["jextime"]
pgagent_utils.delete_pgagent_exception(self, jexdate, jextime)
self.pgagent_schedule_name = "test_sch_add%s" % str(uuid.uuid4())[1:8]
if "jscjobid" in self.data:
self.data["jscjobid"] = self.job_id
self.data["jscname"] = self.pgagent_schedule_name
if self.is_positive_test:
response = schedules_utils.api_create(self)
# Assert response
utils.assert_status_code(self, response)
# Verify in backend
response_data = json.loads(response.data)
self.schedule_id = response_data['node']['_id']
is_present = pgagent_utils.verify_pgagent_schedule(self)
self.assertTrue(is_present,
"pgAgent schedule was not created successfully.")
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = schedules_utils.api_create(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

41
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_delete_schedule.py → web/pgadmin/browser/server_groups/servers/pgagent/schedules/tests/test_pgagent_delete_schedule.py
View File

@ -10,14 +10,15 @@
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as\
pgagent_utils
from . import utils as schedules_utils
class PgAgentDeleteScheduleTestCase(BaseTestGenerator):
"""This class will test the delete pgAgent job schedule API"""
scenarios = [
('Delete pgAgent Schedule', dict(url='/browser/pga_schedule/obj/'))
]
scenarios = utils.generate_scenarios("pgagent_delete_schedule",
schedules_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
@ -26,26 +27,38 @@ class PgAgentDeleteScheduleTestCase(BaseTestGenerator):
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_delete%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch_name = "test_schedule_delete%s" % str(uuid.uuid4())[1:8]
self.schedule_id = pgagent_utils.create_pgagent_schedule(
self, sch_name, self.job_id)
# multiple schedules
if self.is_list:
sch_name2 = "test_schedule_delete%s" % str(uuid.uuid4())[1:8]
self.schedule_id_2 = pgagent_utils.create_pgagent_schedule(
self, sch_name2, self.job_id)
def runTest(self):
"""This function will deletes pgAgent job schedule"""
response = self.tester.delete(
'{0}{1}/{2}/{3}/{4}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id), str(self.schedule_id)
),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
if self.is_positive_test:
if self.is_list:
self.data['ids'] = [self.schedule_id, self.schedule_id_2]
response = schedules_utils.api_delete(self, '')
else:
response = schedules_utils.api_delete(self)
# Assert response
utils.assert_status_code(self, response)
is_present = pgagent_utils.verify_pgagent_schedule(self)
self.assertFalse(
is_present, "pgAgent schedule was not deleted successfully"
)
is_present, "pgAgent schedule was not deleted successfully")
def tearDown(self):
"""Clean up code"""

68
web/pgadmin/browser/server_groups/servers/pgagent/schedules/tests/test_pgagent_get_msql_schedule.py Normal file
View File

@ -0,0 +1,68 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as \
pgagent_utils
from . import utils as schedules_utils
class PgAgentGetMsqlScheduleTestCase(BaseTestGenerator):
"""This class will test the msql pgAgent job schedule API"""
scenarios = utils.generate_scenarios("pgagent_msql_schedule",
schedules_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_msql%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch_name = "test_schedule_msql%s" % str(uuid.uuid4())[1:8]
self.schedule_id = pgagent_utils.create_pgagent_schedule(
self, sch_name, self.job_id)
if self.is_list:
sch_name2 = "test_schedule_msql%s" % str(uuid.uuid4())[1:8]
self.schedule_id_2 = pgagent_utils.create_pgagent_schedule(
self, sch_name2, self.job_id)
def runTest(self):
"""This function will get pgAgent msql job schedule"""
if self.is_positive_test:
url_encode_data = self.data
if self.is_list:
self.data["jschedules"]["changed"][0]["jscid"] = \
self.schedule_id
self.data["jschedules"]["changed"][1]["jscid"] = \
self.schedule_id_2
url_encode_data["jobid"] = self.job_id
response = schedules_utils.\
api_get_msql(self, url_encode_data, '')
else:
url_encode_data["jscid"] = self.schedule_id
response = schedules_utils.api_get_msql(self, url_encode_data)
# Assert response
utils.assert_status_code(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

70
web/pgadmin/browser/server_groups/servers/pgagent/schedules/tests/test_pgagent_get_nodes_schedule.py Normal file
View File

@ -0,0 +1,70 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as\
pgagent_utils
from . import utils as schedules_utils
class PgAgentGetNodesScheduleTestCase(BaseTestGenerator):
"""This class will test the get nodes pgAgent job schedule API"""
scenarios = utils.generate_scenarios("pgagent_get_nodes_schedule",
schedules_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_get_nodes%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch_name = "test_schedule_get_nodes%s" % str(uuid.uuid4())[1:8]
self.schedule_id = pgagent_utils.create_pgagent_schedule(
self, sch_name, self.job_id)
if self.is_list:
sch_name2 = "test_schedule_get_nodes%s" % str(uuid.uuid4())[1:8]
self.schedule_id_2 = pgagent_utils.create_pgagent_schedule(
self, sch_name2, self.job_id)
def runTest(self):
"""This function will gets pgAgent job schedule"""
if self.is_positive_test:
if self.is_list:
response = schedules_utils.api_get(self, '')
else:
response = schedules_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = schedules_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

70
web/pgadmin/browser/server_groups/servers/pgagent/schedules/tests/test_pgagent_get_schedule.py Normal file
View File

@ -0,0 +1,70 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as\
pgagent_utils
from . import utils as schedules_utils
class PgAgentGetScheduleTestCase(BaseTestGenerator):
"""This class will test the get pgAgent job schedule API"""
scenarios = utils.generate_scenarios("pgagent_get_schedule",
schedules_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch_name = "test_schedule_get%s" % str(uuid.uuid4())[1:8]
self.schedule_id = pgagent_utils.create_pgagent_schedule(
self, sch_name, self.job_id)
if self.is_list:
sch_name2 = "test_schedule_get%s" % str(uuid.uuid4())[1:8]
self.schedule_id_2 = pgagent_utils.create_pgagent_schedule(
self, sch_name2, self.job_id)
def runTest(self):
"""This function will gets pgAgent job schedule"""
if self.is_positive_test:
if self.is_list:
response = schedules_utils.api_get(self, '')
else:
response = schedules_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = schedules_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

94
web/pgadmin/browser/server_groups/servers/pgagent/schedules/tests/test_pgagent_put_schedule.py Normal file
View File

@ -0,0 +1,94 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as \
pgagent_utils
from . import utils as schedules_utils
class PgAgentPutScheduleTestCase(BaseTestGenerator):
"""This class will test the update pgAgent schedule API"""
# Generates scenarios
scenarios = utils.generate_scenarios("pgagent_put_schedule",
schedules_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_update%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch_name = "test_schedule_update%s" % str(uuid.uuid4())[1:8]
self.schedule_id = pgagent_utils.create_pgagent_schedule(
self, sch_name, self.job_id)
def runTest(self):
"""This function will update pgAgent schedule"""
# Check and Delete entry for pga_exception table for the specified
# date and time as no primary key is defined for pga_exception table
# and there is a unique constraint for date and time. So when we run
# the test cases multiple time then it will fail with unique
# constraint error.
if 'jscexceptions' in self.data:
exception_data = self.data['jscexceptions']
if 'added' in exception_data:
pgagent_utils.delete_pgagent_exception(
self, self.data['jscexceptions']['added'][0]['jexdate'],
self.data['jscexceptions']['added'][0]['jextime'])
elif 'changed' in exception_data:
date = self.data['jscexceptions']['changed'][0]['jexdate']
time = self.data['jscexceptions']['changed'][0]['jextime']
self.excp_id = pgagent_utils.create_pgagent_exception(
self, self.schedule_id, date, time)
self.data['jscexceptions']['changed'][0]['jexid'] = \
self.excp_id
elif 'deleted' in exception_data:
date = self.data['jscexceptions']['deleted'][0]['jexdate']
time = self.data['jscexceptions']['deleted'][0]['jextime']
self.excp_id = pgagent_utils.create_pgagent_exception(
self, self.schedule_id, date, time)
self.data['jscexceptions']['deleted'][0]['jexid'] = \
self.excp_id
self.data['jscid'] = str(self.schedule_id)
if self.is_positive_test:
response = schedules_utils.api_put(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = pgagent_utils.api_put(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_schedule(self)
pgagent_utils.delete_pgagent_job(self)

62
web/pgadmin/browser/server_groups/servers/pgagent/schedules/tests/utils.py Normal file
View File

@ -0,0 +1,62 @@
import sys
import traceback
import os
import json
from urllib.parse import urlencode
from regression.python_test_utils import test_utils as utils
# Load test data from json file.
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/schedules_test_data.json") as data_file:
test_cases = json.load(data_file)
# api methods
def api_create(self):
return self.tester.post('{0}{1}/{2}/{3}/'.
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id),
data=json.dumps(self.data),
content_type='html/json')
def api_delete(self, schedule_id=None):
if schedule_id is None:
schedule_id = self.schedule_id
return self.tester.delete('{0}{1}/{2}/{3}/{4}'.
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id,
schedule_id),
data=json.dumps(self.data),
content_type='html/json')
def api_put(self):
return self.tester.put('{0}{1}/{2}/{3}/{4}'.
format(self.url, utils.SERVER_GROUP, self.server_id,
self.job_id, self.schedule_id),
data=json.dumps(self.data),
content_type='html/json')
def api_get(self, schedule_id=None):
if schedule_id is None:
schedule_id = self.schedule_id
return self.tester.get('{0}{1}/{2}/{3}/{4}'.
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id,
schedule_id),
content_type='html/json')
def api_get_msql(self, url_encode_data, schedule_id=None):
if schedule_id is None:
schedule_id = '/' + str(self.schedule_id)
return self.tester.get("{0}{1}/{2}/{3}{4}?{5}".
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id,
schedule_id,
urlencode(url_encode_data)),
data=json.dumps(self.data),
follow_redirects=True)

0
web/pgadmin/browser/server_groups/servers/pgagent/steps/tests/__init__.py Normal file
View File

368
web/pgadmin/browser/server_groups/servers/pgagent/steps/tests/steps_test_data.json Normal file
View File

@ -0,0 +1,368 @@
{
"pgagent_create_step": [
{
"name": "Create step: For existing pgAgent job.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jstjobid": "job_id",
"jstname": "pgagent_step_name",
"jstdesc": "",
"jstenabled": true,
"jstkind": true,
"jstconntype": true,
"jstcode": "SELECT 1;",
"jstconnstr": null,
"jstdbname": "postgres",
"jstonerror": "f",
"jstnextrun": ""
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Create step: For existing pgAgent job while server id down.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": false,
"inventory_data": {},
"test_data": {
"jstjobid": "job_id",
"jstname": "pgagent_step_name",
"jstdesc": "",
"jstenabled": true,
"jstkind": true,
"jstconntype": true,
"jstcode": "SELECT 1;",
"jstconnstr": null,
"jstdbname": "postgres",
"jstonerror": "f",
"jstnextrun": ""
},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
}
}
],
"pgagent_delete_step": [
{
"name": "Delete step: For existing pgAgent job.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Delete multiple steps: For existing pgAgent steps.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": true
}
],
"pgagent_put_step": [
{
"name": "Update step: For existing pgAgent job step with kind, description, code and error.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jstdesc": "Test Steps",
"jstkind": false,
"jstcode": "SELECT 12345",
"jstonerror": "i"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Update step: For existing pgAgent job step with connection type and string.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jstconntype": false,
"jstconnstr": "host=localhost port=5432 dbname=mydb connect_timeout=10"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Update step: For existing pgAgent job step with connection string.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jstconnstr": "host=localhost port=5432 dbname=mydb connect_timeout=10"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Update step: For non-existing pgAgent job step.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": false,
"inventory_data": {},
"test_data": {
"step_id": 9999,
"jstconnstr": "host=localhost port=5432 dbname=mydb connect_timeout=10"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410,
"error_msg": "Could not find the specified job step.",
"test_result_data": {}
}
},
{
"name": "Update step: For existing pgAgent job step while server is down.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": false,
"inventory_data": {},
"test_data": {
"jstconnstr": "host=localhost port=5432 dbname=mydb connect_timeout=10"
},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
}
}
],
"pgagent_get_step": [
{
"name": "Get step: For existing pgAgent job step.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get multiple steps: For existing pgAgent steps.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": true
},
{
"name": "Get step: For existing pgAgent job step while server is down.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": false,
"inventory_data": {},
"test_data": {},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get step: For non-existing pgAgent job step.",
"url": "/browser/pga_jobstep/nodes/",
"is_positive_test": false,
"inventory_data": {},
"test_data": {
"step_id": 9999
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410,
"error_msg": "Could not find the specified job step.",
"test_result_data": {}
},
"is_list": false
}
],
"pgagent_get_nodes_step": [
{
"name": "Get step nodes: For existing pgAgent job step.",
"url": "/browser/pga_jobstep/nodes/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get multiple steps nodes: For existing pgAgent steps.",
"url": "/browser/pga_jobstep/nodes/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": true
},
{
"name": "Get step node: For non-existing pgAgent job step.",
"url": "/browser/pga_jobstep/nodes/",
"is_positive_test": false,
"inventory_data": {},
"test_data": {
"step_id": 9999
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410,
"error_msg": "Could not find the specified job step.",
"test_result_data": {}
},
"is_list": false
}
],
"pgagent_get_step_stats": [
{
"name": "Get step stats: For existing pgAgent job step.",
"url": "/browser/pga_jobstep/stats/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Get step stats: For existing pgAgent job step while server is down.",
"url": "/browser/pga_jobstep/stats/",
"is_positive_test": false,
"inventory_data": {},
"test_data": {},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
}
}
],
"pgagent_get_msql_step": [
{
"name": "Get step msql: For existing pgAgent job step with kind, description, code and error.",
"url": "/browser/pga_jobstep/msql/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jstdesc": "Test Steps",
"jstkind": false,
"jstcode": "SELECT 12345",
"jstonerror": "i"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Get step msql: For existing pgAgent job step with connection type and string.",
"url": "/browser/pga_jobstep/obj/",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"jstconntype": false,
"jstconnstr": "host=localhost port=5432 dbname=mydb connect_timeout=10"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
}
]
}

70
web/pgadmin/browser/server_groups/servers/pgagent/steps/tests/test_pgagent_add_steps.py Normal file
View File

@ -0,0 +1,70 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from unittest.mock import patch
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as \
pgagent_utils
from . import utils as steps_utils
class PgAgentAddStepTestCase(BaseTestGenerator):
"""This class will test the add step in the pgAgent job API"""
scenarios = utils.generate_scenarios("pgagent_create_step",
steps_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
# Create job
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
self.pgagent_step_name = "test_step_add%s" % str(uuid.uuid4())[1:8]
self.data["jstjobid"] = self.job_id
self.data["jstname"] = self.pgagent_step_name
if self.is_positive_test:
response = steps_utils.api_create(self)
# Assert response
utils.assert_status_code(self, response)
# Verify in backend
response_data = json.loads(response.data)
self.step_id = response_data['node']['_id']
is_present = pgagent_utils.verify_pgagent_step(self)
self.assertTrue(is_present,
"pgAgent step was not created successfully.")
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = steps_utils.api_create(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

40
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_delete_steps.py → web/pgadmin/browser/server_groups/servers/pgagent/steps/tests/test_pgagent_delete_steps.py
View File

@ -10,14 +10,15 @@
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as \
pgagent_utils
from . import utils as steps_utils
class PgAgentDeleteStepTestCase(BaseTestGenerator):
"""This class will test the delete pgAgent job step API"""
scenarios = [
('Delete pgAgent Step', dict(url='/browser/pga_jobstep/obj/'))
]
scenarios = utils.generate_scenarios("pgagent_delete_step",
steps_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
@ -26,26 +27,37 @@ class PgAgentDeleteStepTestCase(BaseTestGenerator):
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_delete%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
step_name = "test_step_delete%s" % str(uuid.uuid4())[1:8]
self.step_id = pgagent_utils.create_pgagent_step(
self, step_name, self.job_id)
if self.is_list:
step_name_2 = "test_step_delete%s" % str(uuid.uuid4())[1:8]
self.step_id_2 = pgagent_utils.create_pgagent_step(
self, step_name_2, self.job_id)
def runTest(self):
"""This function will deletes pgAgent job step"""
response = self.tester.delete(
'{0}{1}/{2}/{3}/{4}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id), str(self.step_id)
),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
if self.is_positive_test:
if self.is_list:
self.data['ids'] = [self.step_id, self.step_id_2]
response = steps_utils.api_delete(self, '')
else:
response = steps_utils.api_delete(self)
# Assert response
utils.assert_status_code(self, response)
is_present = pgagent_utils.verify_pgagent_step(self)
self.assertFalse(
is_present, "pgAgent step was not deleted successfully"
)
is_present, "pgAgent step was not deleted successfully")
def tearDown(self):
"""Clean up code"""

75
web/pgadmin/browser/server_groups/servers/pgagent/steps/tests/test_pgagent_get_nodes_steps.py Normal file
View File

@ -0,0 +1,75 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as \
pgagent_utils
from . import utils as steps_utils
class PgAgentGetNodesStepTestCase(BaseTestGenerator):
"""This class will fetch pgAgent job step nodes"""
scenarios = utils.generate_scenarios("pgagent_get_nodes_step",
steps_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_get_nodes%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
step_name = "test_step_get_nodes%s" % str(uuid.uuid4())[1:8]
self.step_id = pgagent_utils.create_pgagent_step(
self, step_name, self.job_id)
if self.is_list:
step_name_2 = "test_step_get_nodes%s" % str(uuid.uuid4())[1:8]
self.step_id_2 = pgagent_utils.create_pgagent_step(
self, step_name_2, self.job_id)
def runTest(self):
"""This function will fetch pgAgent job nodes"""
if self.is_positive_test:
if self.is_list:
self.data['ids'] = [self.step_id, self.step_id_2]
response = steps_utils.api_get(self, '')
else:
response = steps_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = steps_utils.api_get(self)
else:
if "step_id" in self.data:
self.step_id = self.data["step_id"]
response = steps_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

75
web/pgadmin/browser/server_groups/servers/pgagent/steps/tests/test_pgagent_get_steps.py Normal file
View File

@ -0,0 +1,75 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as \
pgagent_utils
from . import utils as steps_utils
class PgAgentGetStepTestCase(BaseTestGenerator):
"""This class will test the get pgAgent job step API"""
scenarios = utils.generate_scenarios("pgagent_get_step",
steps_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
step_name = "test_step_get%s" % str(uuid.uuid4())[1:8]
self.step_id = pgagent_utils.create_pgagent_step(
self, step_name, self.job_id)
if self.is_list:
step_name_2 = "test_step_get%s" % str(uuid.uuid4())[1:8]
self.step_id_2 = pgagent_utils.create_pgagent_step(
self, step_name_2, self.job_id)
def runTest(self):
"""This function will get pgAgent job step"""
if self.is_positive_test:
if self.is_list:
self.data['ids'] = [self.step_id, self.step_id_2]
response = steps_utils.api_get(self, '')
else:
response = steps_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = steps_utils.api_get(self)
else:
if "step_id" in self.data:
self.step_id = self.data["step_id"]
response = steps_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

62
web/pgadmin/browser/server_groups/servers/pgagent/steps/tests/test_pgagent_get_steps_stats.py Normal file
View File

@ -0,0 +1,62 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as \
pgagent_utils
from . import utils as steps_utils
class PgAgentGetStepTestCase(BaseTestGenerator):
"""This class will test the get pgAgent job step stats"""
scenarios = utils.generate_scenarios("pgagent_get_step_stats",
steps_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_get_stats%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
step_name = "test_step_get_stats%s" % str(uuid.uuid4())[1:8]
self.step_id = pgagent_utils.create_pgagent_step(
self, step_name, self.job_id)
def runTest(self):
"""This function will get pgAgent job step stats"""
if self.is_positive_test:
response = steps_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = steps_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

57
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_put_step.py → web/pgadmin/browser/server_groups/servers/pgagent/steps/tests/test_pgagent_put_steps.py
View File

@ -6,33 +6,21 @@
# This software is released under the PostgreSQL Licence
#
##########################################################################
from unittest.mock import patch
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
from pgadmin.browser.server_groups.servers.pgagent.tests import utils as \
pgagent_utils
from . import utils as steps_utils
class PgAgentPutStepTestCase(BaseTestGenerator):
"""This class will test the update pgAgent steps API"""
scenarios = [
('Update step with kind, description, code and error', dict(
url='/browser/pga_jobstep/obj/',
data={
'jstdesc': 'Test Steps',
'jstkind': False,
'jstcode': 'SELECT 12345',
'jstonerror': 'i'
})),
('Update step with connection type and string', dict(
url='/browser/pga_jobstep/obj/',
data={
'jstconntype': False,
'jstconnstr':
'host=localhost port=5432 dbname=mydb connect_timeout=10'
}))
]
scenarios = utils.generate_scenarios("pgagent_put_step",
steps_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
@ -41,25 +29,38 @@ class PgAgentPutStepTestCase(BaseTestGenerator):
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_update%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
step_name = "test_step_update%s" % str(uuid.uuid4())[1:8]
self.step_id = pgagent_utils.create_pgagent_step(
self, step_name, self.job_id)
def runTest(self):
"""This function will update pgAgent steps"""
self.data['jstid'] = str(self.step_id)
response = self.tester.put(
'{0}{1}/{2}/{3}/{4}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id), str(self.step_id)
),
data=json.dumps(self.data),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
if self.is_positive_test:
response = steps_utils.api_put(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = steps_utils.api_put(self)
else:
if "step_id" in self.data:
self.step_id = self.data["step_id"]
response = steps_utils.api_put(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""

59
web/pgadmin/browser/server_groups/servers/pgagent/steps/tests/utils.py Normal file
View File

@ -0,0 +1,59 @@
import os
import json
from urllib.parse import urlencode
from regression.python_test_utils import test_utils as utils
# Load test data from json file.
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/steps_test_data.json") as data_file:
test_cases = json.load(data_file)
# api methods
def api_create(self):
return self.tester.post('{0}{1}/{2}/{3}/'.
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id),
data=json.dumps(self.data),
content_type='html/json')
def api_delete(self, step_id=None):
if step_id is None:
step_id = self.step_id
return self.tester.delete('{0}{1}/{2}/{3}/{4}'.
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id, step_id),
data=json.dumps(self.data),
content_type='html/json')
def api_put(self):
return self.tester.put('{0}{1}/{2}/{3}/{4}'.
format(self.url, utils.SERVER_GROUP, self.server_id,
self.job_id, self.step_id),
data=json.dumps(self.data),
content_type='html/json')
def api_get(self, step_id=None):
if step_id is None:
step_id = self.step_id
return self.tester.get('{0}{1}/{2}/{3}/{4}'.
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id,
step_id),
content_type='html/json')
def api_get_msql(self, url_encode_data, step_id=None):
if step_id is None:
step_id = '/' + str(self.step_id)
return self.tester.get("{0}{1}/{2}/{3}{4}?{5}".
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id,
step_id,
urlencode(url_encode_data)),
data=json.dumps(self.data),
follow_redirects=True)

1375
web/pgadmin/browser/server_groups/servers/pgagent/tests/pgagent_test_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

90
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_add.py
View File

@ -1,90 +0,0 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentAddTestCase(BaseTestGenerator):
"""This class will test the add pgAgent job API"""
scenarios = [
('Add pgAgent job', dict(url='/browser/pga_job/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
def runTest(self):
"""This function will adds pgAgent job"""
self.pgagent_job = "test_job_add%s" % str(uuid.uuid4())[1:8]
data = {
'jobname': self.pgagent_job,
'jobenabled': True,
'jobhostagent': '',
'jobjclid': 1,
'jobdesc': '',
'jsteps': [{
'jstid': None,
'jstjobid': None,
'jstname': 'test_step',
'jstdesc': '',
'jstenabled': True,
'jstkind': True,
'jstconntype': True,
'jstcode': 'SELECT 1;',
'jstconnstr': None,
'jstdbname': 'postgres',
'jstonerror': 'f',
'jstnextrun': '',
}],
'jschedules': [{
'jscid': None,
'jscjobid': None,
'jscname': 'test_sch',
'jscdesc': '',
'jscenabled': True,
'jscstart': '2050-01-01 12:14:21 +05:30',
'jscend': None,
'jscweekdays': [False] * 7,
'jscmonthdays': [False] * 32,
'jscmonths': [False] * 12,
'jschours': [False] * 24,
'jscminutes': [False] * 60,
'jscexceptions': [{'jexdate': '2050-01-01',
'jextime': '12:00'}],
}],
}
response = self.tester.post(
'{0}{1}/{2}/'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id)
),
data=json.dumps(data),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.job_id = response_data['node']['_id']
is_present = pgagent_utils.verify_pgagent_job(self)
self.assertTrue(
is_present, "pgAgent job was not created successfully"
)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

47
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_delete_multiple.py
View File

@ -1,47 +0,0 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
import json
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentDeleteMultipleTestCase(BaseTestGenerator):
"""This class will test the delete multiple pgAgent job API"""
scenarios = [
('Delete multiple pgAgent job', dict(url='/browser/pga_job/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name1 = "test_job1_delete%s" % str(uuid.uuid4())[1:8]
self.job_id1 = pgagent_utils.create_pgagent_job(self, name1)
name2 = "test_job2_delete%s" % str(uuid.uuid4())[1:8]
self.job_id2 = pgagent_utils.create_pgagent_job(self, name2)
def runTest(self):
"""This function will deletes pgAgent job"""
response = self.tester.delete(
'{0}{1}/{2}/'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id)
),
data=json.dumps({'ids': [self.job_id1, self.job_id2]}),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
def tearDown(self):
"""Clean up code"""

74
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_job_add.py Normal file
View File

@ -0,0 +1,74 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from unittest.mock import patch
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentAddTestCase(BaseTestGenerator):
"""This class will test the add pgAgent job API"""
# Generates scenarios
scenarios = utils.generate_scenarios("pgagent_job_create",
pgagent_utils.test_cases)
def setUp(self):
# Load test data
self.data = self.test_data
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
def runTest(self):
"""This function will adds pgAgent job"""
self.pgagent_job = "test_job_add%s" % str(uuid.uuid4())[1:8]
if "jobname" in self.data:
self.data["jobname"] = self.pgagent_job
if self.is_positive_test:
response = pgagent_utils.api_create(self)
# Assert response
utils.assert_status_code(self, response)
# Verify in backend
response_data = json.loads(response.data)
self.job_id = response_data['node']['_id']
is_present = pgagent_utils.verify_pgagent_job(self)
self.assertTrue(is_present,
"pgAgent job was not created successfully")
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = pgagent_utils.api_create(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
else:
response = pgagent_utils.api_create(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
if self.is_positive_test:
pgagent_utils.delete_pgagent_job(self)

37
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_delete.py → web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_job_delete.py
View File

@ -15,9 +15,9 @@ from . import utils as pgagent_utils
class PgAgentDeleteTestCase(BaseTestGenerator):
"""This class will test the delete pgAgent job API"""
scenarios = [
('Delete pgAgent job', dict(url='/browser/pga_job/obj/'))
]
# Generates scenarios
scenarios = utils.generate_scenarios("pgagent_job_delete",
pgagent_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
@ -26,24 +26,35 @@ class PgAgentDeleteTestCase(BaseTestGenerator):
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_delete%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
if self.is_list:
name2 = "test_job2_delete%s" % str(uuid.uuid4())[1:8]
self.job_id2 = pgagent_utils.create_pgagent_job(self, name2)
def runTest(self):
"""This function will deletes pgAgent job"""
response = self.tester.delete(
'{0}{1}/{2}/{3}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
if self.is_positive_test:
if self.is_list:
self.data['ids'] = [self.job_id, self.job_id2]
response = pgagent_utils.api_delete(self, '')
else:
response = pgagent_utils.api_delete(self)
# Assert response
utils.assert_status_code(self, response)
is_present = pgagent_utils.verify_pgagent_job(self)
self.assertFalse(
is_present, "pgAgent job was not deleted successfully"
)
is_present, "pgAgent job was not deleted successfully")
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)
if self.is_list:
pgagent_utils.delete_pgagent_job(self, self.job_id2)

73
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_job_get.py Normal file
View File

@ -0,0 +1,73 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentGetTestCase(BaseTestGenerator):
"""This class will test the get pgAgent job API"""
scenarios = utils.generate_scenarios("pgagent_job_get",
pgagent_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
# In case of multiple views
if self.is_list:
name_2 = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id_2 = pgagent_utils.create_pgagent_job(self, name_2)
def runTest(self):
"""This function will get pgAgent job"""
if self.is_positive_test:
if self.is_list:
response = pgagent_utils.api_get(self, '')
else:
response = pgagent_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = pgagent_utils.api_get(self)
elif 'job_id' in self.data:
# Non-existing job id
existing_job_id = self.job_id
self.job_id = self.data["job_id"]
response = pgagent_utils.api_get(self)
self.job_id = existing_job_id
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)
if self.is_list:
pgagent_utils.delete_pgagent_job(self, self.job_id_2)

37
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_get.py → web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_job_get_msql.py
View File

@ -6,18 +6,20 @@
# This software is released under the PostgreSQL Licence
#
##########################################################################
from unittest.mock import patch
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentGetTestCase(BaseTestGenerator):
"""This class will test the get pgAgent job API"""
scenarios = [
('Get pgAgent job', dict(url='/browser/pga_job/obj/'))
]
class PgAgentGetMsqlTestCase(BaseTestGenerator):
"""This class will test the put pgAgent job API"""
# Generates scenarios
scenarios = utils.generate_scenarios("pgagent_job_msql",
pgagent_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
@ -26,19 +28,24 @@ class PgAgentGetTestCase(BaseTestGenerator):
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
# Load test data
self.data = self.test_data
name = "test_job_msql%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
"""This function will get pgAgent job"""
response = self.tester.get(
'{0}{1}/{2}/{3}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
"""This function will get msql for pgAgent job"""
if self.is_positive_test:
url_encode_data = self.data
url_encode_data["jobid"] = self.job_id
response = pgagent_utils.api_get_msql(self, url_encode_data)
# Assert response
utils.assert_status_code(self, response)
def tearDown(self):
"""Clean up code"""

73
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_job_get_nodes.py Normal file
View File

@ -0,0 +1,73 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentGetNodesTestCase(BaseTestGenerator):
"""This class will test the get pgAgent job nodes API"""
scenarios = utils.generate_scenarios("pgagent_job_get_nodes",
pgagent_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_get_nodes%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
# In case of multiple views
if self.is_list:
name_2 = "test_job_get_nodes%s" % str(uuid.uuid4())[1:8]
self.job_id_2 = pgagent_utils.create_pgagent_job(self, name_2)
def runTest(self):
"""This function will get pgAgent job nodes"""
if self.is_positive_test:
if self.is_list:
response = pgagent_utils.api_get(self, '')
else:
response = pgagent_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = pgagent_utils.api_get(self)
elif 'job_id' in self.data:
# Non-existing job id
existing_job_id = self.job_id
self.job_id = self.data["job_id"]
response = pgagent_utils.api_get(self)
self.job_id = existing_job_id
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)
if self.is_list:
pgagent_utils.delete_pgagent_job(self, self.job_id_2)

57
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_job_get_statistics.py Normal file
View File

@ -0,0 +1,57 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentGetStatsTestCase(BaseTestGenerator):
"""This class will test the get pgAgent job stats API"""
scenarios = utils.generate_scenarios("pgagent_job_get_statistics",
pgagent_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_get_stats%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
"""This function will get pgAgent job stats"""
if self.is_positive_test:
response = pgagent_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = pgagent_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

64
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_job_put.py Normal file
View File

@ -0,0 +1,64 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from unittest.mock import patch
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentPutTestCase(BaseTestGenerator):
"""This class will test the put pgAgent job API"""
# Generates scenarios
scenarios = utils.generate_scenarios("pgagent_job_put",
pgagent_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_put%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
"""This function will put pgAgent job"""
if 'jschedules' in self.data and 'added' in self.data['jschedules']:
self.data['jschedules']['added'][0]['jscjobid'] = self.job_id
if 'jsteps' in self.data and 'added' in self.data['jsteps']:
self.data['jsteps']['added'][0]['jstjobid'] = self.job_id
if self.is_positive_test:
response = pgagent_utils.api_put(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = pgagent_utils.api_put(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

70
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_job_sql.py Normal file
View File

@ -0,0 +1,70 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentSqlTestCase(BaseTestGenerator):
"""This class will test the get pgAgent job sql API"""
scenarios = utils.generate_scenarios("pgagent_job_sql",
pgagent_utils.test_cases)
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
# Load test data
self.data = self.test_data
name = "test_job_sql%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch_name = "test_schedule_update%s" % str(uuid.uuid4())[1:8]
self.schedule_id = pgagent_utils.create_pgagent_schedule(
self, sch_name, self.job_id)
self.excp_id = pgagent_utils.create_pgagent_exception(
self, self.schedule_id, "2050-01-01", "12:00:00")
def runTest(self):
"""This function will get pgAgent job sql"""
if self.is_positive_test:
response = pgagent_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = pgagent_utils.api_get(self)
elif 'job_id' in self.data:
# Non-existing job id
existing_job_id = self.job_id
self.job_id = self.data["job_id"]
response = pgagent_utils.api_get(self)
self.job_id = existing_job_id
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

120
web/pgadmin/browser/server_groups/servers/pgagent/tests/test_pgagent_put.py
View File

@ -1,120 +0,0 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentPutTestCase(BaseTestGenerator):
"""This class will test the put pgAgent job API"""
scenarios = [
('Update pgAgent job description', dict(
url='/browser/pga_job/obj/',
data={
"jobdesc": "This is a test comment",
})),
('Update pgagent job add schedule', dict(
url='/browser/pga_job/obj/',
data={'jschedules': {
'added': [{
'jscjobid': '',
'jscenabled': True,
'jscdesc': 'This is a test comment',
'jscname': 'test_sc1',
'jscexceptions': [{'jexdate': '2050-01-01',
'jextime': '12:30'}],
'jscstart': '2050-01-01 12:14:21 +05:30',
'jscend': '2050-03-01 12:14:21 +05:30',
'jscminutes': [False] * 60,
# Below format is added to test the malformed array
# literal issue.
'jscweekdays': '[false, true, false, true, false, '
'false, false]',
'jscmonthdays': [False] * 32,
'jschours': [True] * 24,
# Below format is added to test the malformed array
# literal issue.
'jscmonths': '[true, false, false, true, false, false,'
'true, false, false, true, false, false]'
}]
}
})),
('Update pgagent job add steps with local connection', dict(
url='/browser/pga_job/obj/',
data={'jsteps': {
'added': [{
'jstjobid': '',
'jstname': 'test_st1',
'jstdesc': '',
'jstenabled': True,
'jstkind': True,
'jstconntype': True,
'jstcode': 'SELECT 1;',
'jstconnstr': None,
'jstdbname': 'postgres',
'jstonerror': 'f',
'jstnextrun': '',
}]
}
})),
('Update pgagent job add steps with remote connection', dict(
url='/browser/pga_job/obj/',
data={'jsteps': {
'added': [{
'jstjobid': '',
'jstname': 'test_st1',
'jstdesc': '',
'jstenabled': True,
'jstkind': True,
'jstconntype': False,
'jstcode': 'SELECT 1;',
'jstconnstr': 'host=localhost port=5432 dbname=postgres',
'jstdbname': '',
'jstonerror': 'f',
'jstnextrun': '',
}]
}
})),
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_put%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
"""This function will put pgAgent job"""
if 'jschedules' in self.data and 'added' in self.data['jschedules']:
self.data['jschedules']['added'][0]['jscjobid'] = self.job_id
if 'jsteps' in self.data and 'added' in self.data['jsteps']:
self.data['jsteps']['added'][0]['jstjobid'] = self.job_id
response = self.tester.put(
'{0}{1}/{2}/{3}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
data=json.dumps(self.data),
follow_redirects=True,
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

80
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_add_schedule.py
View File

@ -1,80 +0,0 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentAddScheduleTestCase(BaseTestGenerator):
"""This class will test the add schedule in the pgAgent job API"""
scenarios = [
('Create schedule with exception in pgAgent job', dict(
url='/browser/pga_schedule/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
# Check and Delete entry for pga_exception table for the above
# date and time as no primary key is defined for pga_exception table
# and there is a unique constraint for date and time. So when we run
# the test cases multiple time then it will fail with unique
# constraint error.
jexdate = '2050-01-01'
jextime = '12:00:00'
pgagent_utils.delete_pgagent_exception(self, jexdate, jextime)
self.pgagent_schedule_name = "test_sch_add%s" % str(uuid.uuid4())[1:8]
data = {
'jscjobid': self.job_id,
'jscenabled': True,
'jscdesc': '',
'jscname': self.pgagent_schedule_name,
'jscexceptions': [{'jexdate': jexdate,
'jextime': jextime}],
'jscstart': '2050-01-01 12:14:21 +05:30',
'jscend': '2050-03-01 12:14:21 +05:30',
'jscminutes': [False] * 60,
'jscweekdays': [True] * 7,
'jscmonthdays': [True] * 32,
'jschours': [False] * 24,
'jscmonths': [True] * 12
}
response = self.tester.post(
'{0}{1}/{2}/{3}/'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
data=json.dumps(data),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.schedule_id = response_data['node']['_id']
is_present = pgagent_utils.verify_pgagent_schedule(self)
self.assertTrue(
is_present, "pgAgent schedule was not created successfully"
)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

69
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_add_steps.py
View File

@ -1,69 +0,0 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentAddStepTestCase(BaseTestGenerator):
"""This class will test the add step in the pgAgent job API"""
scenarios = [
('Create step for pgAgent job', dict(
url='/browser/pga_jobstep/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
self.pgagent_step_name = "test_step_add%s" % str(uuid.uuid4())[1:8]
data = {
'jstjobid': self.job_id,
'jstname': self.pgagent_step_name,
'jstdesc': '',
'jstenabled': True,
'jstkind': True,
'jstconntype': True,
'jstcode': 'SELECT 1;',
'jstconnstr': None,
'jstdbname': 'postgres',
'jstonerror': 'f',
'jstnextrun': '',
}
response = self.tester.post(
'{0}{1}/{2}/{3}/'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
data=json.dumps(data),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
response_data = json.loads(response.data)
self.step_id = response_data['node']['_id']
is_present = pgagent_utils.verify_pgagent_step(self)
self.assertTrue(
is_present, "pgAgent step was not created successfully"
)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

56
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_delete_multiple_schedules.py
View File

@ -1,56 +0,0 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
import json
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentDeleteMultipleSchedulesTestCase(BaseTestGenerator):
"""This class will test the delete pgAgent job schedule API"""
scenarios = [
('Delete multiple pgAgent schedules',
dict(url='/browser/pga_schedule/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_multi_sc_job_delete%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch1_name = "test_multi_schedule1_delete%s" % str(uuid.uuid4())[1:8]
self.schedule_id1 = pgagent_utils.create_pgagent_schedule(
self, sch1_name, self.job_id)
# Create one more schedule
sch2_name = "test_multi_schedule2_delete%s" % str(uuid.uuid4())[1:8]
self.schedule_id2 = pgagent_utils.create_pgagent_schedule(
self, sch2_name, self.job_id)
def runTest(self):
"""This function will deletes pgAgent job schedule"""
response = self.tester.delete(
'{0}{1}/{2}/{3}/'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
data=json.dumps({'ids': [self.schedule_id1, self.schedule_id2]}),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

54
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_delete_multiple_steps.py
View File

@ -1,54 +0,0 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
import json
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentDeleteMultipleStepsTestCase(BaseTestGenerator):
"""This class will test the delete pgAgent job steps API"""
scenarios = [
('Delete multiple pgAgent steps',
dict(url='/browser/pga_jobstep/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_multiple_st_job_delete%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
step_name1 = "test_multiple_step1_delete%s" % str(uuid.uuid4())[1:8]
self.step_id1 = pgagent_utils.create_pgagent_step(
self, step_name1, self.job_id)
step_name2 = "test_multiple_step2_delete%s" % str(uuid.uuid4())[1:8]
self.step_id2 = pgagent_utils.create_pgagent_step(
self, step_name2, self.job_id)
def runTest(self):
"""This function will deletes pgAgent job schedule"""
response = self.tester.delete(
'{0}{1}/{2}/{3}/'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
data=json.dumps({'ids': [self.step_id1, self.step_id2]}),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

131
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_put_schedule.py
View File

@ -1,131 +0,0 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentPutScheduleTestCase(BaseTestGenerator):
"""This class will test the update pgAgent schedule API"""
scenarios = [
('Update schedule with start and end time', dict(
url='/browser/pga_schedule/obj/',
data={
'jscdesc': 'Test Schedule',
'jscstart': '2050-01-01 12:00:00 +05:30',
'jscend': '2050-01-20 12:00:00 +05:30',
})),
('Update schedule with repeat', dict(
url='/browser/pga_schedule/obj/',
data={
'jscmonthdays': '[true,false,true,false,true,false,false,'
'false,false,false,false,false,false,false,'
'false,false,false,false,false,false,false,'
'false,false,false,false,false,false,false,'
'false,false,false,false]',
'jscweekdays': '[true,false,false,true,false,false,false]',
'jscmonths': '[true,false,false,true,false,false,false,false,'
'false,false,false,false]',
'jschours': '[false,false,false,false,true,false,false,false,'
'false,false,false,false,false,false,false,false,'
'false,false,false,false,false,false,false,false]'
})),
('Update schedule add exception', dict(
url='/browser/pga_schedule/obj/',
data={
'jscexceptions': {
'added': [{'jexdate': '2050-01-01',
'jextime': '12:00:00'}]
}},
delete_existing_exception=True)),
('Update schedule change exception date and time', dict(
url='/browser/pga_schedule/obj/',
data={
'jscexceptions': {
'changed': [{'jexdate': '2050-01-31',
'jextime': '20:00:00'}]
}},
create_exception=True)),
('Update schedule delete exception', dict(
url='/browser/pga_schedule/obj/',
data={
'jscexceptions': {
'deleted': [{'jexdate': '2050-01-01',
'jextime': '12:00:00'}]
}},
create_exception=True,
is_delete=True)),
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_update%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch_name = "test_schedule_update%s" % str(uuid.uuid4())[1:8]
self.schedule_id = pgagent_utils.create_pgagent_schedule(
self, sch_name, self.job_id)
def runTest(self):
"""This function will update pgAgent schedule"""
# Check and Delete entry for pga_exception table for the specified
# date and time as no primary key is defined for pga_exception table
# and there is a unique constraint for date and time. So when we run
# the test cases multiple time then it will fail with unique
# constraint error.
if hasattr(self, 'delete_existing_exception'):
pgagent_utils.delete_pgagent_exception(
self, self.data['jscexceptions']['added'][0]['jexdate'],
self.data['jscexceptions']['added'][0]['jextime'])
# Create exception for update and delete
if hasattr(self, 'create_exception'):
date = None
time = None
if hasattr(self, 'is_delete'):
date = self.data['jscexceptions']['deleted'][0]['jexdate']
time = self.data['jscexceptions']['deleted'][0]['jextime']
else:
date = self.data['jscexceptions']['changed'][0]['jexdate']
time = self.data['jscexceptions']['changed'][0]['jextime']
self.excp_id = pgagent_utils.create_pgagent_exception(
self, self.schedule_id, date, time)
# Add created exception id in data
if hasattr(self, 'is_delete'):
self.data['jscexceptions']['deleted'][0]['jexid'] = \
self.excp_id
else:
self.data['jscexceptions']['changed'][0]['jexid'] = \
self.excp_id
self.data['jscid'] = str(self.schedule_id)
response = self.tester.put(
'{0}{1}/{2}/{3}/{4}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id), str(self.schedule_id)
),
data=json.dumps(self.data),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_schedule(self)
pgagent_utils.delete_pgagent_job(self)

45
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_stats.py
View File

@ -1,45 +0,0 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentStatsTestCase(BaseTestGenerator):
"""This class will test the stats pgAgent job API"""
scenarios = [
('Check the stats of pgAgent job', dict(url='/browser/pga_job/stats/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
"""This function will check stats of pgAgent job"""
response = self.tester.get(
'{0}{1}/{2}/{3}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
content_type='html/json'
)
self.assertEqual(response.status_code, 200)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

59
web/pgadmin/browser/server_groups/servers/pgagent/tests/utils.py
View File

@ -9,11 +9,64 @@
import sys
import traceback
import os
import json
from urllib.parse import urlencode
from regression.python_test_utils import test_utils as utils
from regression import parent_node_dict
from pgadmin.utils import server_utils as server_utils
# Load test data from json file.
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/pgagent_test_data.json") as data_file:
test_cases = json.load(data_file)
# api methods
def api_create(self):
return self.tester.post('{0}{1}/{2}/'.
format(self.url, str(utils.SERVER_GROUP),
str(self.server_id)),
data=json.dumps(self.data),
content_type='html/json')
def api_get(self, job_id=None):
if job_id is None:
job_id = self.job_id
return self.tester.get('{0}{1}/{2}/{3}'.
format(self.url, utils.SERVER_GROUP,
self.server_id, job_id),
content_type='html/json')
def api_put(self):
return self.tester.put('{0}{1}/{2}/{3}'.
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id),
data=json.dumps(self.data),
follow_redirects=True,
content_type='html/json')
def api_get_msql(self, url_encode_data):
return self.tester.get("{0}{1}/{2}/{3}?{4}".
format(self.url, utils.SERVER_GROUP,
self.server_id, self.job_id,
urlencode(url_encode_data)),
follow_redirects=True)
def api_delete(self, job_id=None):
if job_id is None:
job_id = self.job_id
return self.tester.delete('{0}{1}/{2}/{3}'.
format(self.url, utils.SERVER_GROUP,
self.server_id, job_id),
data=json.dumps(self.data),
content_type='html/json')
def is_valid_server_to_run_pgagent(self):
"""
@ -121,10 +174,12 @@ def create_pgagent_job(self, name):
traceback.print_exc(file=sys.stderr)
def delete_pgagent_job(self):
def delete_pgagent_job(self, job_id=None):
"""
This function deletes the pgAgent job.
"""
if job_id is None:
job_id = self.job_id
try:
connection = utils.get_db_connection(
self.server['db'],
@ -139,7 +194,7 @@ def delete_pgagent_job(self):
pg_cursor = connection.cursor()
pg_cursor.execute(
"DELETE FROM pgagent.pga_job "
"WHERE jobid = '%s'::integer;" % self.job_id
"WHERE jobid = '%s'::integer;" % job_id
)
connection.set_isolation_level(old_isolation_level)
connection.commit()

9
web/regression/runtests.py
View File

@ -681,10 +681,11 @@ def print_test_results():
failed_cases, failed_cases_json = test_utils.get_scenario_name(
failed_cases)
total_failed = len(dict((key, value) for key, value in
failed_cases.items()).values())
total_skipped = len(dict((key, value) for key, value in
skipped_cases.items()).values())
total_failed = sum(list((len(value)) for key, value in
failed_cases.items()))
total_skipped = sum(list((len(value)) for key, value in
skipped_cases.items()))
total_passed_cases = int(
test_result[server_res][0]) - total_failed - total_skipped