Smock testing on a small project: how it started and what results

    I am developing a project in C ++. I decided to try test
    scripts in Python on my project instead of manually testing the code. Usually this is not required from programmers in our company, so it was an experiment. He wrote about 100 tests in a year and this experiment turned out to be quite useful. Tests are performed for several minutes and allow you to quickly check both my pool requests and the pool requests of other developers.

    Prior to this experiment, I, as a developer, performed manual
    testing after adding a new feature . The fact that a programmer tested new features manually was not a problem in the company - at least in the groups where I worked, developers usually tested like that.

    From a design point of view, test scripts have a very simple organization. A class for each test plus several classes for modeling interacting programs. These classes are for modeling interacting programs and require writing time at the beginning. It took quite a lot of time to write the first test scripts. The task that can be done in 1 hour did 1 day. So the first few tests are time consuming. And in the future, on small improvements, more time is spent on writing a test than on a manual test. So not for every revision I did a test.

    However, for tasks with a long development, the ratio is already different. Once written
    automatic test saves time because it is used many times in the development process. For example, during the development of one task, 18 tests were written, and it was they that guaranteed the correctness of the algorithm, which consisted of C ++, Lua, SQL and used message exchange with RabbitMQ and work with the database.

    Since I did the tests for myself, I added a test run mode in which the test program does not start during testing, and the tests expect that the test program is already running. This gives me the opportunity to run a test when the program is running under the IDE and set breakpoints in the required places. This mode turned out to be convenient for debugging complex test cases.

    After six months of adding tests, when they were already several dozen and managed to get rid of false positives, tangible benefits for the project began to appear from them. I began to use them to quickly check the request pool of other developers. After code review, I ran the tests on the request pool branch. It took several minutes for the tests to work and it was clear if there were problems in the existing code - crashes or incorrect processing. As a result, I began to use these test scripts for testing smocks on the project.

    Single test example
    $ python3 autotests.py -c VirtualPaymentsDeleteWithShard             
    [ ========== ] Running 1 tests
    [ ========== ] autotest dir /home/sergey.kurenkov/src.git/dp.confirm_bfam/User_Part/build/autotests.dir
    [ RUN        ] BisrtAddon.VirtualPaymentsDeleteWithShard [test #1, time: 2017-07-31 18:09:05, test suite duration: 2.62]
    [         OK ] BisrtAddon.VirtualPaymentsDeleteWithShard [8.012 sec, time: 2017-07-31 18:09:13, test suite duration: 10.64]
    [ ========== ] 1 tests ran
    [ PASSED     ] 1 tests
    [            ] test suite duration (21.678 sec)
    


    Test example - test code in the run_in_test_env method
    class VirtualPaymentsBase(object):
        def __init__(self, autotest_cfg):
            self.autotest_cfg = autotest_cfg
            self.table_name = "virtual_payments"
            self.db_records = []
            self.rabbit_srv = None
            self.snmp_agent = None
            self.con = None
            self.cart_consumer = None
            self.pub = None
            self.test_env = None
            self.sent_cart_records = []
            self.sent_hrs_records = []
            self.sent_brt_records = []
            self.sent_bfam_records = []
            self.cart_consumer = None
            self.hrs_consumer = None
            self.brt_consumer = None
            self.bfam_consumer = None
            self.test_clnt_id = random.randint(1, 100000000)
        def test_name(self):
            raise NotImplementedError
        def publish_records(self):
            raise NotImplementedError
        def check_db_records(self):
            raise NotImplementedError
        def check_sent_cart_records(self):
            utility.check_number_of_records(self.sent_cart_records, 0)
        def expect_cart_records(self):
            return 0
        def check_sent_hrs_records(self):
            utility.check_number_of_records(self.sent_hrs_records, 0)
        def expect_hrs_records(self):
            return 0
        def check_sent_brt_records(self):
            raise NotImplementedError
        def expect_brt_records(self):
            raise NotImplementedError
        def check_sent_bfam_records(self):
            raise NotImplementedError
        def expect_bfam_records(self):
            raise NotImplementedError
        def db_records_has_been_fetched(self, db_records):
            return True if len(db_records) > 0 else False
        def prepare_db(self):
            raise NotImplementedError
        def on_finish(self):
            pass
        @utility.log_run
        def run_in_test_env(self, test_env):
            self.snmp_agent = test_env.snmp_agent
            self.con = test_env.con
            self.test_env = test_env
            self.pub = test_env.pub
            self.cart_consumer = test_env.cart_consumer
            self.hrs_consumer = test_env.hrs_consumer
            self.brt_consumer = test_env.brt_consumer
            self.bfam_consumer = test_env.bfam_consumer
            self.prepare_db()
            self.publish_records()
            self.db_records = fetch_table_records(partial(db_functions.fetch_virtual_payments,
                                                          clnt_id=self.test_clnt_id),
                                                  self.con, self.db_records_has_been_fetched)
            logging.info("checking db records")
            self.check_db_records()
            logging.info("checking cart records")
            self.sent_cart_records = self.cart_consumer.get_records(10, self.expect_cart_records())
            self.check_sent_cart_records()
            logging.info("checking brt records")
            self.sent_brt_records = self.brt_consumer.get_records(10, self.expect_brt_records())
            self.check_sent_brt_records()
            logging.info("checking hrs records")
            self.sent_hrs_records = self.hrs_consumer.get_records(10, self.expect_hrs_records())
            self.check_sent_hrs_records()
            logging.info("checking bfam records")
            self.sent_bfam_records = self.bfam_consumer.get_records(10, self.expect_bfam_records())
            self.check_sent_bfam_records()
            self.on_finish()
            logging.info("done")
    class VirtualPaymentsWithShard(VirtualPaymentsBase):
        def __init__(self, autotest_cfg):
            VirtualPaymentsBase.__init__(self, autotest_cfg)
            self.routing_key = "ps.ocsdb_tevt.virtual_payments.100"
            self.brt_routing_key = "ps.ocsdb.virtual_payments"
            self.bfam_routing_key = "ps.ocsdb_bfam.confirm_virt"
        def test_name(self):
            return "BisrtAddon.VirtualPaymentsWithShard"
        def prepare_db(self):
            cur = self.con.cursor()
            cur.execute("delete from virtual_payments t "
                        "where t.clnt_clnt_id = {clnt_id}".format(clnt_id=self.test_clnt_id))
            self.con.commit()
        def publish_records(self):
            record = {
                'last_record' : 1,
                'virt_id' : self.test_clnt_id,
                'vrtp_vrtp_id' : 1,
                'clnt_clnt_id' : self.test_clnt_id,
                'amount_r' : 123.4,
                'exp_date' : '20900102',
                'virtual_date' : '20690203',
                'amount_' :  12.3,
                'vrnt_vrnt_id' : 2,
                'vrct_vrct_id' : 3,
                'start_date' : '20160203',
                'end_date' : '20890405',
                'navi_date' : '20170405',
                }
            message_str = json.dumps([record], indent=4)
            logging.info(message_str)
            self.pub.publish(self.routing_key, message_str)
        def check_db_records(self):
            utility.check_number_of_records(self.db_records, 1)
            expected_recs = [(self.test_clnt_id,
                              1,
                              self.test_clnt_id,
                              123.4,
                              datetime(2090, 1, 2),
                              datetime(2069, 2, 3),
                              12.3,
                              None,
                              2,
                              None,
                              None,
                              None,
                              None,
                              None,
                              3,
                              datetime(2016, 2, 3),
                              datetime(2089, 4, 5),
                              datetime(2017, 4, 5),
                              None,
                              None,
                              None,
                              None,
                              None,
                              None,
                             )]
            compare_db_records(self.db_records, expected_recs)
        def expect_brt_records(self):
            return 1
        def check_sent_brt_records(self):
            utility.check_number_of_records(self.sent_brt_records, 1)
            a_message = self.sent_brt_records[0]
            check_message_routing_key(a_message, self.brt_routing_key)
            check_message_header_type(a_message, self.brt_routing_key)
            a_record = a_message['record']
            check_amqp_field(a_record, 'clnt_id', self.test_clnt_id)
            check_amqp_field(a_record, 'virt_id', self.test_clnt_id)
            check_amqp_field(a_record, 'vrtp_id', 1)
            check_amqp_field(a_record, 'vrct_id', 3)
            check_amqp_field_not_present(a_record, 'bltp_id')
            check_amqp_field_not_present(a_record, 'chrg_id')
            check_amqp_field(a_record, 'amount', 12.3)
            check_amqp_field(a_record, 'amount_r', 123.4)
            check_amqp_field(a_record, "start_date", '2016-02-03')
            check_amqp_field(a_record, "end_date", '2089-04-05')
            check_amqp_field(a_record, "deleted", False)
        def expect_bfam_records(self):
            return 1
        def check_sent_bfam_records(self):
            utility.check_number_of_records(self.sent_bfam_records, 1)
            a_message = self.sent_bfam_records[0]
            check_message_routing_key(a_message, self.bfam_routing_key)
            check_message_header_type(a_message, self.bfam_routing_key)
            a_record = a_message['record']
            utility.check_amqp_field(a_record, 'virt_id', self.test_clnt_id)
    


    Also popular now: