Pick a complement of tests that exercises enough parts to define a thread of execution. This is sometimes referred to as thread testing. Not because we are using software threading, but instead, because we are focusing on a story thread. Often, our threads either come from customer scenarios or are at least inspired by them. Other threads can involve other groups of operations.
For example, a network management system may push out customer-affecting alarms, but the internal operations team that has to solve the network problems may have a totally different perspective. Both of these situations demonstrate valid end-to-end threads that are good places to invest in automated testing.
- Copy the SQL script from Chapter 7, Creating a network management application into a new file called
recipe61_network.sqland replace the insert statements at the bottom with the following:INSERT into EQUIPMENT (ID, HOST_NAME, STATUS) values (1, 'pyhost1', 1); INSERT into EQUIPMENT (ID, HOST_NAME, STATUS) values (2, 'pyhost2', 1); INSERT into EQUIPMENT (ID, HOST_NAME, STATUS) values (3, 'pyhost3', 1); INSERT into SERVICE (ID, NAME, STATUS) values (1, 'service-abc', 'Operational'), INSERT into SERVICE_MAPPING (SERVICE_FK, EQUIPMENT_FK) values (1,1); INSERT into SERVICE_MAPPING (SERVICE_FK, EQUIPMENT_FK) values (1,2);
In this set of test data,
pyhost1andpyhost2map intoservice-abc. However,pyhost3doesn't map into any service.
With these steps, we will build up an end-to-end test scenario.
- Create a test module called
recipe61_test.py. - Create a test case where each test method captures a different thread of execution.
import logging from network import * import unittest from springpython.database.factory import * from springpython.database.core import * class EventCorrelatorEquipmentThreadTests(unittest.TestCase): def setUp(self): db_name = "recipe61.db" factory = Sqlite3ConnectionFactory(db=db_name) self.correlator = EventCorrelator(factory) dt = DatabaseTemplate(factory) sql = open("recipe61_network.sql").read().split(";") for statement in sql: dt.execute(statement + ";") def tearDown(self): self.correlator = None - Create a test method that captures the thread of failing and recovering a piece of equipment.
def test_equipment_failing(self): # This alarm maps to a device # but doesn't map to any service. - Have the test method inject a single, faulting alarm and then confirm that a related piece of equipment has failed.
evt1 = Event("pyhost3", "serverRestart", 5) stored_event, is_active, updated_services, updated_equipment = self.correlator.process(evt1) self.assertTrue(is_active) self.assertEquals(len(updated_services), 0) self.assertEquals(len(updated_equipment), 1) self.assertEquals(updated_equipment[0]["HOST_NAME"], "pyhost3") # 5 is the value for a failed piece of equipment self.assertEquals(updated_equipment[0]["STATUS"], 5) - In the same test method, add code that injects a single, clearing alarm and confirms that the equipment has recovered.
evt2 = Event("pyhost3", "serverRestart", 1) stored_event, is_active, updated_services, updated_equipment = self.correlator.process(evt2) self.assertFalse(is_active) self.assertEquals(len(updated_services), 0) self.assertEquals(len(updated_equipment), 1) self.assertEquals(updated_equipment[0]["HOST_NAME"], "pyhost3") # 1 is the value for a clear piece of equipment self.assertEquals(updated_equipment[0]["STATUS"], 1) - Create another test method that captures the thread of failing and clearing a service.
def test_service_failing(self): # This alarm maps to a service. - Write a test method that injects a single, faulting alarm and confirms that both a piece of equipment and a related service fails.
evt1 = Event("pyhost1", "serverRestart", 5) stored_event, is_active, updated_services, updated_equipment = self.correlator.process(evt1) self.assertEquals(len(updated_services), 1) self.assertEquals("service-abc", updated_services[0]["service"]["NAME"]) self.assertEquals("Outage", updated_services[0]["service"]["STATUS"]) - In the same test method, add code that injects a single, clearing alarm and confirms that both the equipment and the service have recovered.
evt2 = Event("pyhost1", "serverRestart", 1) stored_event, is_active, updated_services, updated_equipment = self.correlator.process(evt2) self.assertEquals(len(updated_services), 1) self.assertEquals("service-abc", updated_services[0]["service"]["NAME"]) self.assertEquals("Operational", updated_services[0]["service"]["STATUS"]) - Create a test runner called
recipe61.pythat imports both of these thread tests.from recipe61_test import * if __name__ == "__main__": import unittest unittest.main() - Run the test suite.
In this recipe we coded two end-to-end test scenarios:
- The first scenario tested how our application processes a fault followed by a clear that only impacts a piece of equipment
- The second scenario tested how our application processes a fault followed by a clear that impacts a service
We injected a fault and then checked the results to confirm the proper piece of inventory failed. Then we injected a clear and again confirmed that the proper piece of inventory recovered.
Both of these scenarios show how our application processes different types of events from the beginning to the end.
In a more complex, realistic version of this application, what other systems do you think would be involved in an end-to-end thread? What about security? Transactions? Publishing results to an external interface?
This is where we need to define where the ends are. Imagine that our application was grown to the point where incoming events are received by a web request and equipment and service updates are pushed out as JSON data to be received by a web page.
A good end-to-end test would include these parts as well. For the JSON output, we can use Python's JSON library to decode the output and then confirm the results. For the incoming web request, we can use many different techniques including acceptance testing tools like the Robot Framework.