Method test_custom_arguments

locust/test/test_main.py:80–118

(self)

    @unittest.skipIf(IS_WINDOWS, reason="Signal handling on windows is hard")
    def test_custom_arguments(self):
        port = get_free_tcp_port()
        with temporary_file(
            content=textwrap.dedent(
                """
            from locust import User, task, constant, events
            @events.init_command_line_parser.add_listener
            def _(parser, **kw):
                parser.add_argument("--custom-string-arg")

            class TestUser(User):
                wait_time = constant(10)
                @task
                def my_task(self):
                    print(self.environment.parsed_options.custom_string_arg)
        """
            )
        ) as file_path:
            # print(subprocess.check_output(["cat", file_path]))
            with TestProcess(
                f"locust -f {file_path} --custom-string-arg command_line_value --web-port {port}",
            ) as tp:
                tp.expect("Starting Locust")
                wait_for_server(f"http://127.0.0.1:{port}/swarm")
                requests.post(
                    f"http://127.0.0.1:{port}/swarm",
                    data={
                        "user_count": 1,
                        "spawn_rate": 1,
                        "host": "https://localhost",
                        "custom_string_arg": "web_form_value",
                    },
                )
                tp.expect('All users spawned: {"TestUser": 1} (1 total users)')
                tp.expect("web_form_value", stream="stdout")
                tp.terminate()
                tp.expect("Shutting down")
                tp.expect("Aggregated")
                tp.not_expect_any("command_line_value")

    @unittest.skipIf(IS_WINDOWS, reason="Signal handling on windows is hard")
    def test_custom_arguments_in_file(self):