KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_partial method in avocado

Best Python code snippet using avocado_python

test_pgpartialindex.py

Source:test_pgpartialindex.py

1from django.db import connection2from django.test import TestCase3from misago.threads.models import Thread4from misago.core.pgutils import PgPartialIndex5class PgPartialIndexTests(TestCase):6    def test_multiple_fields(self):7        """multiple fields are supported"""8        with connection.schema_editor() as editor:9            sql = PgPartialIndex(10                fields=['has_events', 'is_hidden'],11                name='test_partial',12                where={'has_events': True},13            ).create_sql(Thread, editor)14            self.assertIn('CREATE INDEX "test_partial" ON "misago_threads_thread"', sql)15            self.assertIn('ON "misago_threads_thread" ("has_events", "is_hidden")', sql)16    def test_where_clauses(self):17        """where clauses generate correctly"""18        with connection.schema_editor() as editor:19            sql = PgPartialIndex(20                fields=['has_events'],21                name='test_partial',22                where={'has_events': True},23            ).create_sql(Thread, editor)24            self.assertTrue(sql.endswith('WHERE "has_events" = true'))25            sql = PgPartialIndex(26                fields=['has_events'],27                name='test_partial',28                where={'has_events': False},29            ).create_sql(Thread, editor)30            self.assertTrue(sql.endswith('WHERE "has_events" = false'))31            sql = PgPartialIndex(32                fields=['has_events'],33                name='test_partial',34                where={'has_events': 42},35            ).create_sql(Thread, editor)36            self.assertTrue(sql.endswith('WHERE "has_events" = 42'))37            sql = PgPartialIndex(38                fields=['has_events'],39                name='test_partial',40                where={'has_events__lt': 42},41            ).create_sql(Thread, editor)42            self.assertTrue(sql.endswith('WHERE "has_events" < 42'))43            sql = PgPartialIndex(44                fields=['has_events'],45                name='test_partial',46                where={'has_events__gt': 42},47            ).create_sql(Thread, editor)48            self.assertTrue(sql.endswith('WHERE "has_events" > 42'))49            sql = PgPartialIndex(50                fields=['has_events'],51                name='test_partial',52                where={'has_events__lte': 42},53            ).create_sql(Thread, editor)54            self.assertTrue(sql.endswith('WHERE "has_events" <= 42'))55            sql = PgPartialIndex(56                fields=['has_events'],57                name='test_partial',58                where={'has_events__gte': 42},59            ).create_sql(Thread, editor)60            self.assertTrue(sql.endswith('WHERE "has_events" >= 42'))61    def test_multiple_where_clauses(self):62        """where clause with multiple conditions generates correctly"""63        with connection.schema_editor() as editor:64            sql = PgPartialIndex(65                fields=['has_events'],66                name='test_partial',67                where={68                    'has_events__gte': 42,69                    'is_hidden': True,70                },71            ).create_sql(Thread, editor)72            self.assertTrue(sql.endswith('WHERE "has_events" >= 42 AND "is_hidden" = true'))73    def test_set_name_with_model(self):74        """valid index name is autogenerated"""75        index = PgPartialIndex(76            fields=['has_events', 'is_hidden'],77            where={'has_events': True},78        )79        index.set_name_with_model(Thread)80        self.assertEqual(index.name, 'misago_thre_has_eve_1b05b8_part')81        index = PgPartialIndex(82            fields=['has_events', 'is_hidden', 'is_closed'],83            where={'has_events': True},84        )85        index.set_name_with_model(Thread)86        self.assertEqual(index.name, 'misago_thre_has_eve_eaab5e_part')87        index = PgPartialIndex(88            fields=['has_events', 'is_hidden', 'is_closed'],89            where={90                'has_events': True,91                'is_closed': False,92            },93        )94        index.set_name_with_model(Thread)95        self.assertEqual(index.name, 'misago_thre_has_eve_e738fe_part')96    def test_index_repr(self):97        """index creates descriptive representation string"""98        index = PgPartialIndex(99            fields=['has_events'],100            where={'has_events': True},101        )102        self.assertEqual(repr(index), "<PgPartialIndex: fields='has_events', where='has_events=True'>")103        index = PgPartialIndex(104            fields=['has_events', 'is_hidden'],105            where={'has_events': True},106        )107        self.assertIn("fields='has_events, is_hidden',", repr(index))108        self.assertIn(", where='has_events=True'", repr(index))109        index = PgPartialIndex(110            fields=['has_events', 'is_hidden', 'is_closed'],111            where={112                'has_events': True,113                'is_closed': False,114                'replies__gte': 5,115            },116        )117        self.assertIn("fields='has_events, is_hidden, is_closed',", repr(index))...

benchmark.py

Source:benchmark.py

1import sys2import scoop3import time4import argparse5import logging6import random7from serialization import find_pickling_speed8from functools import partial9def make_parser():10    parser = argparse.ArgumentParser(11        description=('Run a parametric benchmark of scoop.')12    )13    parser.add_argument('--time', type = float, default = 5.0,14                        help = "The mean time of each individual task")15    parser.add_argument('--serialization-time', type = float, default = 0.01,16                        help = "The mean serialization time for each task")17    parser.add_argument('--tries', type = int, default = 10,18                        help = ("The number of functions sent to the workers "19                                "for each level of the hierchy"))20    parser.add_argument('--log', help = ("A filename to log the output "21                                        "(optional). This is different than the"22                                        'scoop "--log" option'))23    parser.add_argument('--level', help = "Number of level in the hierarchy",24                        type = int, default = 2)25    return parser26def print_header(args):27    header = ("-------------------------------------------------\n"28              "Benchmarking using these parameters:\n"29              "tries: {0.tries}^{0.level} = {1}\n"30              "time: {0.time} s\n"31              "serialization time: {0.serialization_time} s\n"32              "SCOOP Parameters:\n"33              "number of workers: {2} workers\n"34              "number of brokers: {3} brokers\n"35              "SCOOP version: {4}\n"36              "Python version {5}\n"37              "-------------------------------------------------\n")38    header = header.format(args, args.tries ** args.level,39            scoop.SIZE, 1, scoop.__version__ + scoop.__revision__,40            sys.version)41    if args.log:42        with open(args.log, 'a') as f:43            f.write(header)44    else:45        print(header)46def test_function(_fake_data, cpu_time = 3.0, level = 0, number_of_tests = 1):47    start_time = time.time()48    test_partial = partial(test_function, number_of_tests = number_of_tests,49                           level = level - 1, cpu_time = cpu_time)50    test_partial.__name__ = "test_partial"51    total = 052    number_of_times = 053    while time.time() - start_time < cpu_time:54        total += random.random()55        number_of_times += 156    if level <= 1:57        if number_of_times != 0:58            return total / number_of_times59        else:60            return 0.561    else:62        test_data = (_fake_data for _ in range(number_of_tests))63        children = scoop.futures.map(test_partial, test_data)64        return sum(children) / number_of_tests65def test(number_of_tests, cpu_time, serialization_time, log, levels):66    test_partial = partial(test_function, number_of_tests = number_of_tests,67                           level = levels, cpu_time = cpu_time)68    test_partial.__name__ = "test_partial"69    fake_data_len = find_serialization_time(serialization_time, log)70    fake_data = [random.random() for _ in range(fake_data_len)]71    send_data = (fake_data for _ in range(number_of_tests))72    begin_time = time.time()73    result = list(scoop.futures.map(test_partial, send_data))74    total_time = time.time() - begin_time75    return result, total_time76def find_serialization_time(wanted_time, log):77    speed = find_pickling_speed(500, 14)78    if log:79        with open(log, 'a') as f:80            f.write("The pickling speed is {:g} bytes/s.\n".format(1/speed))81    else:82        print("The pickling speed is {:g} bytes/s.".format(1/speed))83    return int(wanted_time / speed)84if __name__ == "__main__":85    args = make_parser().parse_args()86    print_header(args)87    start_time = time.time()88    result, test_time = test(args.tries, args.time,89                             args.serialization_time, args.log,90                             args.level)91    end_time = time.time()92    if args.log:93        with open(args.log, 'a') as f:94            f.write("Total time: {}\n".format(end_time - start_time))95            f.write("Test time: {}\n".format(test_time))96    else:97        print("Total time: {}".format(end_time - start_time))...

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.