31.07.2017 13:46
sergei_kurenkov 6 2700 Источник
Разрабатываю проект на 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)

Поделиться с друзьями
-->

Комментарии (6)


  1. sergei_kurenkov
    31.07.2017 18:17
    #10337078

    Добавил в пост пример теста


  1. Eugene_Y
    01.08.2017 08:19
    #10337658

    Привет! А можно примеры тестов?


    1. sergei_kurenkov
      01.08.2017 08:28
      #10337672

      Добавил в пост пример теста


  1. shockable
    01.08.2017 08:19
    #10337660

    Скажите, пожалуйста, почему решили не использовать имеющиеся библиотеки для юнит-тестирования(PyUnit, PyTest etc.)?


    1. sergei_kurenkov
      01.08.2017 08:28
      #10337670

      Мы на самом деле делаем юнит-тестирования нашего C++ кода с использованием GoogleTest. Я об этом рассказывал тут.

      Но здесь я рассказал про смок тестирование. Вот тут определение смок тестирования:

      Smoke tests, in which engineers test very simple but critical behavior, are among the simplest type of system tests. Smoke tests are also known as sanity testing, and serve to short-circuit additional and more expensive testing.


      Для смок тестирования я рассматривал использование Robot Framework, но решил, что для меня удобно будет просто писать тесты на Python.


  1. shockable
    01.08.2017 08:41
    #10337676

    Просто могли бы переиспользовать готовую функциональность, например, error handling, reporting, test suites, setup/tear down etc. В самом простом варианте можно всего лишь отнаследоваться от класса unittest.TestCase и сразу получить неплохой набор готовых решений рутины. Если будет время — рекомендую: http://gahcep.github.io/blog/2013/02/10/qa-in-python-unittest/