working prototype

Author: pinzon

.gitignore

@@ -0,0 +1,2 @@
+.venv
+__pycache__

Makefile

@@ -0,0 +1,21 @@
+VENV_DIR ?= .venv
+TEST_PATH ?= ./tests
+PIP_CMD ?= pip
+NOSE_LOG_LEVEL ?= WARNING
+
+ifeq ($(OS), Windows_NT)
+	VENV_RUN = . $(VENV_DIR)/Scripts/activate
+else
+	VENV_RUN = . $(VENV_DIR)/bin/activate
+endif
+
+setup-venv:
+	(test `which virtualenv` || $(PIP_CMD) install --user virtualenv) && \
+		(test -e $(VENV_DIR) || virtualenv $(VENV_OPTS) $(VENV_DIR))
+
+install-venv:
+	make setup-venv && \
+		test ! -e requirements.txt || ($(VENV_RUN); $(PIP_CMD) -q install -r requirements.txt)
+
+test:
+	($(VENV_RUN); DEBUG=$(DEBUG) PYTHONPATH=`pwd` nosetests $(NOSE_ARGS) --with-timer --with-coverage --logging-level=$(NOSE_LOG_LEVEL) --nocapture --no-skip --exe --cover-erase --cover-tests --cover-inclusive --cover-package=localstack --with-xunit --exclude='$(VENV_DIR).*' --ignore-files='lambda_python3.py' $(TEST_PATH))

common_utils.py

@@ -0,0 +1 @@
+

localstack.py

@@ -0,0 +1,95 @@
+import re
+import sys
+import docker
+import logging
+from localstack_docker.container import Container
+from localstack_docker.localstack_docker_configuration import LocalstackDockerConfiguration
+from localstack_docker.localstack_logger import LocalstackLogger
+
+ENV_CONFIG_USE_SSL = "USE_SSL"
+ENV_CONFIG_EDGE_PORT = "EDGE_PORT"
+INIT_SCRIPTS_PATH = "/docker-entrypoint-initaws.d"
+TMP_PATH = "/tmp/localstack"
+READY_TOKEN = re.compile("Ready\\.")
+DEFAULT_EDGE_PORT = 4566
+PORT_CONFIG_FILENAME = "/opt/code/localstack/.venv/lib/python3.8/site-packages/localstack_client/config.py"
+# DEFAULT_PORT_PATTERN = re.compile("'(\\w+)'\\Q: '{proto}://{host}:\\E(\\d+)'")
+
+localstack_instance = None
+LOG = logging.getLogger(__name__);
+
+
+class Localstack:
+    
+    localstack_container = None
+    service_to_map = {}
+    locked = False
+
+    @staticmethod
+    def INSTANCE():
+        return Localstack()
+
+    external_hostName = '' 
+
+    def startup(self, docker_configuration) :
+        if self.locked:
+            raise Exception("A docker instance is starting or already started.")
+        self.locked = True
+        self.external_hostname = docker_configuration.external_hostname
+
+        try:
+            self.localstack_container = Container.create_localstack_container(
+                docker_configuration.external_hostname,
+                docker_configuration.pull_new_image,
+                docker_configuration.randomize_ports,
+                docker_configuration.image_name,
+                docker_configuration.image_tag,
+                docker_configuration.port_edge,
+                docker_configuration.port_elastic_search,
+                docker_configuration.environment_variables,
+                docker_configuration.port_mappings,
+                docker_configuration.bind_mounts,
+                docker_configuration.platform
+            )
+
+            self.setup_logger()
+
+            Container.waitForReady(self.localstack_container, READY_TOKEN)
+
+        except docker.errors.APIError as error:
+            if not docker_configuration.ignore_docker_runerrors:
+                raise "Unable to start docker"
+
+        except:
+            error = sys.exc_info()
+            print(error)
+
+    def stop(self):
+        try:
+            self.localstack_container.stop()
+        except:
+            error = sys.exc_info()
+            print(error)
+    
+    def setup_logger(self):
+        localstack_logger = LocalstackLogger(self.localstack_container)
+        localstack_logger.start()
+
+    # @param services 
+def startup_localstack(port=4566, services=[], ignore_docker_errors = False):
+    global localstack_instance
+    localstack_instance = Localstack.INSTANCE()
+    config = LocalstackDockerConfiguration()
+
+    if len(services):
+        config.envirement_variables.update({'SERVICES': ','.join(services)})
+    if port != 4566:
+        config.port_edge = str(port)
+    
+    config.ignore_docker_runerrors = ignore_docker_errors
+
+    localstack_instance.startup(config)
+
+def stop_localstack():
+    if localstack_instance:
+        localstack_instance.stop()

localstack_docker/container.py

@@ -0,0 +1,57 @@
+import logging
+import re
+import docker
+from time import sleep
+
+LOCALSTACK_NAME = "localstack/localstack";
+LOCALSTACK_TAG = "latest";
+LOCALSTACK_PORT_EDGE = "4566";
+LOCALSTACK_PORT_ELASTICSEARCH = "4571";
+
+MAX_PORT_CONNECTION_ATTEMPTS = 10;
+MAX_LOG_COLLECTION_ATTEMPTS = 120;
+POLL_INTERVAL = 1;
+NUM_LOG_LINES = 10;
+
+ENV_DEBUG = "DEBUG";
+ENV_USE_SSL = "USE_SSL";
+ENV_DEBUG_DEFAULT = "1";
+LOCALSTACK_EXTERNAL_HOSTNAME = "HOSTNAME_EXTERNAL"
+DEFAULT_CONTAINER_ID = "localstack_main"
+
+DOCKER_CLIENT = docker.from_env()
+LOG = logging.getLogger(__name__);
+
+class Container:
+
+    @staticmethod
+    def create_localstack_container(external_hostname, pull_new_image, randomize_ports, image_name, image_tag, port_edge, port_elasticsearch, environment_variables, port_mappings, bind_mounts, platform):
+
+        environment_variables = {} if environment_variables == None else environment_variables
+        bind_mounts = {} if bind_mounts == None else bind_mounts
+        port_mappings = {} if port_mappings == None else port_mappings
+        image_name_or_default = LOCALSTACK_NAME if image_name == None else image_name 
+        image_exists = True if len(DOCKER_CLIENT.images.list(name=image_name_or_default)) else False
+        
+        fullPortEdge = {(LOCALSTACK_PORT_EDGE if port_edge == None else port_edge) : (LOCALSTACK_PORT_EDGE)}
+
+        if pull_new_image or not image_exists:
+            LOG.info("Pulling latest image")
+            DOCKER_CLIENT.images.pull(image_name_or_default, image_tag)
+        
+        return DOCKER_CLIENT.containers.run(image_name_or_default, ports=fullPortEdge, environment=environment_variables, detach=True)
+    
+    @staticmethod
+    def waitForReady(container, pattern):
+        attemps = 0
+        
+        while True:
+            logs = container.logs(tail=NUM_LOG_LINES).decode('utf-8')
+            if re.search(pattern, logs):
+                return;
+            
+            sleep(POLL_INTERVAL)
+            attemps += 1
+
+            if attemps >= MAX_LOG_COLLECTION_ATTEMPTS:
+                raise "Could not find token: " +pattern.toString()+ "in logs"

localstack_docker/localstack_docker_configuration.py

@@ -0,0 +1,17 @@
+class LocalstackDockerConfiguration:
+    pull_new_image = False
+    randomize_ports = False
+    image_name = None
+    image_tag = None
+    platform = None
+
+    port_edge = '4566'
+    port_elastic_search = '4571'
+    external_hostname = 'localhost'
+    environment_variables = {}
+    port_mappings = {}
+    bind_mounts = {}
+
+    initialization_token = None
+    use_dingle_docker_container = False
+    ignore_docker_runerrors = False

localstack_docker/localstack_logger.py

@@ -0,0 +1,22 @@
+import threading
+import logging
+
+LOG = logging.getLogger(__name__);
+
+class LocalstackLogger():
+    stream = None
+    def __init__(self, localsack_container) -> None:
+        stream = localsack_container.logs(stream=True, follow=True)    
+
+    
+    def start(self):
+        log_thread = threading.Thread(target=log_stream,args=[self.stream])
+            
+def log_stream(stream):
+    try:
+        print("starting logs")
+        while True:
+            line = next(stream).decode("utf-8")
+            LOG.warning(line)
+    except StopIteration:
+        print("LOGS ENDED")    

nosetests.xml

@@ -0,0 +1,2 @@
+<?xml version="1.0" encoding="UTF-8"?><testsuite name="nosetests" tests="1" errors="0" failures="0" skip="0"><testcase classname="kinesis_test.kinesis_test" name="test_create_stream" time="35.622"><system-out><![CDATA[(<class 'KeyboardInterrupt'>, KeyboardInterrupt(), <traceback object at 0x7fa229b1ae80>)
+]]></system-out></testcase></testsuite>

requirements.txt

@@ -0,0 +1,4 @@
+docker>=5.0.0
+nose>=1.3.7
+nose-timer>=0.7.5
+boto3>=1.17.70

tests/kinesis_test.py

@@ -0,0 +1,25 @@
+import time
+import boto3
+import unittest
+from localstack import startup_localstack, stop_localstack
+class kinesis_test(unittest.TestCase):
+        def setUp(self):
+            startup_localstack()
+            session = boto3.session.Session()
+
+        def tearDown(self):
+            stop_localstack()
+            return super().tearDown()
+
+        def test_create_stream(self):
+            kinesis = boto3.client(
+                service_name='kinesis',
+                aws_access_key_id='test',
+                aws_secret_access_key='test',
+                endpoint_url='http://localhost:4566')
+
+            kinesis.create_stream(StreamName='test', ShardCount=1)
+            time.sleep(10)
+
+            response = kinesis.list_streams()
+            self.assertGreater(len(response.get('StreamNames', [])),0)