Смок тестирование на небольшом проекте: как началось и какие результаты

в 13:46, , рубрики: testing, Тестирование IT-систем

Разрабатываю проект на C++. Решил попробовать на своем проекте тестовые сценарии
скриптовать на Python вместо того, чтобы тестировать код вручную. Обычно от программистов у нас в компании это не требуется, так что это был эксперимент. За год написал около 100 тестов и этот эксперимент оказался вполне полезным. Тесты выполняются несколько минут и позволяют быстро проверить как мои пул реквесты так и пул реквесты других разработчиков.

До этого эксперимента я, как разработчик, после добавления новой фичи выполнял ручное
тестирование. То, что тестирование программистом новых фич делалось вручную не было проблемой в компании — по крайней мере, в тех группах, где я работал, разработчики обычно так и тестировали.

С точки зрения проектирования, скрипты с тестами имеют очень простую организацию. Класс на каждый тест плюс несколько классы для моделирования взаимодействующих программ. Вот эти классы для моделирования взаимодействующих программ и требуют в начале время на написание. Времени на написание первых тестовых скриптов уходило достаточно много. Задачу, которую можно сделать за 1 час делал 1 день. Так что первые несколько тестов самые затратные по времени. Да и в дальнейшем на маленьких доработках на написание теста тратится больше времени, чем на ручной тест. Так что не на каждую доработку я делал тест.

Однако на задачах с длительной разработкой соотношение уже другое. Один раз написанный
автоматический тест дает экономию времени, поскольку используется много раз в процессе разработки. Например, в ходе разработки одной задачи был написан 18 тестов, и именно они гарантировали корректность алгоритма, который состоял из C++, Lua, SQL и использовал обмен сообщения с RabbitMQ и работу с БД.

Поскольку тесты делал для себя, то добавил режим запуска тестов в котором тестируемая программа не запускается при тестировании, а тесты ожидают, что тестируемая программа уже запущена. Это дает мне возможность выполнить тест когда программа запущена под IDE и установить брейкпоинты в требуемых местах. Этот режим оказался удобным для отладки сложных тестовых сценариев.

После полугода добавления тестов, когда их было было уже несколько десятков и удалось избавиться от ложных срабатываний, от них стала появляться ощутимая польза для проекта. Я стал использовать их для быстрой проверки пул реквестов других разработчиков. После code review выполнял прогон тестов на ветке пул реквеста. Несколько минут работы тестов и было ясно, есть ли проблемы в существующем уже коде — падения или неправильная обработка. В итоге эти тестовые скрипты я стал использовать для смок тестирования на проекте.

Пример выполнения отдельного теста

$ 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)

Пример теста - код теста в методе run_in_test_env

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)

Автор: sergei_kurenkov

Источник

Поделиться

* - обязательные к заполнению поля