Best Python code snippet using molecule_python
test_parallel.py
Source:test_parallel.py  
1# Copyright 2017 Canonical Ltd.  This software is licensed under the2# GNU Affero General Public License version 3 (see the file LICENSE).3import os4import random5from unittest.mock import ANY6import junitxml7import subunit8import testtools9from testtools import (10    ExtendedToOriginalDecorator,11    MultiTestResult,12    TextTestResult,13)14from testtools.matchers import (15    AfterPreprocessing,16    Equals,17    Is,18    IsInstance,19    MatchesAll,20    MatchesListwise,21    MatchesSetwise,22    MatchesStructure,23)24from maastesting import parallel25from maastesting.fixtures import CaptureStandardIO26from maastesting.matchers import (27    DocTestMatches,28    MockCalledOnceWith,29    MockNotCalled,30)31from maastesting.testcase import MAASTestCase32class TestSelectorArguments(MAASTestCase):33    """Tests for arguments that select scripts."""34    def setUp(self):35        super().setUp()36        self.stdio = self.useFixture(CaptureStandardIO())37        self.patch_autospec(parallel, "test")38        parallel.test.return_value = True39    def assertScriptsMatch(self, *matchers):40        self.assertThat(parallel.test, MockCalledOnceWith(ANY, ANY, ANY))41        suite, results, processes = parallel.test.call_args[0]42        self.assertThat(43            suite, AfterPreprocessing(list, MatchesSetwise(*matchers))44        )45    def test_all_scripts_are_selected_when_no_selectors(self):46        sysexit = self.assertRaises(SystemExit, parallel.main, [])47        self.assertThat(sysexit.code, Equals(0))48        self.assertScriptsMatch(49            MatchesUnselectableScript("bin/test.region.legacy"),50            MatchesSelectableScript("rack"),51            MatchesSelectableScript("region"),52        )53    def test_scripts_can_be_selected_by_path(self):54        sysexit = self.assertRaises(55            SystemExit,56            parallel.main,57            [58                "src/provisioningserver/002",59                "src/maasserver/003",60                "src/metadataserver/004",61            ],62        )63        self.assertThat(sysexit.code, Equals(0))64        self.assertScriptsMatch(65            MatchesSelectableScript("rack", "src/provisioningserver/002"),66            MatchesSelectableScript(67                "region", "src/maasserver/003", "src/metadataserver/004"68            ),69        )70    def test_scripts_can_be_selected_by_module(self):71        sysexit = self.assertRaises(72            SystemExit,73            parallel.main,74            [75                "provisioningserver.002",76                "maasserver.003",77                "metadataserver.004",78            ],79        )80        self.assertThat(sysexit.code, Equals(0))81        self.assertScriptsMatch(82            MatchesSelectableScript("rack", "provisioningserver.002"),83            MatchesSelectableScript(84                "region", "maasserver.003", "metadataserver.004"85            ),86        )87def MatchesUnselectableScript(what, *selectors):88    return MatchesAll(89        IsInstance(parallel.TestScriptUnselectable),90        MatchesStructure.byEquality(script=what),91        first_only=True,92    )93def MatchesSelectableScript(what, *selectors):94    return MatchesAll(95        IsInstance(parallel.TestScriptSelectable),96        MatchesStructure.byEquality(97            script="bin/test.%s" % what, selectors=selectors98        ),99        first_only=True,100    )101class TestSubprocessArguments(MAASTestCase):102    """Tests for arguments that adjust subprocess behaviour."""103    def setUp(self):104        super().setUp()105        self.stdio = self.useFixture(CaptureStandardIO())106        self.patch_autospec(parallel, "test")107        parallel.test.return_value = True108    def test_defaults(self):109        sysexit = self.assertRaises(SystemExit, parallel.main, [])110        self.assertThat(sysexit.code, Equals(0))111        self.assertThat(112            parallel.test,113            MockCalledOnceWith(ANY, ANY, max(os.cpu_count() - 2, 2)),114        )115    def test_subprocess_count_can_be_specified(self):116        count = random.randrange(100, 1000)117        sysexit = self.assertRaises(118            SystemExit, parallel.main, ["--subprocesses", str(count)]119        )120        self.assertThat(sysexit.code, Equals(0))121        self.assertThat(parallel.test, MockCalledOnceWith(ANY, ANY, count))122    def test_subprocess_count_of_less_than_1_is_rejected(self):123        sysexit = self.assertRaises(124            SystemExit, parallel.main, ["--subprocesses", "0"]125        )126        self.assertThat(sysexit.code, Equals(2))127        self.assertThat(parallel.test, MockNotCalled())128        self.assertThat(129            self.stdio.getError(),130            DocTestMatches(131                "usage: ... argument --subprocesses: 0 is not 1 or greater"132            ),133        )134    def test_subprocess_count_non_numeric_is_rejected(self):135        sysexit = self.assertRaises(136            SystemExit, parallel.main, ["--subprocesses", "foo"]137        )138        self.assertThat(sysexit.code, Equals(2))139        self.assertThat(parallel.test, MockNotCalled())140        self.assertThat(141            self.stdio.getError(),142            DocTestMatches(143                "usage: ... argument --subprocesses: 'foo' is not an integer"144            ),145        )146    def test_subprocess_per_core_can_be_specified(self):147        sysexit = self.assertRaises(148            SystemExit, parallel.main, ["--subprocess-per-core"]149        )150        self.assertThat(sysexit.code, Equals(0))151        self.assertThat(152            parallel.test, MockCalledOnceWith(ANY, ANY, os.cpu_count())153        )154    def test_subprocess_count_and_per_core_cannot_both_be_specified(self):155        sysexit = self.assertRaises(156            SystemExit,157            parallel.main,158            ["--subprocesses", "3", "--subprocess-per-core"],159        )160        self.assertThat(sysexit.code, Equals(2))161        self.assertThat(parallel.test, MockNotCalled())162        self.assertThat(163            self.stdio.getError(),164            DocTestMatches(165                "usage: ... argument --subprocess-per-core: not allowed with "166                "argument --subprocesses"167            ),168        )169class TestEmissionArguments(MAASTestCase):170    """Tests for arguments that adjust result emission behaviour."""171    def setUp(self):172        super().setUp()173        self.stdio = self.useFixture(CaptureStandardIO())174        self.patch_autospec(parallel, "test")175        parallel.test.return_value = True176    def test_results_are_human_readable_by_default(self):177        sysexit = self.assertRaises(SystemExit, parallel.main, [])178        self.assertThat(sysexit.code, Equals(0))179        self.assertThat(parallel.test, MockCalledOnceWith(ANY, ANY, ANY))180        _, result, _ = parallel.test.call_args[0]181        self.assertThat(182            result,183            IsMultiResultOf(184                IsInstance(TextTestResult),185                IsInstance(testtools.TestByTestResult),186            ),187        )188    def test_results_can_be_explicitly_specified_as_human_readable(self):189        sysexit = self.assertRaises(190            SystemExit, parallel.main, ["--emit-human"]191        )192        self.assertThat(sysexit.code, Equals(0))193        self.assertThat(parallel.test, MockCalledOnceWith(ANY, ANY, ANY))194        _, result, _ = parallel.test.call_args[0]195        self.assertThat(196            result,197            IsMultiResultOf(198                IsInstance(TextTestResult),199                IsInstance(testtools.TestByTestResult),200            ),201        )202    def test_results_can_be_specified_as_subunit(self):203        sysexit = self.assertRaises(204            SystemExit, parallel.main, ["--emit-subunit"]205        )206        self.assertThat(sysexit.code, Equals(0))207        self.assertThat(parallel.test, MockCalledOnceWith(ANY, ANY, ANY))208        _, result, _ = parallel.test.call_args[0]209        self.assertThat(result, IsInstance(subunit.TestProtocolClient))210        self.assertThat(211            result, MatchesStructure(_stream=Is(self.stdio.stdout.buffer))212        )213    def test_results_can_be_specified_as_junit(self):214        sysexit = self.assertRaises(215            SystemExit, parallel.main, ["--emit-junit"]216        )217        self.assertThat(sysexit.code, Equals(0))218        self.assertThat(parallel.test, MockCalledOnceWith(ANY, ANY, ANY))219        _, result, _ = parallel.test.call_args[0]220        self.assertThat(result, IsInstance(junitxml.JUnitXmlResult))221        self.assertThat(222            result, MatchesStructure(_stream=Is(self.stdio.stdout))223        )224def IsMultiResultOf(*results):225    """Match a `MultiTestResult` wrapping the given results."""226    return MatchesAll(227        IsInstance(MultiTestResult),228        MatchesStructure(229            _results=MatchesListwise(230                [231                    MatchesAll(232                        IsInstance(ExtendedToOriginalDecorator),233                        MatchesStructure(decorated=matcher),234                        first_only=True,235                    )236                    for matcher in results237                ]238            )239        ),240        first_only=True,...test_dkconfig.py
Source:test_dkconfig.py  
...4from mock import MagicMock5import sys6from dkconfig import main, dkconfig7@pytest.fixture8def sysexit(monkeypatch):9    mock_exit = MagicMock()10    monkeypatch.setattr(sys, 'exit', mock_exit)11    return mock_exit12def test_debug(tmpdir, sysexit, capsys):13    file = tmpdir.join("foo.ini")14    main("%s get header key -d" % file)15    sysexit.assert_called_with(1)16    out, err = capsys.readouterr()17    err = err.replace('\\', '').strip()18    file = str(file).replace('\\','')19    # compare = """ARGS: Namespace(command='get', debug=True, filename=['%s']) ['header', 'key']""" % file20    # py39 = """ARGS: Namespace(filename=['%s'], command='get', debug=True) ['header', 'key']""" % file21    assert err.endswith(" ['header', 'key']")22    err = err[:-len(" ['header', 'key']")]...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
