In my previous post, I listed several PL/SQL unit-testing frameworks. While it is my intent to evaluate all – or at least most – of these, I’m starting with Python.
I first encountered Python a year or so ago while researching Hadoop – Python is one of the languages used in the provided samples. Since then, I’ve run across Python more and more frequently. Various teams at my work are now using it for tasks such as Financial engineering, QA automation, etc. At first glance, it seems easier to learn than Java, especially now; Java has matured into a very complex ecosystem with a pretty steep learning curve when you factor in all the third-party modules, API’s and plug-ins. Python seemed to me to be a better way to go – a modern language with many interesting and useful features and still a deep and comprehensive library of add-ons.
I’m not going to try to cover all of Python in this blog; just the parts that have proven useful for my unit-testing evaluations. If interested, there are many fine books and internet resources available on this topic. I started with this one. This one is frequently mentioned online as a definitive work but I’ve never looked at it. Also, this book gives a good overview of PyTest, the unit-test framework I used.
Approach
For this blog, I chose to first create a small PL/SQL package in Oracle that would insert and/or update data in one of the sample Oracle schemas. I will be demonstrating how to unit-test this package with Python.
Here’s the package specification:
create or replace PACKAGE order_management AS /* 1. Create Order a. Verify customer exists b. Verify Promotion exists c. Verify Sales Rep exists d. Verify Order Id is unique 2. Create Line Items a. Verify Order Id exists b. Verify Product exists c. Verify Line Item Id is unique within order 3. Modify Line Item a. Verify Order & Line Item exist b. Verify Product exists, if changing 4. Delete Line Item a. Verify Line Item exists b. Override flag to delete order if last line item 5. Delete Order a. Verify Order exists b. Override to delete if line items present */ vresult INTEGER := 0; vamount NUMBER; invalid_customer EXCEPTION; invalid_promotion EXCEPTION; invalid_sales_rep EXCEPTION; invalid_order EXCEPTION; invalid_order_item EXCEPTION; invalid_product EXCEPTION; line_items_exist EXCEPTION; PROCEDURE create_order ( p_order_id IN oe.orders.order_id%TYPE, p_order_date IN oe.orders.order_date%TYPE, p_order_mode IN oe.orders.order_mode%TYPE, p_customer_id IN oe.orders.customer_id%TYPE, p_order_status IN oe.orders.order_status%TYPE, p_sales_rep_id IN oe.orders.sales_rep_id%TYPE, p_promotion_id IN oe.orders.promotion_id%TYPE ); PROCEDURE create_order_item ( p_order_id IN oe.order_items.order_id%TYPE, p_line_item_id IN oe.order_items.line_item_id%TYPE, p_product_id IN oe.order_items.product_id%TYPE, p_unit_price IN oe.order_items.unit_price%TYPE, p_quantity IN oe.order_items.quantity%TYPE ); PROCEDURE modify_order_item ( p_order_id IN oe.order_items.order_id%TYPE, p_line_item_id IN oe.order_items.line_item_id%TYPE, p_product_id IN oe.order_items.product_id%TYPE, p_unit_price IN oe.order_items.unit_price%TYPE, p_quantity IN oe.order_items.quantity%TYPE ); PROCEDURE delete_order_item ( p_order_id IN oe.order_items.order_id%TYPE, p_line_item_id IN oe.order_items.line_item_id%TYPE, p_delete_order IN BOOLEAN DEFAULT false ); PROCEDURE delete_order ( p_order_id IN oe.order_items.order_id%TYPE, p_delete_order IN BOOLEAN DEFAULT false ); END;
For today, I’ll review a small PyTest script that runs a single unit-test on the create_order procedure.
Before we begin, let’s take a quick look at the structure of a Python – and in particular a PyTest – script.
1. import datetime 2. import cx_oracle 3. @pytest.fixture(scope='session') 4. def openDb(): 5. """Generic database connection fixture.""" 6. <Setup code goes here> 7. yield 8. <Teardown code goes here>
- Lines 1-2: imports – import any external modules
- Line 3: Fixture annotation – fixtures enable reuse of setup and teardown code. These are a PyTest feature.
- Line 4: Function definition –
openDbis the function name, any parameters are defined within the parentheses. Note also the code within the function is indented by 4 spaces – this is a Python requirement. It’s also applied to loops and if statements - Line 5: Function description – this will appear in the Python documentation
- Line 6: Setup – Any setup code this fixture executes is defined here
- Line 7: Yield – When this line is executed, control is passed back to the calling function
- Line 8: Teardown – After the calling function has completed, the teardown code here is executed
So, here’s the first unit-test script. It is executing a test to confirm that the create_order procedure creates a new order record and that all the columns contain the expected values. I’ll do this in sections with notes after each.
First, the imports and title:
1. import cx_Oracle 2. import pytest 3. import datetime 4. import pandas as pd 5. """Test procedures in the the order_management package."""
- Line 1:
import cx_Oracle– this is the Python module provided by Oracle to enable database connections from Python scripts. It has similar features to the ojdbc JDBC module - Line 2:
import pytest– the unit-testing framework - Line 3:
import datetime– a utility module for working with dates and timestamps - Line 4:
import pandas as pd– pandas is a module that provides many functions for working with data in array-like structures named DataFrames. Theas pdpart defines an alias for the module that is used in the subsequent functions - Line 5: script title
Next the fixtures. Note: there are several fixtures that make use of one or more other fixtures using the same parameter definition syntax as for regular function definitions:
1. @pytest.fixture(scope='session')
2. def openDb():
3. """Generic database connection fixture."""
4. params = pd.read_json("./parameters.json")
5. dbconnectionstring = params.connection.hostname + ':' + params.connection.port + '/' + params.connection.service
6. db = cx_Oracle.connect(params.connection.account, params.connection.password, dbconnectionstring)
7. cursor1 = db.cursor()
8. yield cursor1
9. cursor1.close()
10. db.close()
11. @pytest.fixture(scope='session')
12. def add_customer(openDb):
12. openDb.execute("""
13. insert into oe.customers (customer_id, cust_first_name, cust_last_name)
values (100, 'Customer', 'NumberOne')""")
14. yield
15. openDb.execute("""
16. delete from oe.customers where customer_id = 100""")
- Line 1: The fixture annotation. Note the scope=’session’ instruction. There are three possible scopes for a fixture: function, module and session. This controls whether a fixture is run for every function or just once for a module, i.e. class, or session. In this script, it will only be run once for the entire test session
- Line 2:
def openDb():This is the function definition - Line 4:
params = pd.read_json("./parameters.json")Here, we’re creating a DataFrame object (params) by reading a JSON format parameter file using theread_jsonfunction in pandas - Line 5:
dbconnectionstring– construct a string object using the parameters previously loaded - Line 6: open the database connection
- Line 7: create a cursor object.
- Line 8: pass the cursor object handle back to the calling function via the yield command
- Lines 9-10: close the cursor and db connection once all functions have completed
- Line 12:
def add_customer(openDb)– Define a new fixture function, this time referencing theopenDbfunction as a parameter - Lines 12-13: Insert customer. We insert this customer in a setup fixture in order to use it in a test function
- Lines 15-16: once all dependent functions have completed the customer is deleted
Continuing with more fixtures:
17. @pytest.fixture(scope='session')
18. def add_product(openDb):
19. openDb.execute("""
20. insert into oe.product_information (product_id)
21. values (100)""")
22. yield
23. openDb.execute("""
24. delete from oe.product_information where product_id = 100""")
25. @pytest.fixture(scope='session')
26. def add_sales_rep(openDb):
27. hiredt = datetime.date.today()
28. openDb.execute("""
29. insert into hr.employees (employee_id, last_name, email, hire_date, job_id, department_id)
30. values (50, 'Employee', 'employee@company.com', :hire_dt, 'SA_REP', 80)""",
40. hire_dt = hiredt)
41. yield
42. openDb.execute("""
43. delete from hr.employees where employee_id = 50""")
- Lines 17-24: create product fixture
- Lines 25-43: create sales rep fixture
Last fixture:
44. @pytest.fixture(scope='session')
45. def add_promotion(openDb):
46. openDb.execute("""
47. insert into oe.promotions (promo_id, promo_name)
48. values (50, 'Promotion1')""")
49. yield
50. openDb.execute("""
51. delete from oe.promotions where promo_id = 50"""
- Lines 44-51: create promotion
Now the actual test code:
1. def test_create_order(openDb,add_customer,add_product,add_promotion,add_sales_rep):
2. """Call create_order with specified inputs and test actual result matches expected."""
3. order = {
4. "ORDER_ID": [1],
5. "ORDER_DATE": ['01-JUN-10'],
6. "ORDER_MODE": ['direct'],
7. "CUSTOMER_ID": [100],
8. "ORDER_STATUS": [1],
9. "SALES_REP_ID": [50],
10. "PROMOTION_ID": [50]
11. }
12. expected_results = pd.DataFrame.from_dict(order)
13. expected_results.sort_index(axis=1, inplace=True)
14. execute_proc = openDb.callproc('order_management.create_order', [
15. order["ORDER_ID"][0],
16. order["ORDER_DATE"][0],
17. order["ORDER_MODE"][0],
18. order["CUSTOMER_ID"][0],
19. order["ORDER_STATUS"][0],
20. order["SALES_REP_ID"][0],
21. order["PROMOTION_ID"][0]
22. ])
- Line 1: For pytest, the test function must either begin or end with ‘test’. All the required fixtures are called as parameters
- Line 3: creates a dictionary object named
order– a dictionary is a special object type in Python. “A dictionary maps keys to values. Keys need to be hashable objects, while values can be of any arbitrary type” (Learning Python). - Line 12; creates the
expected_resultsDataFrame from theorderdictionary - Line 13: sorts the columns alphabetically in the
expected_resultsobject - Line 14: use the openDb connection to call the procedure being tested. If successful, this procedure call will create a new order record in the
oe.orderstable. Note the references to the order dictionary. The first pair of brackets denote the column and the second the row, zero reference
23. openDb.execute("""
24. select customer_id, to_char(order_date, 'DD-MON-RR') order_date, order_id, order_mode, order_status,
25. promotion_id, sales_rep_id from oe.orders where order_id = 1""")
26. compare_status = True
27. dbdata = pd.DataFrame(openDb.fetchall())
28. cols = [i[0] for i in openDb.description]
29. dbdata.columns = cols
30. if openDb.rowcount > 0:
31. for column in expected_results.columns:
32. if (dbdata[column] != expected_results[column]).all():
33. compare_status = False
34. else:
35. compare_status = False
36. assert compare_status
37. execute_proc = openDb.callproc('eric.order_management.delete_order', [
38. order["ORDER_ID"][0]
39. ])
- Line 23: execute a query using the same db connection to obtain the data for the newly created order
- Line 26: create a boolean defaulting to True
- Line 27: create a DataFrame object named
dbdatafrom the query result set - Line 28: extract the column headers from the resultset
- Line 29: set the
dbdatacolumn headers to those extracted from the resultset - Line 30: If data is returned from the database:
- Line 31: loop through the DataFrame data
- Line 32: check if any data returned from the database does not match the expected results, column-by-column
- Line 33: if any mismatches are found set the boolean to False
- Line 35: if no rows are returned, set the boolean to false
- Line 36: assert the boolean is true. If it is, the test passes; if not, it fails
- Line 37: delete the new order. Note: this has to be done here since the order ID is returned by the procedure call; there’s no way for any of the fixtures to know what it is
Thanks to the use of fixtures, the actual test code is very concise and easy to follow.
Database Engineering 