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

How to use _restart method in lisa

Best Python code snippet using lisa_python

recipe-81613.py

Source:recipe-81613.py

...9    netsvc.Service.__init__(self,name)10    self.joinGroup("database-services")11    self._cursor = cursor12    self._timeout = timeout13    self._restart()14    self.exportMethod(self.execute)15    self.exportMethod(self.executemany)16    self.exportMethod(self.description)17    self.exportMethod(self.rowcount)18    self.exportMethod(self.fetchone)19    self.exportMethod(self.fetchmany)20    self.exportMethod(self.fetchall)21    self.exportMethod(self.arraysize)22    self.exportMethod(self.close)23  def encodeObject(self,object):24    if hasattr(MySQLdb,"DateTime"):25      if type(object) == MySQLdb.DateTimeType:26	return ("xsd:string",object.strftime())27      elif type(object) == MySQLdb.DateTimeDeltaType:28	return ("xsd:string",str(object))29    return netsvc.Service.encodeObject(self,object)30  def executeMethod(self,name,method,params):31    try:32      return netsvc.Service.executeMethod(self,name,method,params)33    except MySQLdb.ProgrammingError,exception:34      self.abortResponse(1,"Programming Error","db",str(exception))35    except MySQLdb.Error,(error,description):36      self.abortResponse(error,description,"mysql")37  def _restart(self):38    self.cancelTimer("idle")39    self.startTimer(self._expire,self._timeout,"idle")40  def _expire(self,name):41    if name == "idle":42      self.close()43  def execute(self,query,args=None):44    result = self._cursor.execute(query,args)45    self._restart()46    return result47  def executemany(self,query,args=None):48    result = self._cursor.executemany(query,args)49    self._restart()50    return result51  def description(self):52    self._restart()53    return self._cursor.description54  def rowcount(self):55    self._restart()56    return self._cursor.rowcount57  def fetchone(self):58    result = self._cursor.fetchone()59    self._restart()60    return result61  def fetchmany(self,size=None):62    if size == None:63      size = self._cursor.arraysize64    result = self._cursor.fetchmany(size)65    self._restart()66    return result67  def fetchall(self):68    result = self._cursor.fetchall()69    self._restart()70    return result71  def arraysize(self):72    self._restart()73    return self._cursor.arraysize74  def close(self):75    self._cursor.close()76    self.cancelTimer("idle")77    self.destroyReferences()78    return 079class Database(netsvc.Service):80  def __init__(self,name,**kw):81    netsvc.Service.__init__(self,name)82    self._name = name83    self.joinGroup("database-services")84    self._database = MySQLdb.connect(**kw)85    self._cursors = 086    self.exportMethod(self.execute)...

Sram.py

Source:Sram.py

...22        self.log.info("  cocotbext-sram version %s", self._version)23        self.log.info("  Copyright (c) 2022 Torsten Meissner")24    25        self._active = None26        self._restart()27    def _restart(self):28        self.log.debug("SramRead._restart()")29        if self._active is not None:30            self._active.kill()31        # Schedule SRAM read to run concurrently32        self._active = cocotb.start_soon(self._read())33    async def _read(self):34        self.log.debug("SramRead._read()")35        while True:36            await self._clkedge37            if self._ren.value == 1:38                _data = self._mem[str(self._adr.value)]39                self._din.value = _data40                self.log.info(f"Read data:  {hex(_data)} from adr: {hex(self._adr.value)}")41class SramWrite(Sram):42    def __init__(self, clk, wen, adr, dout, mem, *args, **kwargs):43        super().__init__(clk, wen, None, adr, None, dout, mem, *args, **kwargs)44    45        self.log.info("SRAM write")46        self.log.info("  cocotbext-sram version %s", self._version)47        self.log.info("  Copyright (c) 2022 Torsten Meissner")48    49        self._active = None50        self._restart()51    def _restart(self):52        self.log.debug("SramWrite._restart()")53        if self._active is not None:54            self._active.kill()55        # Schedule SRAM write to run concurrently56        self._active = cocotb.start_soon(self._write())57    async def _write(self):58        self.log.debug("SramWrite._write()")59        while True:60            await self._clkedge61            if self._wen.value == 1:62                self._mem[str(self._adr.value)] = self._dout.value63                self.log.info(f"Wrote data: {hex(self._dout.value)} to adr:   {hex(self._adr.value)}")64class SramMonitor(Sram):65    def __init__(self, clk, wen, ren, adr, din, dout, *args, **kwargs):66        super().__init__(clk, wen, ren, adr, din, dout, None, *args, **kwargs)67    68        self.log.info("SRAM monitor")69        self.log.info("  cocotbext-sram version %s", self._version)70        self.log.info("  Copyright (c) 2022 Torsten Meissner")71    72        self._active = None73        self._transactions = {}74        self._restart()75    def _restart(self):76        self.log.debug("SramMonitor._restart()")77        if self._active is not None:78            self._active.kill()79        # Schedule SRAM read to run concurrently80        self._active = cocotb.start_soon(self._read())81    async def _read(self):82        self.log.debug("SramMonitor._read()")83        while True:84            await self._clkedge85            if self._wen.value:86                self._transactions[str(get_sim_time('ns'))] = {87                    "type" : "write",88                    "adr"  : self._adr.value,89                    "data" : self._dout.value}90            elif self._ren.value:...

miner.py

Source:miner.py

1import threading2import time3from cryptonet.debug import debug4from cryptonet.errors import ValidationError5class Miner:6    def __init__(self, chain, seek_n_build):7        self._shutdown = False8        self._restart = False9        self.threads = [threading.Thread(target=self.mine)]10        self.chain = chain11        self.chain.set_miner(self)12        self.seek_n_build = seek_n_build13    def run(self):14        for t in self.threads:15            t.start()16    def shutdown(self):17        self._shutdown = True18        debug('miner: shutdown called')19        for t in self.threads:20            t.join()21    def restart(self):22        self._restart = True23    def mine(self, provided_block=None):24        print(provided_block)25        while not self._shutdown:26            self._restart = False27            if provided_block is None:28                block = self.chain.head.get_candidate(self.chain)29                if hasattr(block, 'update_roots'):30                    block.update_roots()31            else:32                block = provided_block33            count = 034            debug('miner (re)starting', block.serialize(), self._shutdown, self._restart)35            while not self._shutdown and not self._restart:36                count += 137                block.increment_nonce()38                if block.valid_proof():39                    try:40                        block.assert_internal_consistency()41                        break42                    except ValidationError as e:43                        debug('Miner: invalid block generated: %s' % block.serialize())44                        continue45            if self._shutdown: break46            provided_block = None47            if self._restart:48                self._restart = False49                time.sleep(0.01)50                print('miner -restarting')51                continue52            debug('Miner: Found Soln : %064x' % block.get_hash(), block.serialize())53            if block.height == 0:  # print genesis54                debug('Miner: ser\'d genesis block: ', block.serialize())55                break  # break and let chain restart miner56            self.seek_n_build.add_block(block.__class__.deserialize(block.serialize()))57            debug('Miner: submitted block')58            while not self._restart and not self._shutdown:59                time.sleep(0.02)...

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.