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

How to use _copy_results method in autotest

Best Python code snippet using autotest_python

batchfind.py

Source:batchfind.py

...80            t.info("{}.results: {}".format(self.__class__.__name__, ", ".join(count)))81    def _process_results(self, parent, results):82        for key, value in results.items():83            if key in BatchFind._OBJECTS:84                self._copy_results(parent, key, value)85    def _get_obj(self, key):86        obj = copy.copy(BatchFind._OBJECTS[key]["base"])87        obj._clear()88        if key not in self._results.__dict__:89            self._results.__dict__[key] = []90        self._results.__dict__[key].append(obj)91        return obj92    def _copy_results(self, parent, key, value):93        if isinstance(value, list):94            for item in value:95                self._copy_results(parent, key, item)96        else:97            obj = self._get_obj(key)98            obj._parent = parent99            obj_values = {}100            for attr_name, attr_value in value.items():101                if attr_name == "href" or attr_name in obj._SDM_ATT_MAP.values():102                    obj_values[attr_name] = attr_value103                elif attr_name in BatchFind._OBJECTS:104                    self._process_results(obj, {attr_name: attr_value})105            obj._object_properties.append(obj_values)106    @property107    def results(self):...

handler.py

Source:handler.py

...42    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)43    sock.connect((host, port))44    sock.send(b("READY:" + lambda_id))45    return bytes_to_str(sock.recv(4096))46def _copy_results(logger, result):47    bucket = boto3.resource('s3').Bucket('bench-results')48    if os.path.isfile(result):49        logger.info('Copying results @ {} to S3...'.format(result))50        with open(result, 'rb') as data:51            bucket.put_object(Key=os.path.join("pocket", os.path.basename(result)), Body=data)52    else:53        logger.warn('Result file {} not found'.format(result))54def _pocket_write_buffer(p, job_id, lambda_id, num_ops, src, size):55    with open('pocket_write_%s.txt' % lambda_id, 'w') as output:56        for i in xrange(num_ops):57            dst_filename = 'tmp-' + lambda_id + '-' + str(i)58            t0 = time.time()59            r = pocket.put_buffer(p, src, size, dst_filename, job_id)60            t1 = time.time()61            tt = (t1 - t0) * 1e662            output.write('%d\n' % tt)63            if r != 0:64                raise Exception("put buffer failed: " + dst_filename)65def _pocket_read_buffer(p, job_id, lambda_id, num_ops, size):66    with open('pocket_read_%s.txt' % lambda_id, 'w') as output:67        text_back = " " * size68        for i in xrange(num_ops):69            dst_filename = 'tmp-' + lambda_id + '-' + str(i)70            t0 = time.time()71            r = pocket.get_buffer(p, dst_filename, text_back, size, job_id)72            t1 = time.time()73            tt = (t1 - t0) * 1e674            output.write('%d\n' % tt)75            if r != 0:76                raise Exception("get buffer failed: " + dst_filename)77def lambda_handler(event, context):78    # Read event info79    host = event.get('host')80    log_port = int(event.get('port'))81    control_port = log_port + 182    mode = event.get('mode')83    lambda_id = event.get('lambda_id')84    job_id = event.get('batch_id')85    object_size = event.get('object_size')86    num_ops = event.get('num_ops')87    namenode_ip = "10.1.0.10"88    namenode_port = 907089    text = 'a' * object_size90    try:91        logger = _connect_logger(host, log_port)92    except Exception as e:93        print('Exception: {}'.format(e))94        raise95    logger.info('Event: {}, Context: {}'.format(event, context))96    response = _signal(host, control_port, lambda_id)97    if response != "RUN":98        logger.abort("Control returned " + response)99        return100    # connect to pocket101    p = pocket.connect(namenode_ip, namenode_port)102    if "write" in mode:103        _pocket_write_buffer(p, job_id, lambda_id, num_ops, text, object_size)104    if "read" in mode:105        _pocket_read_buffer(p, job_id, lambda_id, num_ops, object_size)106    # pocket.close(p)107    _copy_results(logger, "pocket_write_%s.txt" % lambda_id)108    _copy_results(logger, "pocket_read_%s.txt" % lambda_id)109    logger.close()...

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.