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

How to use start_kinesis method in localstack

Best Python code snippet using localstack_python

kinesis_starter.py

Source:kinesis_starter.py

...6from localstack.utils.aws import aws_stack7from localstack.utils.serving import Server8LOG = logging.getLogger(__name__)9_server: Optional[Server] = None  # server singleton10def start_kinesis(port=None, update_listener=None, asynchronous=None) -> Server:11    """12    Creates a singleton of a Kinesis server and starts it on a new thread. Uses either Kinesis Mock or Kinesalite13    based on value of config.KINESIS_PROVIDER14    :param port: port to run server on. Selects an arbitrary available port if None.15    :param update_listener: an update listener instance for server proxy16    :param asynchronous: currently unused but required by localstack.services.plugins.Service.start().17    TODO: either make use of this param or refactor Service.start() to not pass it.18    :returns: A running Kinesis server instance19    :raises: ValueError: Value of config.KINESIS_PROVIDER is not recognized as one of "kinesis-mock" or "kinesalite"20    """21    global _server22    if not _server:23        if config.KINESIS_PROVIDER == "kinesis-mock":24            _server = kinesis_mock_server.create_kinesis_mock_server()25        elif config.KINESIS_PROVIDER == "kinesalite":26            _server = kinesalite_server.create_kinesalite_server()27        else:28            raise ValueError('Unsupported Kinesis provider "%s"' % config.KINESIS_PROVIDER)29    _server.start()30    log_startup_message("Kinesis")31    port = port or config.service_port("kinesis")32    start_proxy_for_service(33        "kinesis",34        port,35        backend_port=_server.port,36        update_listener=update_listener,37    )38    return _server39def check_kinesis(expect_shutdown=False, print_error=False):40    out = None41    if not expect_shutdown:42        assert _server43    try:44        _server.wait_is_up()45        out = aws_stack.connect_to_service(46            service_name="kinesis", endpoint_url=_server.url47        ).list_streams()48    except Exception:49        if print_error:50            LOG.exception("Kinesis health check failed")51    if expect_shutdown:52        assert out is None53    else:54        assert out is not None and isinstance(out.get("StreamNames"), list)55def kinesis_running() -> bool:56    """57    Checks if there is a currently running Kinesis server instance.58    Currently used by localstack_ext/utils/cloud_pods.py59    :returns: True is there is a running Kinesis server instance, False otherwise60    TODO: rename this function to is_kinesis_running() for clarity61    """62    global _server63    if _server is None:64        return False65    return _server.is_running()66def restart_kinesis():67    global _server68    if _server:69        _server.shutdown()70        _server.join(timeout=10)71        _server = None72    LOG.debug("Restarting Kinesis process ...")...

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.