Merge 4d701d6 into 219b6b3

.coveragerc

@@ -1,5 +1,4 @@
 [run]
-omit = ptero_petri/api/wsgi.py
 branch = True
 parallel = True
 source = ptero_petri

.gitignore

@@ -5,5 +5,5 @@
 /.tox
 /AUTHORS
 /ChangeLog
-/logs
 /build
+/var

.travis.yml

@@ -3,13 +3,15 @@ python: "2.7"
 
 install: pip install tox
 
-services:
-    - rabbitmq
-    - redis
+before_script: sudo update-alternatives --install /bin/sh sh /bin/bash 100
 
 script:
     - tox
 
 after_success:
     - pip install coveralls
     - coveralls
+
+after_script:
+    - ps -efl > var/log/ps.out
+    - bash -c 'for f in var/log/*; do echo; echo "============================================"; echo $f; echo "============================================"; cat $f; done'

Procfile

@@ -1,2 +1,2 @@
 web: gunicorn ptero_petri.api.wsgi:app
-orchestrator: ptero orchestrator
+orchestrator: petri-orchestrator

.env → env/development

@@ -7,3 +7,5 @@ PTERO_PETRI_AMQP_PREFETCH_COUNT=10
 PTERO_PETRI_AMQP_HEARTBEAT_INTERVAL=600
 PTERO_PETRI_REDIS_HOST=localhost
 PTERO_PETRI_REDIS_PORT=6379
+PTERO_PETRI_HOST=localhost
+PTERO_PETRI_PORT=6000

ptero_petri/implementation/backend.py

@@ -11,8 +11,9 @@
 
 
 class Backend(object):
-    def __init__(self, redis_connection):
+    def __init__(self, redis_connection, amqp_parameters):
         self.redis_connection = redis_connection
+        self.amqp_parameters = amqp_parameters
 
     def create_net(self, net_data):
         translator = Translator(net_data)
@@ -62,9 +63,17 @@ def cleanup(self):
 
 
     def _send_message(self, exchange, routing_key, body):
-        connection = pika.BlockingConnection()
+        connection = pika.BlockingConnection(self._pika_connection_params())
         channel = connection.channel()
         channel.confirm_delivery()
         channel.basic_publish(exchange=exchange, routing_key=routing_key,
                 body=body, properties=pika.BasicProperties(content_type='application/json',
                     delivery_mode=1))
+
+    def _pika_connection_params(self):
+        credentials = pika.PlainCredentials('guest', 'guest')
+        return pika.ConnectionParameters(
+                self.amqp_parameters.hostname,
+                self.amqp_parameters.port,
+                self.amqp_parameters.virtual_host,
+                credentials)

ptero_petri/implementation/factory.py

@@ -1,17 +1,24 @@
 from . import backend
+from . import interfaces
+from .brokers.amqp.connection_manager import ConnectionParams
+from .configuration.inject.initialize import initialize_injector
 import redis
+import os
 
 __all__ = ['Factory']
 
 
 class Factory(object):
     def __init__(self):
         self._initialized = False
+        self._injector = None
         self._redis = None
+        self._connection_parameters = None
 
     def create_backend(self):
         self._initialize()
-        return backend.Backend(redis_connection=self._redis)
+        return backend.Backend(redis_connection=self._redis,
+                amqp_parameters=self._connection_parameters)
 
     def purge(self):
         self._initialize()
@@ -21,4 +28,6 @@ def _initialize(self):
         # Lazy initialize to be pre-fork friendly.
         if not self._initialized:
             self._initialized = True
-            self._redis = redis.Redis()
+            self._injector = initialize_injector()
+            self._redis = self._injector.get(interfaces.IStorage)
+            self._connection_parameters = self._injector.get(ConnectionParams)

ptero_petri/implementation/petri/webhooks.py

@@ -33,8 +33,8 @@ def _request_body(response_links, data):
 
 
 def _url(net_key, place_idx, color, color_group_idx):
-    host = os.environ.get('PETRI_HOST', 'localhost')
-    port = int(os.environ.get('PETRI_PORT', '5000'))
+    host = os.environ.get('PTERO_PETRI_HOST', 'localhost')
+    port = int(os.environ.get('PTERO_PETRI_PORT', '5000'))
     return "http://%s:%d/v1/nets/%s/places/%d/tokens?color=%d&color_group=%d" % (
             host, port, net_key, place_idx, color, color_group_idx)
 

ptero_petri/implementation/util/exit.py

@@ -1,40 +1,11 @@
 import logging
 import os
-import psutil
-import signal
-import time
 
 
 LOG = logging.getLogger(__name__)
 
 
-_SIGNAL_TIMEOUT = 10
+def exit_process(exit_code):
+    LOG.info('Exitting process.')
 
-
-def exit_process(exit_code, child_signals=[signal.SIGINT, signal.SIGTERM]):
-    LOG.info('Exitting process: signalling children.')
-
-    for signum in child_signals:
-        _signal_child_processes(signum, timeout=_SIGNAL_TIMEOUT)
-
-    _signal_child_processes(signal.SIGKILL, recursive=True,
-            timeout=_SIGNAL_TIMEOUT)
-
-    LOG.info('Children killed, exiting with code %d', exit_code)
     os._exit(exit_code)
-
-
-def _signal_child_processes(signum, recursive=False, timeout=_SIGNAL_TIMEOUT):
-    for child in psutil.Process(os.getpid()).get_children(recursive=recursive):
-        child.send_signal(signum)
-
-    _wait_children(timeout, recursive=recursive)
-
-
-def _wait_children(timeout, recursive=False):
-    final_time = time.time() + timeout
-    for child in psutil.Process(os.getpid()).get_children(recursive=recursive):
-        try:
-            child.wait(max(0, final_time - time.time()))
-        except psutil.TimeoutExpired:
-            break

ptero_petri/implementation/util/signal_handlers.py

@@ -8,11 +8,11 @@
 
 
 def setup_standard_signal_handlers():
-    setup_exit_handler(signal.SIGTERM, [signal.SIGTERM, signal.SIGALRM])
+    setup_exit_handler(signal.SIGTERM)
 
 
-def setup_exit_handler(signum, child_signals):
+def setup_exit_handler(signum):
     def _handler(signum, frame):
         LOG.critical('Received signal %d: %s', signum, frame)
-        exit_process(exit_codes.UNKNOWN_ERROR, child_signals=child_signals)
+        exit_process(exit_codes.UNKNOWN_ERROR)
     signal.signal(signum, _handler)

test-requirements.txt

@@ -1,4 +1,5 @@
 fakeredis
+honcho
 mock
 nose >= 1.3.0
 nose-cov

tests/__init__.py

@@ -0,0 +1,90 @@
+import errno
+import os
+import psutil
+import signal
+import sys
+import time
+
+
+NUM_ORCHESTRATORS = 2
+
+
+instance = None
+
+
+def mkdir_p(path):
+    try:
+        os.makedirs(path)
+    except OSError as exc:
+        if exc.errno == errno.EEXIST and os.path.isdir(path):
+            pass
+        else:
+            raise
+
+
+def wait_time():
+    if os.environ.get('TRAVIS'):
+        return 15
+    else:
+        return 3
+
+def this_dir():
+    return os.path.dirname(__file__)
+
+def procfile_path():
+    return os.path.join(this_dir(), 'scripts', 'Procfile')
+
+def service_command_line():
+    return ['honcho', '-f', procfile_path(), 'start',
+            '-c', 'orchestrator=%d' % NUM_ORCHESTRATORS]
+
+
+def setUp():
+    global instance
+
+    logdir = 'var/log'
+    mkdir_p(logdir)
+    outlog = open(os.path.join(logdir, 'honcho.out'), 'w')
+    errlog = open(os.path.join(logdir, 'honcho.err'), 'w')
+
+    if not os.environ.get('SKIP_PROCFILE'):
+        instance = psutil.Popen(service_command_line(),
+                shell=False, stdout=outlog, stderr=errlog)
+        time.sleep(wait_time())
+        if instance.poll() is not None:
+            raise RuntimeError("honcho instance terminated prematurely")
+
+def signal_processes(processes, sig):
+    signaled_someone = False
+    for p in processes:
+        try:
+            p.send_signal(sig)
+            signaled_someone = True
+        except psutil.NoSuchProcess:
+            pass
+
+    return signaled_someone
+
+def get_descendents():
+    return psutil.Process(instance.pid).get_children(recursive=True)
+
+def cleanup():
+    descendents = get_descendents()
+
+    instance.send_signal(signal.SIGINT)
+    try:
+        instance.wait(timeout=10)
+    except psutil.TimeoutExpired:
+        pass
+
+    if not signal_processes(descendents, signal.SIGINT):
+        return
+
+    time.sleep(3)
+    signal_processes(descendents, signal.SIGKILL)
+
+
+# NOTE If this doesn't run then honcho will be orphaned...
+def tearDown():
+    if not os.environ.get('SKIP_PROCFILE'):
+        cleanup()

tests/api/v1/generator/base_case.py

@@ -1,3 +1,4 @@
+from ptero_petri.implementation.petri.webhooks import _retry as retry
 import abc
 import collections
 import errno
@@ -27,9 +28,9 @@ def validate_json(text):
 class TestCaseMixin(object):
     __metaclass__ = abc.ABCMeta
 
-    @abc.abstractproperty
+    @property
     def api_port(self):
-        pass
+        return int(os.environ['PTERO_PETRI_PORT'])
 
     @abc.abstractproperty
     def callback_port(self):
@@ -58,13 +59,11 @@ def setUp(self):
         super(TestCaseMixin, self).setUp()
         self._clear_memoized_data()
 
-        self._start_devserver()
         self._start_callback_receipt_webserver()
 
     def tearDown(self):
         super(TestCaseMixin, self).tearDown()
         self._stop_callback_receipt_webserver()
-        self._stop_devserver()
 
 
     def _submit_net(self):
@@ -159,7 +158,7 @@ def _assemble_callback_url(self, callback_name, request_data):
         return urlparse.urlunparse((
             'http',
             'localhost:%d' % self.callback_port,
-            '/' + callback_name,
+            '/callbacks/' + callback_name,
             '',
             urllib.urlencode(request_data),
             '',
@@ -191,23 +190,6 @@ def _clear_memoized_data(self):
         self._expected_callbacks = None
 
 
-    def _start_devserver(self):
-        cmd = [
-                self._devserver_path,
-                '--max-run-time', str(self._max_wait_time),
-                '--port', str(self.api_port),
-                '--logdir', str(self._logdir),
-                '--cover',
-        ]
-        if int(os.environ.get('PTERO_TEST_WEBSERVER_DEBUG', 0)) == 1:
-            cmd.append('--debug')
-
-        self._devserver = subprocess.Popen(cmd, close_fds=True)
-        self._wait_for_devserver()
-
-    def _wait_for_devserver(self):
-        time.sleep(5)
-
     def _start_callback_receipt_webserver(self):
         self._callback_webserver = subprocess.Popen(
                 [self._callback_webserver_path,
@@ -216,21 +198,24 @@ def _start_callback_receipt_webserver(self):
                     '--port', str(self.callback_port),
                     ],
                 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        self._wait_for_callback_webserver()
+
+    def _wait_for_callback_webserver(self):
+        response = retry(requests.get, self._callback_ping_url())
+        if response.status_code != 200:
+            raise RuntimeError('Failed to spin up callback webserver: %s'
+                    % response.text)
+
+    def _callback_ping_url(self):
+        return 'http://localhost:%d/ping' % self.callback_port
 
     def _stop_callback_receipt_webserver(self):
         _stop_subprocess(self._callback_webserver)
 
-    def _stop_devserver(self):
-        _stop_subprocess(self._devserver)
-
     @property
     def _callback_webserver_path(self):
         return os.path.join(os.path.dirname(__file__), 'callback_webserver.py')
 
-    @property
-    def _devserver_path(self):
-        return os.path.join(self._repository_root_path, 'devserver')
-
     @property
     def _logdir(self):
         return os.path.join(self._repository_root_path, 'logs', self.test_name)

tests/api/v1/generator/callback_webserver.py

@@ -26,7 +26,12 @@ def parse_arguments():
 app = Flask(__name__)
 
 
-@app.route('/<path:callback_name>', methods=['PUT'])
+@app.route('/ping', methods=['GET'])
+def ping():
+    return 'PONG'
+
+
+@app.route('/callbacks/<path:callback_name>', methods=['PUT'])
 def log_request(callback_name):
     try:
         print callback_name

tests/api/v1/generator/func.py

@@ -25,7 +25,6 @@ def _create_and_attach_test_case(target_module, test_case_directory,
 
 def _create_test_case(test_case_directory, test_case_name):
     class_dict = {
-        'api_port': _get_available_port(),
         'callback_port': _get_available_port(),
         'directory': os.path.join(test_case_directory, test_case_name),
         'test_name': test_case_name,