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

How to use get_subunit_stream method in stestr

Best Python code snippet using stestr_python

memory.py

Source:memory.py

...74    def __init__(self, repository):75        self._repository = repository76    def get_id(self):77        return None78    def get_subunit_stream(self):79        result = BytesIO()80        serialiser = subunit.v2.StreamResultToBytes(result)81        serialiser = testtools.ExtendedToStreamDecorator(serialiser)82        serialiser.startTestRun()83        try:84            self.run(serialiser)85        finally:86            serialiser.stopTestRun()87        result.seek(0)88        return result89    def get_test(self):90        def wrap_result(result):91            # Wrap in a router to mask out startTestRun/stopTestRun from the92            # ExtendedToStreamDecorator.93            result = testtools.StreamResultRouter(result, do_start_stop_run=False)94            # Wrap that in ExtendedToStreamDecorator to convert v1 calls to95            # StreamResult.96            return testtools.ExtendedToStreamDecorator(result)97        return testtools.DecorateTestCaseResult(98            self, wrap_result, methodcaller('startTestRun'),99            methodcaller('stopTestRun'))100    def run(self, result):101        # Speaks original V1 protocol.102        for case in self._repository._failing.values():103            case.run(result)104class _Inserter(AbstractTestRun):105    """Insert test results into a memory repository, and describe them later."""106    def __init__(self, repository, partial):107        self._repository = repository108        self._partial = partial109        self._tests = []110        # Subunit V2 stream for get_subunit_stream111        self._subunit = None112    def startTestRun(self):113        self._subunit = BytesIO()114        serialiser = subunit.v2.StreamResultToBytes(self._subunit)115        self._hook = testtools.CopyStreamResult([116            testtools.StreamToDict(self._handle_test),117            serialiser])118        self._hook.startTestRun()119    def _handle_test(self, test_dict):120        self._tests.append(test_dict)121        start, stop = test_dict['timestamps']122        if test_dict['status'] == 'exists' or None in (start, stop):123            return124        duration_delta = stop - start125        duration_seconds = ((duration_delta.microseconds +126            (duration_delta.seconds + duration_delta.days * 24 * 3600)127            * 10**6) / 10.0**6)128        self._repository._times[test_dict['id']] = duration_seconds129    def stopTestRun(self):130        self._hook.stopTestRun()131        self._repository._runs.append(self)132        self._run_id = len(self._repository._runs) - 1133        if not self._partial:134            self._repository._failing = OrderedDict()135        for test_dict in self._tests:136            test_id = test_dict['id']137            if test_dict['status'] == 'fail':138                case = testtools.testresult.real.test_dict_to_case(test_dict)139                self._repository._failing[test_id] = case140            else:141                self._repository._failing.pop(test_id, None)142        return self._run_id143    def status(self, *args, **kwargs):144        self._hook.status(*args, **kwargs)145    def get_id(self):146        return self._run_id147    def get_subunit_stream(self):148        self._subunit.seek(0)149        return self._subunit150    def get_test(self):151        def wrap_result(result):152            # Wrap in a router to mask out startTestRun/stopTestRun from the153            # ExtendedToStreamDecorator.154            result = testtools.StreamResultRouter(result, do_start_stop_run=False)155            # Wrap that in ExtendedToStreamDecorator to convert v1 calls to156            # StreamResult.157            return testtools.ExtendedToStreamDecorator(result)158        return testtools.DecorateTestCaseResult(159            self, wrap_result, methodcaller('startTestRun'),160            methodcaller('stopTestRun'))161    def run(self, result):...

samba_buildfarm.py

Source:samba_buildfarm.py

...60        self.base_url = base_url61        self.run_id = run_id62        self.url = urllib.basejoin(self.base_url,63            "../../build/%s/+subunit" % self.run_id)64    def get_subunit_stream(self):65        return urllib.urlopen(self.url)66    def get_test(self):...

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.