How to use test_run_once method in tavern

Best Python code snippet using tavern

Entry.py

Source:Entry.py

...14        if measure_rw_from_db:15            return data.show_sum_of_reads_and_writes(), phase_count16        else:17            return Validator.validate(['tape0.txt', 'tape1.txt', 'tape2.txt'])18    def test_run_once(self, max_record_length: int, amount_of_records: int,19                      file_to_generate: str, block_size: int, logger_config: dict,20                      measure_rw_from_db=False, **kwargs):21        log = Logger(dbg_variables=logger_config)22        Helpers.erase_files(['tape0.txt', 'tape1.txt', 'tape2.txt', file_to_generate])23        Helpers.generate(amount_of_records, max_record_length, file_to_generate)24        return self.run_test(input_file=file_to_generate,25                             log=log,26                             block_size=block_size,27                             measure_rw_from_db=measure_rw_from_db)28    def test_run_once_with_kb_input(self, max_record_length: int, file_to_generate: str,29                                    block_size: int, logger_config: dict, measure_rw_from_db=False,30                                    **kwargs):31        log = Logger(dbg_variables=logger_config)32        Helpers.erase_files(['tape0.txt', 'tape1.txt', 'tape2.txt', file_to_generate])33        if not self.insert_data(file_to_generate, max_record_length):34            return35        return self.run_test(input_file=file_to_generate,36                             log=log,37                             block_size=block_size,38                             measure_rw_from_db=measure_rw_from_db)39    def test_run_once_with_file_input(self, file_to_generate: str, block_size: int, logger_config: dict,40                                      measure_rw_from_db=False, **kwargs):41        log = Logger(dbg_variables=logger_config)42        Helpers.erase_files(['tape0.txt', 'tape1.txt', 'tape2.txt'])43        return self.run_test(input_file=file_to_generate,44                             log=log,45                             block_size=block_size,46                             measure_rw_from_db=measure_rw_from_db)47    def insert_data(self, file_to_generate: str, max_record_length: int):48        data = []49        print('Press q to exit')50        while True:51            input_v = input('Pass record value [a-zA-Z0-9]: ')52            if input_v == 'q':53                if str(input('Are you sure you want to exit? [y/n]\n'54                             'If no single character q will be added as record\t')) == 'y':55                    break56            while not input_v:57                input_v = input('Can\'t pass empty record!')58            while len(input_v) > max_record_length:59                input_v = input(f'Passed record is too long. Max length is {max_record_length}')60            data.append(str(input_v))61        with open(file_to_generate, 'r+') as f:62            for record in data[:-1]:63                f.write(f'{record}\n')64            f.write(str(data[-1]))65        if not len(data):66            print('Sorting empty list?')67            return False68        return True69    def test_run_multiple(self, max_record_length: int, amount_of_records: int,70                          file_to_generate: str, block_size: int,71                          logger_config: dict, amount_of_tests: int):72        passed = True73        while passed and amount_of_tests:74            amount_of_tests -= 175            passed = self.test_run_once(max_record_length=max_record_length,76                                        amount_of_records=amount_of_records,77                                        file_to_generate=file_to_generate,78                                        block_size=block_size,79                                        logger_config=logger_config)80        if passed:81            print(f'Every test passed')82        else:83            print(f'{amount_of_tests} tests left. Last test data saved to {file_to_generate}')84class TextInterface:85    def __init__(self):86        self.current_status = ''87        self.entry_class = Entry()88        self.params = {89            'max_record_length': 30,...

test_calvin.py

Source:test_calvin.py

...28    subprocess.run(['./propogate_settings.sh',str(nodes),str(port)])29    rename_portfile(port)30    time.sleep(3)31    print("Build finished, start running...")32def test_run_once(nodes,per_distr,wh,t_out=90):33    print("|Config|nnodes:{}|dist:{}|WH:{}".format(nodes,per_distr,wh))34    try:35        # rename_portfile()36        subprocess.run(37        [   './bin/deployment/cluster', 38            "-c", "./deploy-run.conf",39            '-p','./src/deployment/portfile',40            '-d','./bin/deployment/db',41            't',42            str(per_distr),43            # '2>&1','|','tee','-a','./test_{}'.format(datetime.now().strftime('%Y-%m-%d-%H-%M-%S'))44            ], timeout=t_out45        )46    except subprocess.TimeoutExpired:47        print('Timeout! Starting new run...')48def rename_portfile(port):49    with open('./src/deployment/portfile','w') as f:50        f.write(str(port))51    #time.sleep(3)52def gen_port_list(test_list):53    length = sum([len(dist) for node_wh, dist in test_list])54    # currently we assume the number of experiment won't exceed 500055    # port_list = [ 50000 + 2*x for x in random.sample(range(length), length)]56    port_list = random.sample(range(50000, 60000), length)57    return port_list58if __name__ == "__main__":59    # for nodes in [1,2,4,8,16,32]:60    #     for warehouse in [1,2,4,8,16,32,64,128]:61    #         update_settings(nodes,warehouse)62    #         for dist in [0,10,20,30,40,50,60,70,80,90,100]:63    #             print(datetime.now())64    #             test_run_once(nodes,dist,warehouse)65    66    #for nodes in [32]:67    #    for warehouse in [8]:68    #        update_settings(nodes,warehouse)69    #        for dist in [0]:70    #            test_run_once(nodes,dist,warehouse)71    # test with abnormal configs72    # list: [(node, wh), [dist1, dist2, ..], ...]73    # test_list = [[(4, 4), [70]]]74    # test_list = [[(16,4), [90, 100]], [(4, 4), [70]], [(32,4), [80, 90, 100]]]75    #test_list = [[(16,4), [60, 70, 80, 90, 100]], [(4, 4), [50, 60, 70, 80, 90]], [(32,4), [50, 60, 70, 80, 90, 100]]]76    #test_list = [[(32,4), [70, 90]]]77    #test_list = [[(32,32), [60, 70]]]78    test_list = [[(32,32), [90]]]79    port_list = gen_port_list(test_list)80    index = 081    for elem in test_list:82        nodes = elem[0][0]83        wh = elem[0][1]84        #update_settings(nodes, wh)85        for dist in elem[1]:86            port = port_list[index]87            index = index+188            print("Run exp with setting[node:{0}, wh:{1}, dist:{2}, port:{3}]".format(nodes, wh, dist, port))89            update_settings(nodes, wh, port)90            test_run_once(nodes, dist, wh)91            92            ...

testsample.py

Source:testsample.py

1import unittest2def f(n):3    return n**24class Test(unittest.TestCase):5    def test_run_once(self):6        self.assertEqual(f(2), 4)7        self.assertEqual(f(3), 9)8class Test2(unittest.TestCase):9    def test_run_once(self):10        self.assertEqual(f(5), 25)11        self.assertEqual(f(6), 9)12if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.