KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _test_molotov method in Molotov

Best Python code snippet using molotov_python
test_run.py
Source:test_run.py
...106    @dedicatedloop107    def test_main(self):108        with set_args("molotov", "-cq", "-d", "1", "molotov/tests/example.py"):109            main()110    def _test_molotov(self, *args):111        if "--duration" not in args and "-d" not in args:112            args = list(args) + ["--duration", "10"]113        rc = 0114        with set_args("molotov", *args) as (stdout, stderr):115            try:116                main()117            except SystemExit as e:118                rc = e.code119        return stdout.read().strip(), stderr.read().strip(), rc120    @dedicatedloop121    def test_version(self):122        stdout, stderr, rc = self._test_molotov("--version")123        self.assertEqual(stdout, __version__)124    @dedicatedloop125    def test_empty_scenario(self):126        stdout, stderr, rc = self._test_molotov("")127        self.assertTrue("Cannot import" in stdout)128    @dedicatedloop129    def test_config_no_scenario(self):130        stdout, stderr, rc = self._test_molotov("-c", "--config", _CONFIG, "DONTEXIST")131        wanted = "Can't find 'DONTEXIST' in the config"132        self.assertTrue(wanted in stdout)133    @dedicatedloop134    def test_config_verbose_quiet(self):135        stdout, stderr, rc = self._test_molotov("-qv", "--config", _CONFIG)136        wanted = "You can't"137        self.assertTrue(wanted in stdout)138    @dedicatedloop139    def test_config_no_scenario_found(self):140        stdout, stderr, rc = self._test_molotov("-c", "molotov.tests.test_run")141        wanted = "No scenario was found"142        self.assertTrue(wanted in stdout)143    @dedicatedloop144    def test_config_no_single_mode_found(self):145        @scenario(weight=10)146        async def not_me(session):147            _RES.append(3)148        stdout, stderr, rc = self._test_molotov(149            "-c", "-s", "blah", "molotov.tests.test_run"150        )151        wanted = "Can't find"152        self.assertTrue(wanted in stdout)153    @dedicatedloop154    def test_name(self):155        @scenario(weight=10)156        async def here_three(session):157            _RES.append(3)158        @scenario(weight=30, name="me")159        async def here_four(session):160            _RES.append(4)161        stdout, stderr, rc = self._test_molotov(162            "-cx", "--max-runs", "2", "-s", "me", "molotov.tests.test_run"163        )164        wanted = "SUCCESSES: 2"165        self.assertTrue(wanted in stdout)166        self.assertTrue(_RES, [4, 4])167    @dedicatedloop168    def test_single_mode(self):169        @scenario(weight=10)170        async def here_three(session):171            _RES.append(3)172        stdout, stderr, rc = self._test_molotov(173            "-cx", "--max-runs", "2", "-s", "here_three", "molotov.tests.test_run"174        )175        wanted = "SUCCESSES: 2"176        self.assertTrue(wanted in stdout)177    @dedicatedloop178    def test_fail_mode_pass(self):179        @scenario(weight=10)180        async def here_three(session):181            _RES.append(3)182        stdout, stderr, rc = self._test_molotov(183            "-cx",184            "--max-runs",185            "2",186            "--fail",187            "1",188            "-s",189            "here_three",190            "molotov.tests.test_run",191        )192        wanted = "SUCCESSES: 2"193        self.assertTrue(wanted in stdout)194        self.assertEqual(rc, 0)195    @dedicatedloop196    def test_fail_mode_fail(self):197        @scenario(weight=10)198        async def here_three(session):199            assert False200        stdout, stderr, rc = self._test_molotov(201            "-cx",202            "--max-runs",203            "2",204            "--fail",205            "1",206            "-s",207            "here_three",208            "molotov.tests.test_run",209        )210        self.assertEqual(rc, 1)211    @only_pypy212    @dedicatedloop213    def test_uvloop_pypy(self):214        @scenario(weight=10)215        async def here_three(session):216            _RES.append(3)217        orig_import = __import__218        def import_mock(name, *args):219            if name == "uvloop":220                raise ImportError()221            return orig_import(name, *args)222        with patch("builtins.__import__", side_effect=import_mock):223            stdout, stderr, rc = self._test_molotov(224                "-cx",225                "--max-runs",226                "2",227                "-s",228                "here_three",229                "--uvloop",230                "molotov.tests.test_run",231            )232        wanted = "You can't use uvloop"233        self.assertTrue(wanted in stdout)234    @skip_pypy235    @dedicatedloop236    def test_uvloop_import_error(self):237        @scenario(weight=10)238        async def here_three(session):239            _RES.append(3)240        orig_import = __import__241        def import_mock(name, *args):242            if name == "uvloop":243                raise ImportError()244            return orig_import(name, *args)245        with patch("builtins.__import__", side_effect=import_mock):246            stdout, stderr, rc = self._test_molotov(247                "-cx",248                "--max-runs",249                "2",250                "--console-update",251                "0",252                "-s",253                "here_three",254                "--uvloop",255                "molotov.tests.test_run",256            )257        wanted = "You need to install uvloop"258        self.assertTrue(wanted in stdout)259    @skip_pypy260    @dedicatedloop261    def test_uvloop(self):262        try:263            import uvloop  # noqa264        except ImportError:265            return266        @scenario(weight=10)267        async def here_three(session):268            _RES.append(3)269        stdout, stderr, rc = self._test_molotov(270            "-cx",271            "--max-runs",272            "2",273            "-s",274            "here_three",275            "--uvloop",276            "molotov.tests.test_run",277        )278        wanted = "SUCCESSES: 2"279        self.assertTrue(wanted in stdout, stdout)280    @dedicatedloop281    def test_delay(self):282        with catch_sleep() as delay:283            @scenario(weight=10, delay=0.1)284            async def here_three(session):285                _RES.append(3)286            stdout, stderr, rc = self._test_molotov(287                "--delay",288                ".6",289                "--console-update",290                "0",291                "-cx",292                "--max-runs",293                "2",294                "-s",295                "here_three",296                "molotov.tests.test_run",297            )298            wanted = "SUCCESSES: 2"299            self.assertTrue(wanted in stdout, stdout)300            self.assertEqual(delay[:9], [1, 0.1, 1, 0.6, 1, 0.1, 1, 0.6, 1])301    @dedicatedloop302    def test_rampup(self):303        with catch_sleep() as delay:304            @scenario(weight=10)305            async def here_three(session):306                _RES.append(3)307            stdout, stderr, rc = self._test_molotov(308                "--ramp-up",309                "10",310                "--workers",311                "5",312                "--console-update",313                "0",314                "-cx",315                "--max-runs",316                "2",317                "-s",318                "here_three",319                "molotov.tests.test_run",320            )321            # workers should start every 2 seconds since322            # we have 5 workers and a ramp-up323            # the first one starts immediatly, then each worker324            # sleeps 2 seconds more.325            delay = [d for d in delay if d != 0]326            self.assertEqual(delay, [1, 2.0, 4.0, 6.0, 8.0, 1, 1])327            wanted = "SUCCESSES: 10"328            self.assertTrue(wanted in stdout, stdout)329    @dedicatedloop330    def test_sizing(self):331        _RES2["fail"] = 0332        _RES2["succ"] = 0333        with catch_sleep():334            @scenario()335            async def sizer(session):336                if random.randint(0, 20) == 1:337                    _RES2["fail"] += 1338                    raise AssertionError()339                else:340                    _RES2["succ"] += 1341            stdout, stderr, rc = self._test_molotov(342                "--sizing",343                "--console-update",344                "0",345                "--sizing-tolerance",346                "5",347                "-s",348                "sizer",349                "molotov.tests.test_run",350            )351        ratio = float(_RES2["fail"]) / float(_RES2["succ"]) * 100.0352        self.assertTrue(ratio < 14.75 and ratio >= 4.75, ratio)353        found = re.findall(r"obtained with (\d+) workers", stdout)354        assert int(found[0]) > 50355    @unittest.skipIf(os.name == "nt", "win32")356    @dedicatedloop357    def test_sizing_multiprocess(self):358        counters = SharedCounters("OK", "FAILED")359        with catch_sleep():360            @scenario()361            async def sizer(session):362                if random.randint(0, 10) == 1:363                    counters["FAILED"] += 1364                    raise AssertionError()365                else:366                    counters["OK"] += 1367            with set_args(368                "molotov",369                "--sizing",370                "-p",371                "2",372                "--sizing-tolerance",373                "5",374                "--console-update",375                "0",376                "-s",377                "sizer",378                "molotov.tests.test_run",379            ) as (stdout, stderr):380                try:381                    main()382                except SystemExit:383                    pass384            stdout, stderr = stdout.read().strip(), stderr.read().strip()385            # stdout, stderr, rc = self._test_molotov()386            ratio = (387                float(counters["FAILED"].value) / float(counters["OK"].value) * 100.0388            )389            self.assertTrue(ratio >= 4.75, ratio)390    @unittest.skipIf(os.name == "nt", "win32")391    @dedicatedloop_noclose392    def test_statsd_multiprocess(self):393        test_loop = asyncio.get_event_loop()394        @scenario()395        async def staty(session):396            get_context(session).statsd.increment("yopla")397        server = UDPServer("127.0.0.1", 9999, loop=test_loop)398        _stop = asyncio.Future()399        async def stop():400            await _stop401            await server.stop()402        server_task = asyncio.ensure_future(server.run())403        stop_task = asyncio.ensure_future(stop())404        args = self._get_args()405        args.verbose = 2406        args.processes = 2407        args.max_runs = 5408        args.duration = 1000409        args.statsd = True410        args.statsd_address = "udp://127.0.0.1:9999"411        args.single_mode = "staty"412        args.scenario = "molotov.tests.test_run"413        stream = io.StringIO()414        run(args, stream=stream)415        _stop.set_result(True)416        test_loop.run_until_complete(asyncio.gather(server_task, stop_task))417        udp = server.flush()418        incrs = 0419        for line in udp:420            for el in line.split(b"\n"):421                if el.strip() == b"":422                    continue423                incrs += 1424        # two processes making 5 run each425        # we want at least 5  here426        self.assertTrue(incrs > 5)427        stream.seek(0)428        output = stream.read()429        self.assertTrue("Happy breaking!" in output, output)430    @dedicatedloop431    def test_timed_sizing(self):432        _RES2["fail"] = 0433        _RES2["succ"] = 0434        _RES2["messed"] = False435        with catch_sleep():436            @scenario()437            async def sizer(session):438                if get_context(session).worker_id == 200 and not _RES2["messed"]:439                    # worker 2 will mess with the timer440                    # since we're faking all timers, the current441                    # time in the test is always around 0442                    # so to have now() - get_timer() > 60443                    # we need to set a negative value here444                    # to trick it445                    set_timer(-61)446                    _RES2["messed"] = True447                    _RES2["fail"] = _RES2["succ"] = 0448                if get_context(session).worker_id > 100:449                    # starting to introduce errors passed the 100th450                    if random.randint(0, 10) == 1:451                        _RES2["fail"] += 1452                        raise AssertionError()453                    else:454                        _RES2["succ"] += 1455                # forces a switch456                await asyncio.sleep(0)457            stdout, stderr, rc = self._test_molotov(458                "--sizing",459                "--sizing-tolerance",460                "5",461                "--console-update",462                "0",463                "-cs",464                "sizer",465                "molotov.tests.test_run",466            )467        ratio = float(_RES2["fail"]) / float(_RES2["succ"]) * 100.0468        self.assertTrue(ratio < 20.0 and ratio > 4.75, ratio)469    @unittest.skipIf(os.name == "nt", "win32")470    @dedicatedloop471    def test_sizing_multiprocess_interrupted(self):472        counters = SharedCounters("OK", "FAILED")473        @scenario()474        async def sizer(session):475            if random.randint(0, 10) == 1:476                counters["FAILED"] += 1477                raise AssertionError()478            else:479                counters["OK"] += 1480        async def _stop():481            await asyncio.sleep(2.0)482            os.kill(os.getpid(), signal.SIGINT)483        asyncio.ensure_future(_stop())484        stdout, stderr, rc = self._test_molotov(485            "--sizing",486            "-p",487            "3",488            "--sizing-tolerance",489            "90",490            "--console-update",491            "0",492            "-s",493            "sizer",494            "molotov.tests.test_run",495        )496        self.assertTrue("Sizing was not finished" in stdout)497    @dedicatedloop498    def test_use_extension(self):499        ext = os.path.join(_HERE, "example5.py")500        @scenario(weight=10)501        async def simpletest(session):502            async with session.get("http://localhost:8888") as resp:503                assert resp.status == 200504        with coserver():505            stdout, stderr, rc = self._test_molotov(506                "-cx",507                "--max-runs",508                "1",509                "--use-extension=" + ext,510                "-s",511                "simpletest",512                "molotov.tests.test_run",513            )514        self.assertTrue("=>" in stdout)515        self.assertTrue("<=" in stdout)516    @dedicatedloop517    def test_use_extension_fail(self):518        ext = os.path.join(_HERE, "exampleIDONTEXIST.py")519        @scenario(weight=10)520        async def simpletest(session):521            async with session.get("http://localhost:8888") as resp:522                assert resp.status == 200523        with coserver():524            stdout, stderr, rc = self._test_molotov(525                "-cx",526                "--max-runs",527                "1",528                "--use-extension=" + ext,529                "-s",530                "simpletest",531                "molotov.tests.test_run",532            )533        self.assertTrue("Cannot import" in stdout)534    @dedicatedloop535    def test_use_extension_module_name(self):536        ext = "molotov.tests.example5"537        @scenario(weight=10)538        async def simpletest(session):539            async with session.get("http://localhost:8888") as resp:540                assert resp.status == 200541        with coserver():542            stdout, stderr, rc = self._test_molotov(543                "-cx",544                "--max-runs",545                "1",546                "--use-extension=" + ext,547                "-s",548                "simpletest",549                "molotov.tests.test_run",550            )551        self.assertTrue("=>" in stdout)552        self.assertTrue("<=" in stdout)553    @dedicatedloop554    def test_use_extension_module_name_fail(self):555        ext = "IDONTEXTSIST"556        @scenario(weight=10)557        async def simpletest(session):558            async with session.get("http://localhost:8888") as resp:559                assert resp.status == 200560        with coserver():561            stdout, stderr, rc = self._test_molotov(562                "-cx",563                "--max-runs",564                "1",565                "--use-extension=" + ext,566                "-s",567                "simpletest",568                "molotov.tests.test_run",569            )570        self.assertTrue("Cannot import" in stdout)571    @dedicatedloop572    def test_quiet(self):573        @scenario(weight=10)574        async def here_three(session):575            _RES.append(3)576        stdout, stderr, rc = self._test_molotov(577            "-cx", "--max-runs", "1", "-q", "-s", "here_three", "molotov.tests.test_run"578        )579        self.assertEqual(stdout, "")580        self.assertEqual(stderr, "")581    @dedicatedloop_noclose582    def test_slow_server_force_shutdown(self):583        @scenario(weight=10)584        async def _one(session):585            async with session.get("http://localhost:8888/slow") as resp:586                assert resp.status == 200587                _RES.append(1)588        args = self._get_args()589        args.duration = 0.1590        args.verbose = 2591        args.max_runs = 1592        args.force_shutdown = True593        start = time.time()594        with coserver():595            run(args)596        # makes sure the test is stopped even if the server597        # hangs a socket598        self.assertTrue(time.time() - start < 4)599        self.assertTrue(len(_RES) == 0)600    @dedicatedloop_noclose601    def test_slow_server_graceful(self):602        @scenario(weight=10)603        async def _one(session):604            async with session.get("http://localhost:8888/slow") as resp:605                assert resp.status == 200606                _RES.append(1)607        args = self._get_args()608        args.duration = 0.1609        args.verbose = 2610        args.max_runs = 1611        # graceful shutdown on the other hand will wait612        # for the worker completion613        args.graceful_shutdown = True614        start = time.time()615        with coserver():616            run(args)617        # makes sure the test finishes618        self.assertTrue(time.time() - start > 5)619        self.assertTrue(len(_RES) == 1)620    @dedicatedloop621    def test_single_run(self):622        _RES = defaultdict(int)623        with catch_sleep():624            @scenario()625            async def one(session):626                _RES["one"] += 1627            @scenario()628            async def two(session):629                _RES["two"] += 1630            @scenario()631            async def three(session):632                _RES["three"] += 1633            stdout, stderr, rc = self._test_molotov(634                "--single-run", "molotov.tests.test_run",635            )636        assert rc == 0637        assert _RES["one"] == 1638        assert _RES["two"] == 1639        assert _RES["three"] == 1640    @dedicatedloop641    def _XXX_test_enable_dns(self, m_resolve):642        m_resolve.return_value = ("http://localhost", "http://localhost", "localhost")643        with catch_sleep():644            @scenario()645            async def one(session):646                async with session.get("http://localhost"):647                    pass648            stdout, stderr, rc = self._test_molotov(649                "--single-run", "molotov.tests.test_run",650            )651        m_resolve.assert_called()652    @dedicatedloop653    def xxx_test_disable_dns(self, m_resolve):654        with catch_sleep():655            @scenario()656            async def one(session):657                async with session.get("http://localhost"):658                    pass659            stdout, stderr, rc = self._test_molotov(660                "--disable-dns-resolve", "--single-run", "molotov.tests.test_run",661            )662        m_resolve.assert_not_called()663    @dedicatedloop664    def test_bug_121(self):665        PASSED = [0]666        with catch_sleep():667            @scenario()668            async def scenario_one(session):669                cookies = {670                    "csrftoken": "sometoken",671                    "dtk": "1234",672                    "djdt": "hide",673                    "sessionid": "5678",674                }675                boundary = "----WebKitFormBoundaryFTE"676                headers = {677                    "X-CSRFToken": "sometoken",678                    "Content-Type": "multipart/form-data; boundary={}".format(boundary),679                }680                data = json.dumps({"1": "xxx"})681                with aiohttp.MultipartWriter(682                    "form-data", boundary=boundary683                ) as mpwriter:684                    mpwriter.append(685                        data,686                        {687                            "Content-Disposition": 'form-data; name="json"; filename="blob"',688                            "Content-Type": "application/json",689                        },690                    )691                    async with session.post(692                        "http://localhost:8888",693                        data=mpwriter,694                        headers=headers,695                        cookies=cookies,696                    ) as resp:697                        res = await resp.text()698                        assert data in res699                        PASSED[0] += 1700            args = self._get_args()701            args.verbose = 2702            args.max_runs = 1703            with coserver():704                res = run(args)705            assert PASSED[0] == 1706            assert res["OK"] == 1707    @dedicatedloop708    def test_local_import(self):709        test = os.path.join(_HERE, "example9.py")710        with coserver():711            stdout, stderr, rc = self._test_molotov("--max-runs", "1", test)...
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.