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

How to use test_partition_tests_with_grouping method in stestr

Best Python code snippet using stestr_python

test_testcommand.py

Source:test_testcommand.py

...341        test_ids = frozenset(['zero1', 'zero2'])342        partitions = fixture.partition_tests(test_ids, 2)343        self.assertEqual(1, len(partitions[0]))344        self.assertEqual(1, len(partitions[1]))345    def test_partition_tests_with_grouping(self):346        repo = memory.RepositoryFactory().initialise('memory:')347        result = repo.get_inserter()348        result.startTestRun()349        run_timed("TestCase1.slow", 3, result)350        run_timed("TestCase2.fast1", 1, result)351        run_timed("TestCase2.fast2", 1, result)352        result.stopTestRun()353        ui, command = self.get_test_ui_and_cmd(repository=repo)354        self.set_config(355            '[DEFAULT]\ntest_command=foo $IDLIST $LISTOPT\n'356            'test_list_option=--list\n')357        fixture = self.useFixture(command.get_run_command())358        test_ids = frozenset(['TestCase1.slow', 'TestCase1.fast',359                              'TestCase1.fast2', 'TestCase2.fast1',...

test_command.py

Source:test_command.py

1# Copyright 2018 Matthew Treinish2#3# This file is part of junitxml2subunit4#5# junitxml2subunit is free software: you can redistribute it and/or modify6# it under the terms of the GNU General Public License as published by7# the Free Software Foundation, either version 3 of the License, or8# (at your option) any later version.9#10# junitxml2subunit is distributed in the hope that it will be useful,11# but WITHOUT ANY WARRANTY; without even the implied warranty of12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the13# GNU General Public License for more details.14#15# You should have received a copy of the GNU General Public License16# along with junitxml2subunit.  If not, see <http://www.gnu.org/licenses/>.17import functools18import io19import os20import subprocess21import tempfile22import fixtures23import subunit24import testtools25class TestJunitXML2Subunit(testtools.TestCase):26    @classmethod27    def setUpClass(cls):28        super(TestJunitXML2Subunit, cls).setUpClass()29        cls.examples_dir = os.path.abspath(30            os.path.join(os.path.dirname(__file__), 'examples'))31        test_dir = os.path.abspath(os.path.dirname(__file__))32        cls.repo_root = os.path.dirname(test_dir)33        subprocess.call(['cargo', 'build'], cwd=cls.repo_root)34        cls.command = os.path.join(cls.repo_root,35                                   'target/debug/junitxml2subunit')36    def setUp(self):37        super(TestJunitXML2Subunit, self).setUp()38        stdout = self.useFixture(fixtures.StringStream('stdout')).stream39        self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))40        stderr = self.useFixture(fixtures.StringStream('stderr')).stream41        self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))42        self.useFixture(fixtures.LoggerFixture(nuke_handlers=False,43                                               level=None))44    def _check_subunit(self, output_stream):45        stream = subunit.ByteStreamToStreamResult(output_stream)46        starts = testtools.StreamResult()47        summary = testtools.StreamSummary()48        tests = []49        def _add_dict(test):50            tests.append(test)51        outcomes = testtools.StreamToDict(functools.partial(_add_dict))52        result = testtools.CopyStreamResult([starts, outcomes, summary])53        result.startTestRun()54        try:55            stream.run(result)56        finally:57            result.stopTestRun()58        self.assertThat(len(tests), testtools.matchers.GreaterThan(0))59        return tests60    def test_ant_xml_in_stdout(self):61        ant_xml_path = os.path.join(self.examples_dir, 'ant.xml')62        junitxml_proc = subprocess.Popen([self.command, ant_xml_path],63                                         cwd=self.repo_root,64                                         stdout=subprocess.PIPE)65        stdout, _ = junitxml_proc.communicate()66        tests = self._check_subunit(io.BytesIO(stdout))67        self.assertEqual(1, len(tests))68        self.assertEqual('codereview.cobol.rules.ProgramIdRule',69                         tests[0].get('id'))70        self.assertEqual('fail', tests[0].get('status'))71    def test_ant_xml_in_file_out(self):72        ant_xml_path = os.path.join(self.examples_dir, 'ant.xml')73        out_file, out_path = tempfile.mkstemp()74        os.close(out_file)75        self.addCleanup(os.remove, out_path)76        junitxml_proc = subprocess.Popen([self.command, '-o', out_path,77                                          ant_xml_path],78                                         cwd=self.repo_root,79                                         stdout=subprocess.PIPE)80        stdout, _ = junitxml_proc.communicate()81        with open(out_path, 'r') as fd:82            tests = self._check_subunit(fd)83        self.assertEqual(1, len(tests))84        self.assertEqual('codereview.cobol.rules.ProgramIdRule',85                         tests[0].get('id'))86        self.assertEqual('fail', tests[0].get('status'))87    def test_hudson_xml_in_stdout(self):88        hudson_xml_path = os.path.join(self.examples_dir, 'hudson.xml')89        junitxml_proc = subprocess.Popen([self.command, hudson_xml_path],90                                         cwd=self.repo_root,91                                         stdout=subprocess.PIPE)92        stdout, _ = junitxml_proc.communicate()93        tests = self._check_subunit(io.BytesIO(stdout))94        self.assertEqual(3, len(tests))95        test_ids = [x.get('id') for x in tests]96        test_statuses = [x.get('status') for x in tests]97        self.assertIn('tests.ATest.error',98                      test_ids)99        self.assertIn('tests.ATest.fail', test_ids)100        self.assertIn('tests.ATest.success', test_ids)101        self.assertEqual(['fail', 'fail', 'success'], test_statuses)102    def test_hudson_xml_in_file_out(self):103        hudson_xml_path = os.path.join(self.examples_dir, 'hudson.xml')104        out_file, out_path = tempfile.mkstemp()105        os.close(out_file)106        self.addCleanup(os.remove, out_path)107        junitxml_proc = subprocess.Popen([self.command, '-o', out_path,108                                          hudson_xml_path],109                                         cwd=self.repo_root,110                                         stdout=subprocess.PIPE)111        stdout, _ = junitxml_proc.communicate()112        with open(out_path, 'r') as fd:113            tests = self._check_subunit(fd)114        self.assertEqual(3, len(tests))115        test_ids = [x.get('id') for x in tests]116        test_statuses = [x.get('status') for x in tests]117        self.assertIn('tests.ATest.error',118                      test_ids)119        self.assertIn('tests.ATest.fail', test_ids)120        self.assertIn('tests.ATest.success', test_ids)121        self.assertEqual(['fail', 'fail', 'success'], test_statuses)122    def test_pytest_xml_in_stdout(self):123        pytest_xml_path = os.path.join(self.examples_dir, 'pytest.xml')124        junitxml_proc = subprocess.Popen([self.command, pytest_xml_path],125                                         cwd=self.repo_root,126                                         stdout=subprocess.PIPE)127        stdout, _ = junitxml_proc.communicate()128        tests = self._check_subunit(io.BytesIO(stdout))129        self.assertEqual(118, len(tests))130        skip_count = len([x for x in tests if x.get('status') == 'skip'])131        success_count = len([x for x in tests if x.get('status') == 'success'])132        example_id = ('stestr.tests.test_scheduler.TestScheduler.'133                      'test_partition_tests_with_grouping')134        test_ids = [x.get('id') for x in tests]135        self.assertIn(example_id, test_ids)136        self.assertEqual(skip_count, 2)137        self.assertEqual(success_count, 116)138    def test_pytest_xml_in_file_out(self):139        pytest_xml_path = os.path.join(self.examples_dir, 'pytest.xml')140        out_file, out_path = tempfile.mkstemp()141        os.close(out_file)142        self.addCleanup(os.remove, out_path)143        junitxml_proc = subprocess.Popen([self.command, '-o', out_path,144                                          pytest_xml_path],145                                         cwd=self.repo_root,146                                         stdout=subprocess.PIPE)147        stdout, _ = junitxml_proc.communicate()148        with open(out_path, 'r') as fd:149            tests = self._check_subunit(fd)150        self.assertEqual(118, len(tests))151        test_ids = [x.get('id') for x in tests]152        skip_count = len([x for x in tests if x.get('status') == 'skip'])153        success_count = len([x for x in tests if x.get('status') == 'success'])154        example_id = ('stestr.tests.test_scheduler.TestScheduler.'155                      'test_partition_tests_with_grouping')156        self.assertIn(example_id, test_ids)157        self.assertEqual(skip_count, 2)158        self.assertEqual(success_count, 116)159    def test_no_time_xml_in_stdout(self):160        no_time_xml_path = os.path.join(self.examples_dir, 'no_time.xml')161        junitxml_proc = subprocess.Popen([self.command, no_time_xml_path],162                                         cwd=self.repo_root,163                                         stdout=subprocess.PIPE,164                                         stderr=subprocess.PIPE,165                                         encoding='utf8')166        _, stderr = junitxml_proc.communicate()167        self.assertEqual(2, junitxml_proc.returncode)168        self.assertEqual(169            "Invalid XML: There is no time attribute on a testcase",170            stderr.strip())171    def test_no_time_xml_in_file_out(self):172        no_time_xml_path = os.path.join(self.examples_dir, 'no_time.xml')173        out_file, out_path = tempfile.mkstemp()174        os.close(out_file)175        self.addCleanup(os.remove, out_path)176        junitxml_proc = subprocess.Popen([self.command, '-o', out_path,177                                          no_time_xml_path],178                                         cwd=self.repo_root,179                                         stdout=subprocess.PIPE,180                                         stderr=subprocess.PIPE,181                                         encoding='utf8')182        _, stderr = junitxml_proc.communicate()183        self.assertEqual(2, junitxml_proc.returncode)184        self.assertEqual(185            "Invalid XML: There is no time attribute on a testcase",186            stderr.strip())187    def test_no_id_xml_in_stdout(self):188        no_id_xml_path = os.path.join(self.examples_dir, 'no_id.xml')189        junitxml_proc = subprocess.Popen([self.command, no_id_xml_path],190                                         cwd=self.repo_root,191                                         stdout=subprocess.PIPE,192                                         stderr=subprocess.PIPE,193                                         encoding='utf8')194        _, stderr = junitxml_proc.communicate()195        self.assertEqual(3, junitxml_proc.returncode)196        self.assertEqual(197            "Invalid XML: There is no testname or classname attribute on a "198            "testcase",199            stderr.strip())200    def test_no_id_xml_in_file_out(self):201        no_id_xml_path = os.path.join(self.examples_dir, 'no_id.xml')202        out_file, out_path = tempfile.mkstemp()203        os.close(out_file)204        self.addCleanup(os.remove, out_path)205        junitxml_proc = subprocess.Popen([self.command, '-o', out_path,206                                          no_id_xml_path],207                                         cwd=self.repo_root,208                                         stdout=subprocess.PIPE,209                                         stderr=subprocess.PIPE,210                                         encoding='utf8')211        _, stderr = junitxml_proc.communicate()212        self.assertEqual(3, junitxml_proc.returncode)213        self.assertEqual(214            "Invalid XML: There is no testname or classname attribute on a "215            "testcase",216            stderr.strip())217    def test_missing_file(self):218        out_file, out_path = tempfile.mkstemp()219        os.close(out_file)220        os.remove(out_path)221        junitxml_proc = subprocess.Popen([self.command, out_path],222                                         cwd=self.repo_root,223                                         stdout=subprocess.PIPE,224                                         stderr=subprocess.PIPE,225                                         encoding='utf8')226        _, stderr = junitxml_proc.communicate()227        self.assertEqual(1, junitxml_proc.returncode)228        self.assertEqual(...

test_scheduler.py

Source:test_scheduler.py

...72        test_ids = frozenset(['zero1', 'zero2'])73        partitions = scheduler.partition_tests(test_ids, 2, repo, None)74        self.assertEqual(1, len(partitions[0]))75        self.assertEqual(1, len(partitions[1]))76    def test_partition_tests_with_grouping(self):77        repo = memory.RepositoryFactory().initialise('memory:')78        result = repo.get_inserter()79        result.startTestRun()80        self._add_timed_test("TestCase1.slow", 3, result)81        self._add_timed_test("TestCase2.fast1", 1, result)82        self._add_timed_test("TestCase2.fast2", 1, result)83        result.stopTestRun()84        test_ids = frozenset(['TestCase1.slow', 'TestCase1.fast',85                              'TestCase1.fast2', 'TestCase2.fast1',86                              'TestCase3.test1', 'TestCase3.test2',87                              'TestCase2.fast2', 'TestCase4.test',88                              'testdir.testfile.TestCase5.test'])89        def group_id(test_id, regex=re.compile('TestCase[0-5]')):90            match = regex.match(test_id)...

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.