Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use reset_logs method in lettuce-tools

Best Python code snippet using lettuce-tools_python

test_signals.py

Source:test_signals.py

...41        listenWithArgs.LOGGER.append( text )42        pass43    listenWithArgs.LOGGER = []44    #--------------------------------------------------------------------------------------45    def reset_logs() :46        del Listener.LOGGER[ : ]47        del listenFunction.LOGGER[ : ]48        del listenWithArgs.LOGGER[ : ]49        pass50        51    #--------------------------------------------------------------------------------------52    def count() :53        return \54            len( Listener.LOGGER ) + \55            len( listenFunction.LOGGER ) + \56            len( listenWithArgs.LOGGER )57        58    #--------------------------------------------------------------------------------------59    b = Button()60    l = Listener()61    62    #--------------------------------------------------------------------------------------63    # Demonstrating connecting and calling signals64    b.sigClick.connect( l.onClick )65    reset_logs()66    b.sigClick()67    assert len( Listener.LOGGER ) == 168    #--------------------------------------------------------------------------------------69    # Disconnecting all signals70    b.sigClick.disconnectAll()71    reset_logs()72    b.sigClick()73    assert len( Listener.LOGGER ) == 074    #--------------------------------------------------------------------------------------75    # connecting multiple functions to a signal76    l2 = Listener()77    b.sigClick.connect( l.onClick )78    b.sigClick.connect( l2.onClick )79    reset_logs()80    b.sigClick()81    82    assert len( Listener.LOGGER ) == 283    #--------------------------------------------------------------------------------------84    # disconnecting individual functions85    b.sigClick.disconnect( l.onClick )86    b.sigClick.connect( listenFunction )87    reset_logs()88    b.sigClick()89    90    assert len( Listener.LOGGER ) == 191    assert len( listenFunction.LOGGER ) == 192    #--------------------------------------------------------------------------------------93    # example with arguments and a local signal94    sig = Signal()95    sig.connect( listenWithArgs )96    reset_logs()97    sig( "Hello, World!" )98    assert len( listenWithArgs.LOGGER ) == 199    #--------------------------------------------------------------------------------------100    # signals disconnecting 'method slots' automatically101    b.sigClick.disconnectAll()102    b.sigClick.connect( l.onClick )103    b.sigClick.connect( l2.onClick )104    del l2105    reset_logs()106    b.sigClick()107    108    assert len( Listener.LOGGER ) == 1109    #--------------------------------------------------------------------------------------110    # signals disconnecting 'object slots' automatically111    b.sigClick.disconnectAll()112    b.sigClick.connect( l )113    del l114    reset_logs()115    b.sigClick()116    117    assert len( Listener.LOGGER ) == 0118    #--------------------------------------------------------------------------------------119    # signals disconnecting 'lambda slots' automatically120    sig = Signal()121    sig.connect( lambda *args : listenFunction() )122    reset_logs()123    sig( "Hello, World!" )124    assert len( listenFunction.LOGGER ) == 1125    #--------------------------------------------------------------------------------------126    pass127#--------------------------------------------------------------------------------------128def test_signals_1() :129    #--------------------------------------------------------------------------------------130    from test_utils import adjust_environment131    global_settings = adjust_environment()132    from reinteract.signals import Signal133    #--------------------------------------------------------------------------------------134    class Container( object ) :135        def __init__( self ) :136            self._attr = 'dummy'...

tutorial.py

Source:tutorial.py

...10if os.environ['USE_DTLS'] == '0':11    #12    # Hello world13    #14    reset_logs()15    tutorial1 = tutorial('subscribe ack')16    tutorial2 = tutorial('publish ack')17    expect(tutorial1, 'payload=Hello{}'.format(os.linesep))18    expect(tutorial2, 'payload=World{}'.format(os.linesep))19    #20    # Building a DPS network21    #22    reset_logs()23    tutorial1 = tutorial()24    tutorial2 = tutorial('-p {} subscribe ack'.format(tutorial1.port))25    tutorial3 = tutorial('-p {} publish ack'.format(tutorial1.port))26    expect(tutorial2, 'payload=Hello{}'.format(os.linesep))27    expect(tutorial3, 'payload=World{}'.format(os.linesep))28if os.environ['USE_DTLS'] != '0':29    #30    # DTLS with pre-shared keys31    #32    reset_logs()33    tutorial1 = tutorial('-x network-psk subscribe ack')34    tutorial2 = tutorial('-p {} -x network-psk publish ack'.format(tutorial1.port))35    expect(tutorial1, 'payload=Hello{}'.format(os.linesep))36    expect(tutorial2, 'payload=World{}'.format(os.linesep))37    #38    # DTLS with certificates39    #40    reset_logs()41    tutorial1 = tutorial('-x network-cert subscribe ack')42    tutorial2 = tutorial('-p {} -x network-cert publish ack'.format(tutorial1.port))43    expect(tutorial1, 'payload=Hello{}'.format(os.linesep))44    expect(tutorial2, 'payload=World{}'.format(os.linesep))45if os.environ['USE_DTLS'] == '0':46    #47    # Protecting the payload - symmetric key48    #49    reset_logs()50    tutorial1 = tutorial('-x symmetric subscribe ack')51    tutorial2 = tutorial('-p {} -x symmetric publish ack'.format(tutorial1.port))52    expect(tutorial1, 'payload=Hello{}'.format(os.linesep))53    expect(tutorial2, 'payload=World{}'.format(os.linesep))54    #55    # Protecting the payload - asymmetric key56    #57    reset_logs()58    tutorial1 = tutorial('-x asymmetric subscribe')59    tutorial2 = tutorial('-p {} -x asymmetric publish'.format(tutorial1.port))60    expect(tutorial1, 'payload=Hello{}'.format(os.linesep))61    #62    # Authenticating the sender - symmetric key63    #64    reset_logs()65    tutorial1 = tutorial('-x symmetric auth subscribe ack')66    tutorial2 = tutorial('-p {} -x symmetric auth publish ack'.format(tutorial1.port))67    expect(tutorial1, 'payload=Hello{}'.format(os.linesep))68    expect(tutorial2, 'payload=World{}'.format(os.linesep))69    #70    # Authenticating the sender - asymmetric key71    #72    reset_logs()73    tutorial1 = tutorial('-x asymmetric auth subscribe ack')74    tutorial2 = tutorial('-p {} -x asymmetric auth publish ack'.format(tutorial1.port))75    expect(tutorial1, 'payload=Hello{}'.format(os.linesep))...

logging_patch.py

Source:logging_patch.py

...4import pandas as pd5from .logging_utils import bytes_to_human_readable, collect_log_data, assign_diffs6def log_mem(self, msg='', verbose=False):7    if not hasattr(self, 'logs'):8        self.reset_logs()9    self.logs.append(collect_log_data(msg=msg, verbose=verbose))10def reset_logs(self):11    def resetter(module):12        module.logs = []13        module.t_init = time.time()14    self.apply(resetter)15def save_logs(self, path):16    strang = '\n'.join(self.combine_logs().long_msg.values)17    with open(path, 'w') as f:18        f.write(strang)19def combine_logs(self):20    LOGS = []21    def get_child_logs(module):22        df = getattr(module, 'log_df', pd.DataFrame())23        LOGS.append(df)24    self.apply(get_child_logs)25    log_df = pd.concat(LOGS).sort_values('time')26    return log_df.pipe(assign_diffs).sort_values('time')27def summary_fn(self):28    log_df = self.combine_logs()29    ranges = {x: log_df[x].max() - log_df[x].min() for x in ['cpu_mem', 'gpu_mem', 'time']}30    ranges['cpu_mem_chg'] = bytes_to_human_readable(ranges['cpu_mem'])31    ranges['gpu_mem_chg'] = bytes_to_human_readable(ranges['gpu_mem'])32    ranges['gpu_mem_max'] = bytes_to_human_readable(log_df['gpu_mem'].max())33    ranges['gpu_mem_min'] = bytes_to_human_readable(log_df['gpu_mem'].min())34    ranges['time_second'] = round(ranges['time'], 2)35    return pd.Series(ranges).drop(['gpu_mem', 'cpu_mem'])36from types import MethodType37import warnings38def patch_module_with_memory_mixin(model):39    """create logging methods using MethodType"""40    _method = lambda f: MethodType(f, model)41    try:42        model.reset_logs = _method(reset_logs)43    except AttributeError:44        warnings.warn('Cant patch attribute reset_logs, assuming LoggingMixin already being used')45    model.save_logs = _method(save_logs)46    model.save_log_csv = _method(save_log_csv)47    #model.log_mem = _method(log_mem)48    cls = type(model)49    cls.log_df = property(log_df_fn)50    model.log_mem = _method(log_mem)51    cls.summary = property(summary_fn)52    model.combine_logs = _method(combine_logs)53def log_df_fn(self):54    if not hasattr(self, 'logs'):55        self.reset_logs()56    log_df = pd.DataFrame(self.logs)57    return log_df58def save_log_csv(self, path):...

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.