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

How to use fixture_name method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

loaddata.py

Source:loaddata.py

1from __future__ import unicode_literals2import glob3import gzip4import os5import warnings6import zipfile7from itertools import product8from django.apps import apps9from django.conf import settings10from django.core import serializers11from django.core.exceptions import ImproperlyConfigured12from django.core.management.base import BaseCommand, CommandError13from django.core.management.color import no_style14from django.db import (15    DEFAULT_DB_ALIAS, DatabaseError, IntegrityError, connections, router,16    transaction,17)18from django.utils import lru_cache19from django.utils._os import upath20from django.utils.encoding import force_text21from django.utils.functional import cached_property22from django.utils.glob import glob_escape23try:24    import bz225    has_bz2 = True26except ImportError:27    has_bz2 = False28class Command(BaseCommand):29    help = 'Installs the named fixture(s) in the database.'30    missing_args_message = ("No database fixture specified. Please provide the "31                            "path of at least one fixture in the command line.")32    def add_arguments(self, parser):33        parser.add_argument('args', metavar='fixture', nargs='+',34            help='Fixture labels.')35        parser.add_argument('--database', action='store', dest='database',36            default=DEFAULT_DB_ALIAS, help='Nominates a specific database to load '37            'fixtures into. Defaults to the "default" database.')38        parser.add_argument('--app', action='store', dest='app_label',39            default=None, help='Only look for fixtures in the specified app.')40        parser.add_argument('--ignorenonexistent', '-i', action='store_true',41            dest='ignore', default=False,42            help='Ignores entries in the serialized data for fields that do not '43            'currently exist on the model.')44    def handle(self, *fixture_labels, **options):45        self.ignore = options.get('ignore')46        self.using = options.get('database')47        self.app_label = options.get('app_label')48        self.hide_empty = options.get('hide_empty', False)49        self.verbosity = options.get('verbosity')50        with transaction.atomic(using=self.using):51            self.loaddata(fixture_labels)52        # Close the DB connection -- unless we're still in a transaction. This53        # is required as a workaround for an  edge case in MySQL: if the same54        # connection is used to create tables, load data, and query, the query55        # can return incorrect results. See Django #7572, MySQL #37735.56        if transaction.get_autocommit(self.using):57            connections[self.using].close()58    def loaddata(self, fixture_labels):59        connection = connections[self.using]60        # Keep a count of the installed objects and fixtures61        self.fixture_count = 062        self.loaded_object_count = 063        self.fixture_object_count = 064        self.models = set()65        self.serialization_formats = serializers.get_public_serializer_formats()66        # Forcing binary mode may be revisited after dropping Python 2 support (see #22399)67        self.compression_formats = {68            None: (open, 'rb'),69            'gz': (gzip.GzipFile, 'rb'),70            'zip': (SingleZipReader, 'r'),71        }72        if has_bz2:73            self.compression_formats['bz2'] = (bz2.BZ2File, 'r')74        # Django's test suite repeatedly tries to load initial_data fixtures75        # from apps that don't have any fixtures. Because disabling constraint76        # checks can be expensive on some database (especially MSSQL), bail77        # out early if no fixtures are found.78        for fixture_label in fixture_labels:79            if self.find_fixtures(fixture_label):80                break81        else:82            return83        with connection.constraint_checks_disabled():84            for fixture_label in fixture_labels:85                self.load_label(fixture_label)86        # Since we disabled constraint checks, we must manually check for87        # any invalid keys that might have been added88        table_names = [model._meta.db_table for model in self.models]89        try:90            connection.check_constraints(table_names=table_names)91        except Exception as e:92            e.args = ("Problem installing fixtures: %s" % e,)93            raise94        # If we found even one object in a fixture, we need to reset the95        # database sequences.96        if self.loaded_object_count > 0:97            sequence_sql = connection.ops.sequence_reset_sql(no_style(), self.models)98            if sequence_sql:99                if self.verbosity >= 2:100                    self.stdout.write("Resetting sequences\n")101                with connection.cursor() as cursor:102                    for line in sequence_sql:103                        cursor.execute(line)104        if self.verbosity >= 1:105            if self.fixture_count == 0 and self.hide_empty:106                pass107            elif self.fixture_object_count == self.loaded_object_count:108                self.stdout.write("Installed %d object(s) from %d fixture(s)" %109                    (self.loaded_object_count, self.fixture_count))110            else:111                self.stdout.write("Installed %d object(s) (of %d) from %d fixture(s)" %112                    (self.loaded_object_count, self.fixture_object_count, self.fixture_count))113    def load_label(self, fixture_label):114        """115        Loads fixtures files for a given label.116        """117        show_progress = self.verbosity >= 3118        for fixture_file, fixture_dir, fixture_name in self.find_fixtures(fixture_label):119            _, ser_fmt, cmp_fmt = self.parse_name(os.path.basename(fixture_file))120            open_method, mode = self.compression_formats[cmp_fmt]121            fixture = open_method(fixture_file, mode)122            try:123                self.fixture_count += 1124                objects_in_fixture = 0125                loaded_objects_in_fixture = 0126                if self.verbosity >= 2:127                    self.stdout.write("Installing %s fixture '%s' from %s." %128                        (ser_fmt, fixture_name, humanize(fixture_dir)))129                objects = serializers.deserialize(ser_fmt, fixture,130                    using=self.using, ignorenonexistent=self.ignore)131                for obj in objects:132                    objects_in_fixture += 1133                    if router.allow_migrate_model(self.using, obj.object.__class__):134                        loaded_objects_in_fixture += 1135                        self.models.add(obj.object.__class__)136                        try:137                            obj.save(using=self.using)138                            if show_progress:139                                self.stdout.write(140                                    '\rProcessed %i object(s).' % loaded_objects_in_fixture,141                                    ending=''142                                )143                        except (DatabaseError, IntegrityError) as e:144                            e.args = ("Could not load %(app_label)s.%(object_name)s(pk=%(pk)s): %(error_msg)s" % {145                                'app_label': obj.object._meta.app_label,146                                'object_name': obj.object._meta.object_name,147                                'pk': obj.object.pk,148                                'error_msg': force_text(e)149                            },)150                            raise151                if objects and show_progress:152                    self.stdout.write('')  # add a newline after progress indicator153                self.loaded_object_count += loaded_objects_in_fixture154                self.fixture_object_count += objects_in_fixture155            except Exception as e:156                if not isinstance(e, CommandError):157                    e.args = ("Problem installing fixture '%s': %s" % (fixture_file, e),)158                raise159            finally:160                fixture.close()161            # Warn if the fixture we loaded contains 0 objects.162            if objects_in_fixture == 0:163                warnings.warn(164                    "No fixture data found for '%s'. (File format may be "165                    "invalid.)" % fixture_name,166                    RuntimeWarning167                )168    @lru_cache.lru_cache(maxsize=None)169    def find_fixtures(self, fixture_label):170        """171        Finds fixture files for a given label.172        """173        fixture_name, ser_fmt, cmp_fmt = self.parse_name(fixture_label)174        databases = [self.using, None]175        cmp_fmts = list(self.compression_formats.keys()) if cmp_fmt is None else [cmp_fmt]176        ser_fmts = serializers.get_public_serializer_formats() if ser_fmt is None else [ser_fmt]177        if self.verbosity >= 2:178            self.stdout.write("Loading '%s' fixtures..." % fixture_name)179        if os.path.isabs(fixture_name):180            fixture_dirs = [os.path.dirname(fixture_name)]181            fixture_name = os.path.basename(fixture_name)182        else:183            fixture_dirs = self.fixture_dirs184            if os.path.sep in os.path.normpath(fixture_name):185                fixture_dirs = [os.path.join(dir_, os.path.dirname(fixture_name))186                                for dir_ in fixture_dirs]187                fixture_name = os.path.basename(fixture_name)188        suffixes = ('.'.join(ext for ext in combo if ext)189                for combo in product(databases, ser_fmts, cmp_fmts))190        targets = set('.'.join((fixture_name, suffix)) for suffix in suffixes)191        fixture_files = []192        for fixture_dir in fixture_dirs:193            if self.verbosity >= 2:194                self.stdout.write("Checking %s for fixtures..." % humanize(fixture_dir))195            fixture_files_in_dir = []196            path = os.path.join(fixture_dir, fixture_name)197            for candidate in glob.iglob(glob_escape(path) + '*'):198                if os.path.basename(candidate) in targets:199                    # Save the fixture_dir and fixture_name for future error messages.200                    fixture_files_in_dir.append((candidate, fixture_dir, fixture_name))201            if self.verbosity >= 2 and not fixture_files_in_dir:202                self.stdout.write("No fixture '%s' in %s." %203                                  (fixture_name, humanize(fixture_dir)))204            # Check kept for backwards-compatibility; it isn't clear why205            # duplicates are only allowed in different directories.206            if len(fixture_files_in_dir) > 1:207                raise CommandError(208                    "Multiple fixtures named '%s' in %s. Aborting." %209                    (fixture_name, humanize(fixture_dir)))210            fixture_files.extend(fixture_files_in_dir)211        if not fixture_files:212            # Warning kept for backwards-compatibility; why not an exception?213            warnings.warn("No fixture named '%s' found." % fixture_name)214        return fixture_files215    @cached_property216    def fixture_dirs(self):217        """218        Return a list of fixture directories.219        The list contains the 'fixtures' subdirectory of each installed220        application, if it exists, the directories in FIXTURE_DIRS, and the221        current directory.222        """223        dirs = []224        fixture_dirs = settings.FIXTURE_DIRS225        if len(fixture_dirs) != len(set(fixture_dirs)):226            raise ImproperlyConfigured("settings.FIXTURE_DIRS contains duplicates.")227        for app_config in apps.get_app_configs():228            app_label = app_config.label229            app_dir = os.path.join(app_config.path, 'fixtures')230            if app_dir in fixture_dirs:231                raise ImproperlyConfigured(232                    "'%s' is a default fixture directory for the '%s' app "233                    "and cannot be listed in settings.FIXTURE_DIRS." % (app_dir, app_label)234                )235            if self.app_label and app_label != self.app_label:236                continue237            if os.path.isdir(app_dir):238                dirs.append(app_dir)239        dirs.extend(list(fixture_dirs))240        dirs.append('')241        dirs = [upath(os.path.abspath(os.path.realpath(d))) for d in dirs]242        return dirs243    def parse_name(self, fixture_name):244        """245        Splits fixture name in name, serialization format, compression format.246        """247        parts = fixture_name.rsplit('.', 2)248        if len(parts) > 1 and parts[-1] in self.compression_formats:249            cmp_fmt = parts[-1]250            parts = parts[:-1]251        else:252            cmp_fmt = None253        if len(parts) > 1:254            if parts[-1] in self.serialization_formats:255                ser_fmt = parts[-1]256                parts = parts[:-1]257            else:258                raise CommandError(259                    "Problem installing fixture '%s': %s is not a known "260                    "serialization format." % (''.join(parts[:-1]), parts[-1]))261        else:262            ser_fmt = None263        name = '.'.join(parts)264        return name, ser_fmt, cmp_fmt265class SingleZipReader(zipfile.ZipFile):266    def __init__(self, *args, **kwargs):267        zipfile.ZipFile.__init__(self, *args, **kwargs)268        if len(self.namelist()) != 1:269            raise ValueError("Zip-compressed fixtures must contain one file.")270    def read(self):271        return zipfile.ZipFile.read(self, self.namelist()[0])272def humanize(dirname):...

test_basic_priority.py

Source:test_basic_priority.py

1from unittest import TestCase, mock, skip2from tasks import wait, sleep_seconds3from time import sleep4from celery import group, chord, chain5import unittest6# Priorities:   0, 3, 6, 97# Queues:       a-high, b-medium, c-low8import logging9from logging.handlers import RotatingFileHandler10logger = logging.getLogger('app')11logger.setLevel(logging.DEBUG)12handler = RotatingFileHandler('log/app.log', maxBytes=1024000*1024000, backupCount=2)13logger.addHandler(handler)14"""15NOTE:16This first task fired in each test must ALWAYS assumed to finish first. This17is because when the tasks fire, the queue is empty, so it has no other higher priority18"""19class TestPriority(TestCase):20    def test_simple(self):21        """22        Test a simple FIFO queue with priority (de)escalation23        """24        tasks = [25            { "priority": 0, "fixture_name": "A" },26            { "priority": 0, "fixture_name": "B" },27            { "priority": 0, "fixture_name": "C" },28            { "priority": 9, "fixture_name": "D" }, # deescalate29            { "priority": 0, "fixture_name": "E" },30            { "priority": 0, "fixture_name": "F" },31            { "priority": 9, "fixture_name": "G" }, # deescalate32            { "priority": 0, "fixture_name": "H" },33        ]34        results = [] 35        for task in tasks:36            t = wait.s(**task)37            results.append(t.apply_async(priority=task["priority"]))38        complete = False39        success = []40        while not complete:41            complete = True42            for r in results:43                if r.state != "SUCCESS":44                    complete = False45                else:46                    v = r.result47                    if v not in success:48                        success.append(v)49            sleep(sleep_seconds)50        self.assertEqual(51            success,52            ["A", "B", "C", "E", "F", "H", "D", "G"],53            "Numeric Priority not completed in expected order"54        )55    def test_wait_chain(self):56        chain_tasks = [57            {"fixture_name": "0-A-2"},58            {"fixture_name": "0-A-3"},59            {"fixture_name": "0-B-1"}60        ]61        _c = []62        for task in chain_tasks:63            _c.append(wait.s(**task))64        logger.info(_c)65        _chains = []66        _chains.append(chain(_c))67        # t = chain(68        #         wait.s({"priority":0, "fixture_name": "0-A"}),69        #         chord(70        #             _chains,71        #             wait.s({"priority":0, "fixture_name": "0-B"})72        #         ),73        #         wait.s({"priority":0, "fixture_name": "0-C"}),74        #     )75        temp = chord(76                    _chains,77                    wait.s({"priority":0, "fixture_name": "0-B"})78                )79        result = temp.apply_async(priority=0)80        logger.info(result)81        complete = False82        while not complete:83            complete = True84            if result.state != "SUCCESS":85                complete = False86            else:87                logger.info(result.result)88        self.assertEqual(89            success,90            ["0-C", "1-C", "3-C", "2-C"],91            "Numeric Priority not completed in expected order"92        )93    def test_complex(self):94        """95        Test a complex chain of chords with (de)escalation96        """97        tasks_defs = [98            (0, 0),99            (1, 0),100            (2, 9), # deescalate101            (3, 0),102        ]103        results = []104        for task_id, task_priority in tasks_defs:105            _chains = []106            for chain_id in ["A", "B"]:107                chain_tasks = [108                    { "fixture_name": f"{task_id}-{chain_id}-1" },109                    { "fixture_name": f"{task_id}-{chain_id}-2" },110                    { "fixture_name": f"{task_id}-{chain_id}-3" },111                ]112                _c = []113                for task in chain_tasks:114                    _c.append(wait.s(**task))115                _chains.append(chain(_c))116            117            t = chain(118                wait.s({"priority":task_priority, "fixture_name": f"{task_id}-A"}),119                chord(120                    _chains,121                    wait.s({"priority":task_priority, "fixture_name": f"{task_id}-B"})122                ),123                wait.s({"priority":task_priority, "fixture_name": f"{task_id}-C"}),124            )125            logger.info(t)126            logger.info('\n')127            task_p = t.apply_async(priority=task_priority)128            logger.info(task_p)129            results.append(task_p)130        complete = False131        success = []132        while not complete:133            complete = True134            for r in results:135                if r.state != "SUCCESS":136                    complete = False137                else:138                    v = r.result139                    if v not in success:140                        success.append(v)141            sleep(sleep_seconds)142        self.assertEqual(143            success,144            ["0-C", "1-C", "3-C", "2-C"],145            "Numeric Priority not completed in expected order"146        )147class TestPriorityQueue(TestCase):148    def test_simple(self):149        """150        Test a simple FIFO queue with priority (de)escalation151        This test shows that priority is honored above queue order152        eg: given two queues, "a-work" and "b-work", and 3 tasks,153        "t-1", "t-2", and "t-3", if t-1 and t-2 are in a, and t3 is in b,154        they will complete in order (t1,t2,t3)155    #     However, if t-2 has a priority of 0, and all others have a priority of 3,156    #     they will complete: t-2, t-1, t-3157    #     Further, if t-3 has a priority of 0, and t-1 and t-2 have a priority of 3,158    #     they will complete: t-3, t-1, t-2159    #     """160    #     tasks = [161    #         { "priority": 0, "fixture_name": "A", "queue":"a-high"},162    #         { "priority": 0, "fixture_name": "B", "queue":"b-medium"},163    #         { "priority": 9, "fixture_name": "C", "queue":"b-medium"},164    #         { "priority": 3, "fixture_name": "D", "queue":"a-high"},165    #         { "priority": 3, "fixture_name": "E", "queue":"a-high"},166    #         { "priority": 3, "fixture_name": "F", "queue":"b-medium"},167    #         { "priority": 3, "fixture_name": "G", "queue":"a-high"},168    #         { "priority": 9, "fixture_name": "H", "queue":"a-high"},169    #     ]170    #     results = [] 171    #     for task in tasks:172    #         t = wait.s(**task)173    #         results.append(t.apply_async(priority=task["priority"], queue=task["queue"]))174    #     complete = False175    #     success = []176    #     while not complete:177    #         complete = True178    #         for r in results:179    #             if r.state != "SUCCESS":180    #                 complete = False181    #             else:182    #                 v = r.result183    #                 if v not in success:184    #                     success.append(v)185    #         sleep(sleep_seconds)186    #     self.assertEqual(187    #         success,188    #         ["A", "B", "D", "E", "G", "F", "H", "C"],189    #         "Numeric Priority not completed in expected order"190    #     )191    # def test_complex(self):192    #     """193    #     Test a complex chain of chords with (de)escalation194    #     This test further prooves what TestPriorityQueue.test_simple195    #     already states in its comment, but tests it further.196    #     There are, however, interesting things to note in the output197    #     such as task 0-C sometimes completing _after_ 2-A beings.198    #     """199    #     tasks_defs = [200    #         (0, 0, "a-high"),201    #         (1, 3, "c-low"),202    #         (2, 3, "a-high"),203    #         (3, 6, "c-low"),204    #     ]205    #     results = []206    #     for task_id, task_priority, queue in tasks_defs:207    #         _chains = []208    #         for chain_id in ["A", "B"]:209    #             chain_tasks = [210    #                 { "fixture_name": f"{task_id}-{chain_id}-1" },211    #                 { "fixture_name": f"{task_id}-{chain_id}-2" },212    #                 { "fixture_name": f"{task_id}-{chain_id}-3" },213    #             ]214    #             _c = []215    #             for task in chain_tasks:216    #                 _c.append(wait.s(**task))217    #             _chains.append(chain(_c))218    #         t = chain(219    #             wait.s({"prority":task_priority, "fixture_name": f"{task_id}-A"}),220    #             chord(221    #                 _chains,222    #                 wait.s({"prority":task_priority, "fixture_name": f"{task_id}-B"})223    #             ),224    #             wait.s({"prority":task_priority, "fixture_name": f"{task_id}-C"}),225    #         )226    #         results.append(t.apply_async(priority=task_priority, queue=queue))227    #     complete = False228    #     success = []229    #     while not complete:230    #         complete = True231    #         for r in results:232    #             if r.state != "SUCCESS":233    #                 complete = False234    #             else:235    #                 v = r.result236    #                 if v not in success:237    #                     success.append(v)238    #         sleep(sleep_seconds)239    #     self.assertEqual(240    #         success,241    #         ["0-C", "2-C", "1-C", "3-C"],242    #         "Numeric Priority not completed in expected order"243    #     )244if __name__ == '__main__':...

the_fixture.py

Source:the_fixture.py

1"""Load and manipulate fixtures."""2import json3import os.path4import importlib_resources5from behave import given, step, then  # pylint:disable=no-name-in-module6from tests.bdd.step_context import StepContext7class TheFixture(StepContext):8    MISSING = "*MISSING*"9    NONE = "*NONE*"10    context_key = "the_fixture"11    def __init__(self, **kwargs):12        super().__init__(**kwargs)13        self.base_dir = None14        self.fixtures = {}15    def set_base_dir(self, base_dir):16        base_dir = base_dir.lstrip("/")17        self.base_dir = str(18            importlib_resources.files("tests") / "bdd/fixtures" / base_dir19        )20        if not os.path.isdir(self.base_dir):21            raise EnvironmentError(f"Cannot find fixture dir: {self.base_dir}")22    def set_fixture(self, name, value):23        self.fixtures[name] = value24        return value25    def set_fixture_value(self, name, key, value):26        fixture = self.get_fixture(name)27        if value == self.MISSING:28            fixture.pop(key, None)29            return30        fixture[key] = None if value == self.NONE else value31    def get_fixture(self, name):32        return self.fixtures[name]33    def load_ini(self, filename, fixture_name):34        values = {}35        with open(self.get_path(filename), encoding="utf8") as handle:36            for line in handle:37                line = line.strip()38                if not line or line.startswith("#"):39                    continue40                key, value = line.split("=", 1)41                values[key] = value42        return self.set_fixture(fixture_name, values)43    def get_path(self, filename):44        return os.path.join(self.base_dir, filename)45    def do_teardown(self):46        self.fixtures = {}47@given("fixtures are located in '{location}'")48def fixture_location(context, location):49    context.the_fixture.set_base_dir(location)50@given("I load the fixture '{fixture_file}.ini' as '{fixture_name}'")51def load_ini_fixture(context, fixture_file, fixture_name):52    context.the_fixture.load_ini(fixture_file + ".ini", fixture_name)53@given("I set the fixture '{fixture_name}' key '{key}' to '{value}'")54def set_fixture_value(context, fixture_name, key, value):55    context.the_fixture.set_fixture_value(fixture_name, key, value)56@given("I update the fixture '{fixture_name}' with")57def update_fixture_from_table(context, fixture_name):58    for row in context.table:59        set_fixture_value(context, fixture_name, row[0].strip(), row[1].strip())60@given("I update the fixture '{fixture_name}' from fixture '{other_fixture}'")61def update_fixture_from_fixture(context, fixture_name, other_fixture):62    the_fixture = context.the_fixture63    the_fixture.get_fixture(fixture_name).update(the_fixture.get_fixture(other_fixture))64@then("the fixture '{fixture_name}' key '{key}' is the value")65def set_fixture_key_to_the_value(context, fixture_name, key):66    context.the_value = context.the_fixture.get_fixture(fixture_name)[key]67def diff_dicts(a, b, missing=KeyError):68    return {69        key: (a.get(key, missing), b.get(key, missing))70        for key in dict(set(a.items()) ^ set(b.items())).keys()71    }72@then("the fixture '{fixture_name}' matches the fixture '{other_fixture}'")73def compare_fixtures(context, fixture_name, other_fixture):74    diff = diff_dicts(75        context.the_fixture.get_fixture(fixture_name),76        context.the_fixture.get_fixture(other_fixture),77        missing=TheFixture.MISSING,78    )79    if not diff:80        return81    for key, (value_found, value_expected) in diff.items():82        print(f"Key {key} is different. Found {value_found} expected {value_expected}")83    assert diff == {}, "The fixtures differ"84@step("I dump the fixture '{fixture_name}'")85def dump_fixture(context, fixture_name):86    fixture = context.the_fixture.get_fixture(fixture_name)87    print(f"Fixture '{fixture_name}'")...

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.