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

How to use _run_server method in autotest

Best Python code snippet using autotest_python

test_endtoend.py

Source:test_endtoend.py

...139            self._options.server_port = self.test_port140    def _run_python_command(self, commandline, stdout=None, stderr=None):141        return subprocess.Popen([sys.executable] + commandline, close_fds=True,142                                stdout=stdout, stderr=stderr)143    def _run_server(self, allow_draft75=False):144        args = [self.standalone_command,145                '-H', 'localhost',146                '-V', 'localhost',147                '-p', str(self.test_port),148                '-P', str(self.test_port),149                '-d', self.document_root]150        # Inherit the level set to the root logger by test runner.151        root_logger = logging.getLogger()152        log_level = root_logger.getEffectiveLevel()153        if log_level != logging.NOTSET:154            args.append('--log-level')155            args.append(logging.getLevelName(log_level).lower())156        if allow_draft75:157            args.append('--allow-draft75')158        return self._run_python_command(args,159                                        stderr=self.server_stderr)160    def _kill_process(self, pid):161        if sys.platform in ('win32', 'cygwin'):162            subprocess.call(163                ('taskkill.exe', '/f', '/pid', str(pid)), close_fds=True)164        else:165            os.kill(pid, signal.SIGKILL)166    def _run_hybi_test_with_client_options(self, test_function, options):167        server = self._run_server()168        try:169            # TODO(tyoshino): add some logic to poll the server until it170            # becomes ready171            time.sleep(0.2)172            client = client_for_testing.create_client(options)173            try:174                test_function(client)175            finally:176                client.close_socket()177        finally:178            self._kill_process(server.pid)179    def _run_hybi_test(self, test_function):180        self._run_hybi_test_with_client_options(test_function, self._options)181    def _run_hybi_deflate_test(self, test_function):182        server = self._run_server()183        try:184            time.sleep(0.2)185            self._options.enable_deflate_stream()186            client = client_for_testing.create_client(self._options)187            try:188                test_function(client)189            finally:190                client.close_socket()191        finally:192            self._kill_process(server.pid)193    def _run_hybi_deflate_frame_test(self, test_function):194        server = self._run_server()195        try:196            time.sleep(0.2)197            self._options.enable_deflate_frame()198            client = client_for_testing.create_client(self._options)199            try:200                test_function(client)201            finally:202                client.close_socket()203        finally:204            self._kill_process(server.pid)205    def _run_hybi_close_with_code_and_reason_test(self, test_function, code,206                                                  reason):207        server = self._run_server()208        try:209            time.sleep(0.2)210            client = client_for_testing.create_client(self._options)211            try:212                test_function(client, code, reason)213            finally:214                client.close_socket()215        finally:216            self._kill_process(server.pid)217    def _run_hybi_http_fallback_test(self, options, status):218        server = self._run_server()219        try:220            time.sleep(0.2)221            client = client_for_testing.create_client(options)222            try:223                client.connect()224                self.fail('Could not catch HttpStatusException')225            except client_for_testing.HttpStatusException, e:226                self.assertEqual(status, e.status)227            except Exception, e:228                self.fail('Catch unexpected exception')229            finally:230                client.close_socket()231        finally:232            self._kill_process(server.pid)233    def _run_hybi_mux_test(self, test_function):234        server = self._run_server()235        try:236            time.sleep(0.2)237            client = mux_client_for_testing.MuxClient(self._options)238            try:239                test_function(client)240            finally:241                client.close_socket()242        finally:243            self._kill_process(server.pid)244    def test_echo(self):245        self._run_hybi_test(_echo_check_procedure)246    def test_echo_binary(self):247        self._run_hybi_test(_echo_check_procedure_with_binary)248    def test_echo_server_close(self):249        self._run_hybi_test(_echo_check_procedure_with_goodbye)250    def test_unmasked_frame(self):251        self._run_hybi_test(_unmasked_frame_check_procedure)252    def test_echo_deflate(self):253        self._run_hybi_deflate_test(_echo_check_procedure)254    def test_echo_deflate_server_close(self):255        self._run_hybi_deflate_test(_echo_check_procedure_with_goodbye)256    def test_echo_deflate_frame(self):257        self._run_hybi_deflate_frame_test(_echo_check_procedure)258    def test_echo_deflate_frame_server_close(self):259        self._run_hybi_deflate_frame_test(260            _echo_check_procedure_with_goodbye)261    def test_echo_close_with_code_and_reason(self):262        self._options.resource = '/close'263        self._run_hybi_close_with_code_and_reason_test(264            _echo_check_procedure_with_code_and_reason, 3333, 'sunsunsunsun')265    def test_echo_close_with_empty_body(self):266        self._options.resource = '/close'267        self._run_hybi_close_with_code_and_reason_test(268            _echo_check_procedure_with_code_and_reason, None, '')269    def test_mux_echo(self):270        self._run_hybi_mux_test(_mux_echo_check_procedure)271    def test_close_on_protocol_error(self):272        """Tests that the server sends a close frame with protocol error status273        code when the client sends data with some protocol error.274        """275        def test_function(client):276            client.connect()277            # Intermediate frame without any preceding start of fragmentation278            # frame.279            client.send_frame_of_arbitrary_bytes('\x80\x80', '')280            client.assert_receive_close(281                client_for_testing.STATUS_PROTOCOL_ERROR)282        self._run_hybi_test(test_function)283    def test_close_on_unsupported_frame(self):284        """Tests that the server sends a close frame with unsupported operation285        status code when the client sends data asking some operation that is286        not supported by the server.287        """288        def test_function(client):289            client.connect()290            # Text frame with RSV3 bit raised.291            client.send_frame_of_arbitrary_bytes('\x91\x80', '')292            client.assert_receive_close(293                client_for_testing.STATUS_UNSUPPORTED_DATA)294        self._run_hybi_test(test_function)295    def test_close_on_invalid_frame(self):296        """Tests that the server sends a close frame with invalid frame payload297        data status code when the client sends an invalid frame like containing298        invalid UTF-8 character.299        """300        def test_function(client):301            client.connect()302            # Text frame with invalid UTF-8 string.303            client.send_message('\x80', raw=True)304            client.assert_receive_close(305                client_for_testing.STATUS_INVALID_FRAME_PAYLOAD_DATA)306        self._run_hybi_test(test_function)307    def _run_hybi00_test(self, test_function):308        server = self._run_server()309        try:310            time.sleep(0.2)311            client = client_for_testing.create_client_hybi00(self._options)312            try:313                test_function(client)314            finally:315                client.close_socket()316        finally:317            self._kill_process(server.pid)318    def test_echo_hybi00(self):319        self._run_hybi00_test(_echo_check_procedure)320    def test_echo_server_close_hybi00(self):321        self._run_hybi00_test(_echo_check_procedure_with_goodbye)322    def _run_hixie75_test(self, test_function):323        server = self._run_server(allow_draft75=True)324        try:325            time.sleep(0.2)326            client = client_for_testing.create_client_hixie75(self._options)327            try:328                test_function(client)329            finally:330                client.close_socket()331        finally:332            self._kill_process(server.pid)333    def test_echo_hixie75(self):334        """Tests that the server can talk draft-hixie-thewebsocketprotocol-75335        protocol.336        """337        def test_function(client):338            client.connect()339            client.send_message('test')340            client.assert_receive('test')341        self._run_hixie75_test(test_function)342    def test_echo_server_close_hixie75(self):343        """Tests that the server can talk draft-hixie-thewebsocketprotocol-75344        protocol. At the end of message exchanging, the client sends a keyword345        message that requests the server to close the connection, and then346        checks if the connection is really closed.347        """348        def test_function(client):349            client.connect()350            client.send_message('test')351            client.assert_receive('test')352            client.send_message(_GOODBYE_MESSAGE)353            client.assert_receive(_GOODBYE_MESSAGE)354        self._run_hixie75_test(test_function)355    # TODO(toyoshim): Add tests to verify invalid absolute uri handling like356    # host unmatch, port unmatch and invalid port description (':' without port357    # number).358    def test_absolute_uri(self):359        """Tests absolute uri request."""360        options = self._options361        options.resource = 'ws://localhost:%d/echo' % options.server_port362        self._run_hybi_test_with_client_options(_echo_check_procedure, options)363    def test_origin_check(self):364        """Tests http fallback on origin check fail."""365        options = self._options366        options.resource = '/origin_check'367        # Server shows warning message for http 403 fallback. This warning368        # message is confusing. Following pipe disposes warning messages.369        self.server_stderr = subprocess.PIPE370        self._run_hybi_http_fallback_test(options, 403)371    def test_version_check(self):372        """Tests http fallback on version check fail."""373        options = self._options374        options.version = 99375        self.server_stderr = subprocess.PIPE376        self._run_hybi_http_fallback_test(options, 400)377    def _check_example_echo_client_result(378        self, expected, stdoutdata, stderrdata):379        actual = stdoutdata.decode("utf-8")380        if actual != expected:381            raise Exception('Unexpected result on example echo client: '382                            '%r (expected) vs %r (actual)' %383                            (expected, actual))384        if stderrdata is not None:385            raise Exception('Unexpected error message on example echo '386                            'client: %r' % stderrdata)387    def test_example_echo_client(self):388        """Tests that the echo_client.py example can talk with the server."""389        server = self._run_server()390        try:391            time.sleep(0.2)392            client_command = os.path.join(393                self.top_dir, 'example', 'echo_client.py')394            args = [client_command,395                    '-p', str(self._options.server_port)]396            client = self._run_python_command(args, stdout=subprocess.PIPE)397            stdoutdata, stderrdata = client.communicate()398            expected = ('Send: Hello\n' 'Recv: Hello\n'399                u'Send: \u65e5\u672c\n' u'Recv: \u65e5\u672c\n'400                'Send close\n' 'Recv ack\n')401            self._check_example_echo_client_result(402                expected, stdoutdata, stderrdata)403            # Process a big message for which extended payload length is used....

app.py

Source:app.py

...68            ],69            [Input("start", "on")],70            [State("proposal", "value"), State("run-type", "value")],71        )72        def start_run_server(state, proposal, run_type):73            info, p_dis, r_dis = "", False, False74            if state:75                if not (proposal and run_type):76                    info = "Either Folder or run type missing"77                    return [info], p_dis, r_dis78                proposal = find_proposal(proposal, data=run_type)79                self._run_server = ProposalMonitor(proposal, self._data_queue)80                try:81                    print("Start ", self._run_server)82                    self._run_server.start()83                    info = f"Checking Files of type {run_type} in {proposal}"84                except Exception as ex:85                    info = repr(ex)86                p_dis, r_dis = True, True...

test_cli.py

Source:test_cli.py

1from click.testing import CliRunner2from mock import mock3import pytest4from mlflow.cli import run, server, ui5from mlflow.server import handlers6def test_server_static_prefix_validation():7    with mock.patch("mlflow.cli._run_server") as run_server_mock:8        CliRunner().invoke(server)9        run_server_mock.assert_called_once()10    with mock.patch("mlflow.cli._run_server") as run_server_mock:11        CliRunner().invoke(server, ["--static-prefix", "/mlflow"])12        run_server_mock.assert_called_once()13    with mock.patch("mlflow.cli._run_server") as run_server_mock:14        result = CliRunner().invoke(server, ["--static-prefix", "mlflow/"])15        assert "--static-prefix must begin with a '/'." in result.output16        run_server_mock.assert_not_called()17    with mock.patch("mlflow.cli._run_server") as run_server_mock:18        result = CliRunner().invoke(server, ["--static-prefix", "/mlflow/"])19        assert "--static-prefix should not end with a '/'." in result.output20        run_server_mock.assert_not_called()21def test_server_default_artifact_root_validation():22    with mock.patch("mlflow.cli._run_server") as run_server_mock:23        result = CliRunner().invoke(server, ["--backend-store-uri", "sqlite:///my.db"])24        assert result.output.startswith("Option 'default-artifact-root' is required")25        run_server_mock.assert_not_called()26@pytest.mark.parametrize("command", [server, ui])27def test_tracking_uri_validation_failure(command):28    handlers._store = None29    with mock.patch("mlflow.cli._run_server") as run_server_mock:30        # SQLAlchemy expects postgresql:// not postgres://31        CliRunner().invoke(command,32                           ["--backend-store-uri", "postgres://user:pwd@host:5432/mydb",33                            "--default-artifact-root", "./mlruns"])34        run_server_mock.assert_not_called()35@pytest.mark.parametrize("command", [server, ui])36def test_tracking_uri_validation_sql_driver_uris(command):37    handlers._store = None38    with mock.patch("mlflow.cli._run_server") as run_server_mock,\39            mock.patch("mlflow.store.sqlalchemy_store.SqlAlchemyStore") as sql_store:40        CliRunner().invoke(command,41                           ["--backend-store-uri", "mysql+pymysql://user:pwd@host:5432/mydb",42                            "--default-artifact-root", "./mlruns"])43        sql_store.assert_called_once_with("mysql+pymysql://user:pwd@host:5432/mydb", "./mlruns")44        run_server_mock.assert_called()45def test_mlflow_run():46    with mock.patch("mlflow.cli.projects") as mock_projects:47        result = CliRunner().invoke(run)48        mock_projects.run.assert_not_called()49        assert 'Missing argument "URI"' in result.output50    with mock.patch("mlflow.cli.projects") as mock_projects:51        CliRunner().invoke(run, ["project_uri"])52        mock_projects.run.assert_called_once()53    with mock.patch("mlflow.cli.projects") as mock_projects:54        CliRunner().invoke(run, ["--experiment-id", "5", "project_uri"])55        mock_projects.run.assert_called_once()56    with mock.patch("mlflow.cli.projects") as mock_projects:57        CliRunner().invoke(run, ["--experiment-name", "random name", "project_uri"])58        mock_projects.run.assert_called_once()59    with mock.patch("mlflow.cli.projects") as mock_projects:60        result = CliRunner().invoke(run, ["--experiment-id", "51",61                                          "--experiment-name", "name blah", "uri"])62        mock_projects.run.assert_not_called()...

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.