How to use execute_procedure method in Kiwi

Best Python code snippet using Kiwi_python

iceconnect.py

Source:iceconnect.py

...27            result = None28        finally:29            cursor.close()30        return result[0]31    def execute_procedure(self, proc_name, argument_list=None):32        result_list = list()33        cursor = self.conn.cursor()34        try:35            if argument_list:36                cursor.callproc(proc_name, argument_list)37            else:38                cursor.callproc(proc_name)39            self.conn.commit()40            for result in cursor.stored_results():41                result_list = [list(elem) for elem in result.fetchall()]42        except Error as e:43            self.status = e44        finally:45            cursor.close()46        return result_list47class CourseDB(DbConnector):48    def __init__(self):49        DbConnector.__init__(self)50    def add_course(self, c_numb, c_name, c_cred):51        new_id = 052        result = self.execute_procedure('NewCourse', [c_numb, c_name, c_cred])53        if result:54            new_id = int(result[0][0])55        return new_id56    def get_course(self, c_numb):57        result = self.execute_procedure('SingleCourse', [c_numb])58        if result:59            return result[0]60        else:61            return list()62    def get_restrictors(self):63        result = self.execute_procedure('courseRestrictorList')64        if result:65            return result66        else:67            return list()68    def number_of_courses(self):69        result = self.execute_function('NumberOfCourses')70        if result:71            return result[0]72        else:73            return list()74    def update_course(self, c_numb, c_name, c_cred):75        rows_affected = 076        result = self.execute_procedure('UpdateCourse', [c_numb, c_name, c_cred])77        if result:78            rows_affected = int(result[0][0])79        return rows_affected80    def delete_course(self, c_numb):81        rows_affected = 082        result = self.execute_procedure('DeleteCourse', [c_numb])83        if result:84            rows_affected = int(result[0][0])85        return rows_affected86class StudentDB(DbConnector):87    def __init__(self):88        DbConnector.__init__(self)89    def add_student(self, s_fname, s_lname, s_dob, s_ss):90        new_id = 091        result = self.execute_procedure('AddStudent', [s_fname, s_lname, s_dob, s_ss])92        if result:93            new_id = int(result[0][0])94        return new_id95    def get_student(self, s_id):96        result = self.execute_procedure('SingleStudentJSON', [s_id])97        if result:98            return result[0]99        else:100            return list()101    def student_creds(self, s_id):102        result = self.execute_procedure('studentCreds', [s_id])103        if result:104            return result[0]105        else:106            return list()107    def update_student(self, s_id, s_ss):108        rows_affected = 0109        result = self.execute_procedure('UpdateStudent', [s_id, s_ss])110        if result:111            rows_affected = int(result[0][0])112        return rows_affected113    def delete_student(self, s_id):114        rows_affected = 0115        result = self.execute_procedure('DeleteStudent', [s_id])116        if result:117            rows_affected = int(result[0][0])118        return rows_affected119class SchoolDB(DbConnector):120    def __init__(self):121        DbConnector.__init__(self)122    def add_school(self, school_name):123        new_id = 0124        result = self.execute_procedure('AddSchool', [school_name])125        if result:126            new_id = int(result[0][0])127        return new_id128    def get_school(self, school_id):129        result = self.execute_procedure('SingleSchool', [school_id])130        if result:131            return result[0]132        else:133            return list()134    def update_school(self, school_id, s_name):135        rows_affected = 0136        result = self.execute_procedure('UpdateSchool', [school_id, s_name])137        if result:138            rows_affected = int(result[0][0])139        return rows_affected140    def delete_school(self, school_id):141        rows_affected = 0142        result = self.execute_procedure('DeleteSchool', [school_id])143        if result:144            rows_affected = int(result[0][0])...

test_handlers.py

Source:test_handlers.py

...14    def test_html_escape(self):15        payload = "<html></html>"16        with patch(self.base_exec_procedure, return_value=payload):17            self.assertEqual(18                self.rpc_handler.execute_procedure("method_name"), html.escape(payload)19            )20    def test_timedelta_to_seconds(self):21        with patch(self.base_exec_procedure, return_value=timedelta(hours=1)):22            self.assertEqual(self.rpc_handler.execute_procedure("method_name"), 3600.0)23    def test_dict_escape(self):24        with patch(self.base_exec_procedure, return_value={"html": "<html></html>"}):25            self.assertDictEqual(26                self.rpc_handler.execute_procedure("method_name"),27                {"html": html.escape("<html></html>")},28            )29    def test_dict_with_timedelta(self):30        with patch(31            self.base_exec_procedure, return_value={"duration": timedelta(hours=1)}32        ):33            self.assertDictEqual(34                self.rpc_handler.execute_procedure("method_name"),35                {"duration": 3600.0},36            )37    def test_list_escape(self):38        with patch(self.base_exec_procedure, return_value=["<html></html>"]):39            self.assertListEqual(40                self.rpc_handler.execute_procedure("method_name"),41                [html.escape("<html></html>")],42            )43        with patch(self.base_exec_procedure, return_value=[{"html": "<html></html>"}]):44            self.assertListEqual(45                self.rpc_handler.execute_procedure("method_name"),46                [{"html": html.escape("<html></html>")}],47            )48    def test_list_with_timedelta(self):49        with patch(self.base_exec_procedure, return_value=[timedelta(hours=1)]):50            self.assertListEqual(51                self.rpc_handler.execute_procedure("method_name"),52                [3600.0],53            )54        with patch(55            self.base_exec_procedure, return_value=[{"duration": timedelta(hours=1)}]56        ):57            self.assertListEqual(58                self.rpc_handler.execute_procedure("method_name"),59                [{"duration": 3600.0}],60            )61class TestKiwiTCMSXmlRpcHandler(TestCase):62    @classmethod63    def setUpClass(cls):64        cls.base_exec_procedure = "modernrpc.handlers.XMLRPCHandler.execute_procedure"65        cls.rpc_handler = KiwiTCMSXmlRpcHandler(66            RequestFactory(), entry_point="/xml-rpc/"67        )68    def test_timedelta_to_seconds(self):69        with patch(self.base_exec_procedure, return_value=timedelta(hours=1)):70            self.assertEqual(self.rpc_handler.execute_procedure("method_name"), 3600.0)71    def test_dict_with_timedelta(self):72        with patch(73            self.base_exec_procedure, return_value={"duration": timedelta(hours=1)}74        ):75            self.assertDictEqual(76                self.rpc_handler.execute_procedure("method_name"),77                {"duration": 3600.0},78            )79    def test_list_with_timedelta(self):80        with patch(self.base_exec_procedure, return_value=[timedelta(hours=1)]):81            self.assertListEqual(82                self.rpc_handler.execute_procedure("method_name"),83                [3600.0],84            )85        with patch(86            self.base_exec_procedure, return_value=[{"duration": timedelta(hours=1)}]87        ):88            self.assertListEqual(89                self.rpc_handler.execute_procedure("method_name"),90                [{"duration": 3600.0}],...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.