Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _start_http_server method in fMBT

Best Python code snippet using fMBT_python

test_server.py

Source:test_server.py

...9        if self.server is not None:10            self.server.stop()11    def test_should_call_get_action(self):12        # given13        self._start_http_server(JsonGet("/test", lambda params: {'status': 'ok'}))14        # when15        result = requests.get(f"http://localhost:{PORT}/test")16        # then17        assert 200 == result.status_code18        assert 'ok' == result.json()['status']19    def test_should_call_get_action_with_params(self):20        # given21        self._start_http_server(JsonGet("/test", lambda params: {'params': params}))22        # when23        result = requests.get(f"http://localhost:{PORT}/test?a=0&b=6&b=7")24        # then25        assert 200 == result.status_code26        assert ['0'] == result.json()['params']['a']27        assert ['6', '7'] == result.json()['params']['b']28    def test_should_call_post_action(self):29        # given30        self._start_http_server(JsonPost("/test", lambda params, body: {'status': 'ok'}))31        # when32        result = requests.post(f"http://localhost:{PORT}/test")33        # then34        assert 200 == result.status_code35        assert 'ok' == result.json()['status']36    def test_should_call_post_action_with_params_and_body(self):37        # given38        self._start_http_server(JsonPost("/test", lambda params, body: {'params': params, 'body': body}))39        # when40        result = requests.post(f"http://localhost:{PORT}/test?a=0&b=6&b=7", data='{"hello": "POST BODY"}')41        # then42        assert 200 == result.status_code43        assert ['0'] == result.json()['params']['a']44        assert ['6', '7'] == result.json()['params']['b']45        assert 'POST BODY' == result.json()['body']['hello']46    def test_should_call_post_action_without_body(self):47        # given48        self._start_http_server(JsonPost("/test", lambda params, body: {'body': body}))49        # when50        result = requests.post(f"http://localhost:{PORT}/test?a=0&b=6&b=7")51        # then52        assert 200 == result.status_code53        assert '' == result.json()['body']54    def test_should_return_4xx_when_post_action_with_unparsable_body(self):55        # given56        self._start_http_server(JsonPost("/test", lambda params, body: {'body': body}))57        # when58        result = requests.post(f"http://localhost:{PORT}/test", data='non-json string')59        # then60        assert 400 == result.status_code61    def test_should_redirect(self):62        # given63        self._start_http_server(Redirect("/test", "http://example.com"))64        # when:65        result = requests.get(f'http://localhost:{PORT}/test', allow_redirects=False)66        # then67        assert 301 == result.status_code68        assert 'http://example.com' == result.headers['Location']69    @pytest.mark.parametrize("filename, expected_length, expected_content_type", [70        ("text.txt", 9, "text/plain"),71        ("image.png", 81618, "image/png"),72        ("document.pdf", 38078, "application/pdf")73    ])74    def test_should_serve_static_resources(self, filename, expected_length, expected_content_type):75        # given76        self._start_http_server(StaticResources("/static", "./test-resources"))77        # when:78        result = requests.get(f'http://localhost:{PORT}/static/{filename}')79        # then80        assert 200 == result.status_code81        assert expected_length == len(result.content)82        assert expected_content_type == result.headers['Content-Type']83    def test_should_serve_static_resources_regardless_slash(self):84        # given85        self._start_http_server(StaticResources("/static/", "./test-resources/"))86        # when:87        result = requests.get(f'http://localhost:{PORT}/static/text.txt')88        # then89        assert 200 == result.status_code90        assert 9 == len(result.content)91    def test_should_return_404_if_static_resources_not_found(self):92        # given93        self._start_http_server(StaticResources("/static/", "./test-resources/"))94        # when:95        result = requests.get(f'http://localhost:{PORT}/static/nonexisting.txt')96        # then97        assert 404 == result.status_code98    def test_should_return_404_if_action_not_found(self):99        # given100        self._start_http_server(JsonGet("/test", lambda params: {'status': 'ok'}))101        # when:102        result = requests.get(f'http://localhost:{PORT}/nonexisting-endpoint')103        # then104        assert 404 == result.status_code105    def test_should_find_action_in_order_of_declaration(self):106        # given107        self._start_http_server([108            JsonGet("/test", lambda params: {'status': '1'}),109            JsonGet("/test", lambda params: {'status': '2'}),110        ])111        # when:112        result = requests.get(f'http://localhost:{PORT}/test')113        # then114        assert 200 == result.status_code115        assert '1' == result.json()['status']116    def _start_http_server(self, actions):117        actions = actions if isinstance(actions, list) else [actions]118        self.server = http_server(PORT, actions, thread_count=1)...

server.py

Source:server.py

...9from hypercorn.config import Config as HypercornConfig10from .app import make_application11from .config import Config12LOGGER = logging.getLogger(__name__)13async def _start_http_server(app: Application, config: Config) -> None:14    """Start the hypercorn ASGI server"""15    web_config = HypercornConfig()16    web_config.bind = [f'{config.app.host}:{config.app.port}']17    if config.app.tls is not None and config.app.tls.is_enabled:18        web_config.keyfile = config.app.tls.keyfile19        web_config.certfile = config.app.tls.certfile20    await serve(21        app,  # type: ignore22        web_config23    )24def _initialise_logging(config: Config) -> None:25    if config.log is not None:26        logging.config.dictConfig(config.log)27def _parse_args(argv: list):28    """Parse the command line args"""29    parser = argparse.ArgumentParser(30        description='Order File Service',31        add_help=False)32    parser.add_argument(33        '--help', help='Show usage',34        action='help')35    parser.add_argument(36        '-f', '--config-file', help='Path to the configuration file.',37        action="store", dest='CONFIG_FILE')38    return parser.parse_args(argv)39async def start_server(argv: list) -> None:40    args = _parse_args(argv[1:])41    config = Config.load(args.CONFIG_FILE)42    app = make_application(config)43    await _start_http_server(app, config)...

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.