How to use assert_cache_hits method in locust

Best Python code snippet using locust

accessor_decors.py

Source:accessor_decors.py

...149            num_total = len(rowid_list)150            num_hit   = num_total - num_miss151            print('\n[get] %s.%s %d / %d cache hits' %152                  (tblname, colname, num_hit, num_total))153        def assert_cache_hits(ibs, ismiss_list, rowid_list, kwargs_hash, **kwargs):154            cached_rowid_list = ut.filterfalse_items(rowid_list, ismiss_list)155            cache_ = ibs.table_cache[tblname][colname][kwargs_hash]156            # Load cached values for each rowid157            cache_vals_list = ut.dict_take_list(cache_, cached_rowid_list, None)158            db_vals_list = getter_func(ibs, cached_rowid_list, **kwargs)159            # Assert everything is valid160            msg_fmt = ut.codeblock(161                '''162                [assert_cache_hits] tblname = %r163                [assert_cache_hits] colname = %r164                [assert_cache_hits] cfgkeys = %r165                [assert_cache_hits] CACHE INVALID: %r != %r166                '''167            )168            msg = msg_fmt % (tblname, colname, cfgkeys, cache_vals_list, db_vals_list, )169            try:170                list1 = cache_vals_list171                list2 = db_vals_list172                assert ut.lists_eq(list1, list2), msg173                #if isinstance(db_vals_list, list):174                #    assert cache_vals_list == db_vals_list, msg175                #else:176                #    assert np.all(cache_vals_list == db_vals_list), msg177            except AssertionError as ex:178                raise ex179            except Exception as ex2:180                print(type(cache_vals_list))181                print(type(db_vals_list))182                ut.printex(ex2)183                ut.embed()184                raise185        if False:186            #@profile cannot profile this because it is alrady being profiled by187            def wrp_getter_cacher(ibs, rowid_list, **kwargs):188                """189                Wrapper function that caches rowid values in a dictionary190                """191                # HACK TAKE OUT GETTING DEBUG OUT OF KWARGS192                debug_ = kwargs.pop('debug', False)193                if cfgkeys is not None:194                    #kwargs_hash = ut.get_dict_hashid(ut.dict_take_list(kwargs, cfgkeys, None))195                    kwargs_hash = ut.get_dict_hashid([kwargs.get(key, None) for key in cfgkeys])196                    #ut.dict_take_list(kwargs, cfgkeys, None))197                else:198                    kwargs_hash = None199                #+----------------------------200                # There are 3 levels of caches201                #+----------------------------202                # All caches for this table203                #colscache_ = ibs.table_cache[tblname]204                ## All caches for the this column205                #kwargs_cache_ = colscache_[colname]206                ## All caches for this kwargs configuration207                #cache_ = kwargs_cache_[kwargs_hash]208                cache_ = ibs.table_cache[tblname][colname][kwargs_hash]209                #L____________________________210                # Load cached values for each rowid211                #vals_list = ut.dict_take_list(cache_, rowid_list, None)212                vals_list = [cache_.get(rowid, None) for rowid in rowid_list]213                # Mark rowids with cache misses214                ismiss_list = [val is None for val in vals_list]215                if debug or debug_:216                    debug_cache_hits(ismiss_list, rowid_list)217                    #print('[cache_getter] "debug_cache_hits" turned off')218                # HACK !!! DEBUG THESE GETTERS BY ASSERTING INFORMATION IN CACHE IS CORRECT219                if ASSERT_API_CACHE:220                    assert_cache_hits(ibs, ismiss_list, rowid_list, kwargs_hash, **kwargs)221                # END HACK222                if any(ismiss_list):223                    miss_indices = ut.list_where(ismiss_list)224                    miss_rowids  = ut.filter_items(rowid_list, ismiss_list)225                    # call wrapped function226                    miss_vals = getter_func(ibs, miss_rowids, **kwargs)227                    # overwrite missed output228                    for index, val in zip(miss_indices, miss_vals):229                        vals_list[index] = val  # Output write230                    # cache save231                    for rowid, val in zip(miss_rowids, miss_vals):232                        cache_[rowid] = val     # Cache write233                return vals_list234        else:...

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.