Merge remote-tracking branch 'origin/SSA-6320-workflow-service' into SSA-6320-workflow-service

apps/cli/executables/alma_reingester/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['cx_Oracle', 'pycapo', 'psycopg2', 'events', 'schema'],
+    install_requires=['cx_Oracle', 'pycapo', 'psycopg2', 'ssa-events', 'ssa-schema'],
     keywords=[],
     packages=['alma_reingester'],
     package_dir={'':'src'},

apps/cli/executables/ingestion/setup.py

@@ -22,11 +22,11 @@ setup(
                       'numpy',
                       'psycopg2',
                       'pycapo',
-                      'pymygdala',
                       'requests',
-                      'schema',
-                      'sqlalchemy',
-                      'weblog_thumbs'],
+                      'ssa-pymygdala',
+                      'ssa-schema',
+                      'ssa-weblog_thumbs',
+                      'sqlalchemy'],
     keywords=[],
     packages=['ingestion'],
     package_dir={'':'src'},

apps/cli/executables/null/setup.py

@@ -7,9 +7,6 @@ from setuptools import setup
 VERSION = open('src/null/_version.py').readlines()[-1].split()[-1].strip("\"'")
 README = Path('README.md').read_text()
 
-# requires = [
-# ]
-
 tests_require = [
     'pytest>=5.4,<6.0'
 ]

apps/cli/executables/vlba_grabber/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['requests', 'pymygdala', 'pycapo', 'beautifulsoup4'],
+    install_requires=['requests', 'ssa-pymygdala', 'pycapo', 'beautifulsoup4'],
     keywords=[],
     packages=['vlba_grabber'],
     package_dir={'':'src'},

apps/cli/executables/weblog_thumbs/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['pymygdala'],
+    install_requires=['ssa-pymygdala'],
     keywords=[],
     packages=['weblog_thumbs'],
     package_dir={'':'src'},

apps/cli/launchers/wf/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['pika', 'psycopg2', 'cx_Oracle', 'pymygdala', 'pycapo', 'schema'],
+    install_requires=['pika', 'psycopg2', 'cx_Oracle', 'ssa-pymygdala', 'pycapo', 'ssa-schema'],
     keywords=[],
     packages=['wf'],
     package_dir={'':'src'},

apps/cli/utilities/datafinder/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['pandas', 'requests', 'schema', 'sqlalchemy', 'tqdm'],
+    install_requires=['pandas', 'requests', 'ssa-schema', 'sqlalchemy', 'tqdm'],
     keywords=[],
     packages=['datafinder'],
     package_dir={'':'src'},

apps/cli/utilities/mr_books/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['pendulum', 'pid', 'pycapo', 'pymygdala', 'schema', 'sqlalchemy'],
+    install_requires=['pendulum', 'pid', 'pycapo', 'ssa-pymygdala', 'ssa-schema', 'sqlalchemy'],
     keywords=[],
     packages=['mr_books'],
     package_dir={'':'src'},

apps/cli/utilities/mr_clean/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['pendulum', 'pycapo', 'pymygdala'],
+    install_requires=['pendulum', 'pycapo', 'ssa-pymygdala'],
     keywords=[],
     packages=['mr_clean'],
     package_dir={'':'src'},

apps/cli/utilities/proprietary_setter/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['pymygdala', 'schema', 'support'],
+    install_requires=['ssa-pymygdala', 'ssa-schema', 'ssa-support'],
     keywords=[],
     packages=['proprietary_setter'],
     package_dir={'':'src'},

apps/cli/utilities/qa_results/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['pycapo', 'pymygdala'],
+    install_requires=['pycapo', 'ssa-pymygdala'],
     keywords=[],
     packages=['qa_results'],
     package_dir={'':'src'},

apps/cli/utilities/s_code_project_updater/setup.py

@@ -16,7 +16,7 @@ setup(
     author_email='dms-ssa@nrao.edu',
     url='TBD',
     license="GPL",
-    install_requires=['pycapo', 'pymygdala', 'schema', 'sqlalchemy'],
+    install_requires=['pycapo', 'ssa-pymygdala', 'ssa-schema', 'sqlalchemy'],
     tests_require=['pytest-mock'],
     requires=['sqlalchemy', 'mysqldb'],
     keywords=[],

apps/cli/utilities/wf_monitor/test/test_wf_monitor.py

@@ -11,6 +11,10 @@ WORKFLOW_STATUS_EXCH = 'workspaces.workflow-service.workflow-status'
 
 
 def test_read_log():
+    """
+    Tests whether or not the example log is being correctly read by checking for strings known to
+    exist in the example log file
+    """
     test_strs = [
         'Image size of job updated: 432',
         '432  -  ResidentSetSize of job (KB)',
@@ -22,7 +26,10 @@ def test_read_log():
         assert s in test_monitor.log
 
 
-def test_read_log_timeout(capsys):
+def test_read_log_timeout():
+    """
+    Tests the timeout functionality of WorkflowMonitor.read_htcondor_log()
+    """
     with pytest.raises(SystemExit) as sys_ex:
         WorkflowMonitor('logs/file-that-does-not-exist.txt', 1)
     assert sys_ex.type == SystemExit
@@ -30,6 +37,10 @@ def test_read_log_timeout(capsys):
 
 
 def test_parse_log():
+    """
+    Tests that the example log file is correctly parsed by checking for the known order of event
+    types
+    """
     test_event_types = [
         WorkflowEventType.SUBMITTED,
         WorkflowEventType.EXECUTING,
@@ -45,12 +56,23 @@ def test_parse_log():
 
 
 def test_parse_log_error():
+    """
+    Tests that WorkflowMonitor.parse_log() correctly raises a ValueError when a badly formatted
+    log is parsed
+    """
     with pytest.raises(ValueError) as val_err:
         WorkflowMonitor('logs/test.log')
     assert val_err.type == ValueError
 
 
 def test_monitor_events(mocker: MockerFixture):
+    """
+    Test that mocks the core wf_monitor functionality of creating a connection to the RabbitMQ
+    server, configuring a channel with that connection, and emitting events through that channel
+    :param mocker: Pytest fixture from pytest_mock that wraps the real function and mimics that
+    its been called as normal; allows for testing interaction with the server without actually
+    requiring a connection to the server
+    """
     mock_list = [
         'pika.adapters.blocking_connection.BlockingChannel.basic_publish',
         'channels.amqp_helpers.make_amqp_connection',

services/archive/setup.py

@@ -43,7 +43,7 @@ requires = [
     'pyramid_debugtoolbar',
     'pyramid_tm',
     'requests',
-    'schema',
+    'ssa-schema',
     'sqlalchemy',
     'waitress',
     'zope.sqlalchemy'

services/workflow/setup.py

@@ -43,10 +43,10 @@ requires = [
     'pyramid_debugtoolbar',
     'pyramid_tm',
     'requests',
-    'schema',
+    'ssa-schema',
     'sqlalchemy',
     'waitress',
-    'workspaces',
+    'ssa-workspaces',
     'zope.sqlalchemy'
 ]
 

shared/channels/src/channels/amqp_helpers.py

-from typing import Callable, Any
+from typing import Callable, Any, Optional, Union
 
 import pika
 from marshmallow import Schema
@@ -6,115 +6,98 @@ from pycapo import CapoConfig
 from workspaces.json import WorkflowEventSchema
 
 
-def make_amqp_connection(profile: str) -> pika.BlockingConnection:
-    """
-    Initialize connection to AMQP server given a CAPO profile
-    :param profile: Capo profile to use
-    :return: Established connection
-    """
-    capo_cfg = CapoConfig(profile=profile)
-
-    hostname = capo_cfg.getstring('edu.nrao.archive.configuration.AmqpServer.hostname')
-    username = capo_cfg.getstring('edu.nrao.archive.configuration.AmqpServer.username')
-    password = capo_cfg.getstring('edu.nrao.archive.configuration.AmqpServer.password')
-
-    return pika.BlockingConnection(
-        pika.ConnectionParameters(
-            host=hostname,
-            credentials=pika.PlainCredentials(
-                username=username,
-                password=password
-            )
-        )
-    )
-
-
-def configure_amqp_channel(
-        connection: pika.BlockingConnection,
-        exchange: str,
-        exchange_type: str
-) -> pika.adapters.blocking_connection.BlockingChannel:
-    """
-    Takes in a connection and exchange declaration parameters and creates a channel to the given
-    exchange, configured with the given parameters
-    :param connection: RabbitMQ BlockingConnection
-    :param exchange: Exchange name
-    :param exchange_type: Exchange type
-    :return: Channel for given connection
-    """
-    channel = connection.channel()
-    channel.exchange_declare(exchange=exchange, exchange_type=exchange_type, durable=True)
-    return channel
-
-
-def emit_amqp_message(
-        channel: pika.adapters.blocking_connection.BlockingChannel,
-        exchange: str,
-        routing_key: str,
-        msg: bytes
-):
-    """
-    Emit AMQP message to a given exchange through a given channel
-    :param channel: AMQP connection channel
-    :param exchange: AMQP exchange name
-    :param routing_key: Routing key
-    :param msg: Message to be emitted
-    """
-    channel.basic_publish(exchange=exchange, routing_key=routing_key, body=msg)
-
-
-def receive_amqp_message(
-        channel: pika.adapters.blocking_connection.BlockingChannel,
-        exchange: str,
-        routing_key: str,
-        callback: Callable,
-        auto_ack: bool = False
-):
-    """
-    Establishes queue and binds it to a given channel and consumes messages matching the
-    routing key from given exchange
-    :param channel: AMQP connection channel
-    :param exchange: AMQP exchange name
-    :param routing_key: Routing key
-    :param callback: Callback function for when a message is consumed
-    :param auto_ack: If true, consumer automatically acknowledges when it consumes a message
-    """
-    queue_result = channel.queue_declare(queue='', exclusive=True)
-    queue_name = queue_result.method.queue
-    channel.queue_bind(queue=queue_name, exchange=exchange, routing_key=routing_key)
-    channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=auto_ack)
-    channel.start_consuming()
+CONNECTION = None
+CHANNEL = None
+CONFIG = None
 
 
 class Channel:
     def __init__(self, schema: Schema):
         self.schema = schema
 
-    def connect(self):
-        # FIXME implement this
-        pass
+    def connect(self, **kwargs: Union[int, str]) -> pika.BlockingConnection:
+        """
+        Initialize connection to AMQP server given a CAPO profile
+        :return: Established connection
+        """
+        global CONNECTION
+        global CHANNEL
+        global CONFIG
+
+        if not CONNECTION:
+            CONFIG = CapoConfig(
+                profile=kwargs.get('profile', None)
+            ).settings('edu.nrao.archive.configuration.AmqpServer')
+
+            connection_parameters = pika.ConnectionParameters(
+                host=kwargs.get('hostname', CONFIG.hostname),
+                port=int(kwargs.get('port', CONFIG.port)),
+                # FIXME: Copied from events. Do these args need to be cast to int?
+                connection_attempts=kwargs.get('connection_attempts', 5),
+                socket_timeout=kwargs.get('socket_timeout', 5000),
+                retry_delay=kwargs.get('retry_delay', 500),
+                #
+                credentials=pika.PlainCredentials(
+                    username=kwargs.get('username', CONFIG.username),
+                    password=kwargs.get('password', CONFIG.password)
+                )
+            )
+            CONNECTION = pika.BlockingConnection(connection_parameters)
+            CHANNEL = CONNECTION.channel()
+            CHANNEL.exchange_declare(
+                # FIXME: CAPO config does not currently support 'exchange' value
+                exchange=kwargs.get('exchange', CONFIG.exchange),
+                exchange_type='topic',
+                durable=True,
+                auto_delete=False
+            )
 
     def close(self):
-        # FIXME implement this
-        pass
+        if CONNECTION:
+            CONNECTION.close()
 
-    def send(self, event):
-        # I don't know how to type Event there for Python but it would be
-        # nice if we could, somehow.
+    def send(self, event: WorkflowEventSchema):
         rendered = self.schema.dump(event)
-        # FIXME: transmit rendered over the underlying channel
-        pass
+        routing_key = f'{rendered["job_name"]}.{rendered["job_id"]}.{rendered["event_type"]}'
+        CHANNEL.basic_publish(routing_key=routing_key, body=str(rendered))
+
+    def listen(self, callback: Optional[Callable], pattern: str = '#', **kwargs: Union[str, bool]):
+        """
+        Establishes queue and binds it to a given channel and consumes messages matching the
+        routing key from given exchange
+        :param callback: Callback function for when a message is consumed
+        :param pattern: Pattern to be used as routing key
+        Optional keyword arguments
+        :param exchange: AMQP exchange name
+        :param auto_ack: If true, consumer automatically acknowledges when it consumes a message
+        """
+        exchange = kwargs.get('exchange', CONFIG.exchange)
+        auto_ack = kwargs.get('auto_ack', False)
 
-    def listen(self, callback: Callable[[Any], None], pattern: str = '#'):
         def unwrapping_callback(message):
             event = self.schema.load(message)
             callback(event)
         # FIXME: ask the underlying channel to call us back
         # but only for messages matching pattern
-        pass
 
-    def __enter__(self):
-        self.connect()
+        queue = CHANNEL.queue_declare(queue='', exclusive=True).method.queue
+        CHANNEL.queue_bind(queue=queue, exchange=exchange, routing_key=pattern)
+        CHANNEL.basic_consume(queue=queue, on_message_callback=callback, auto_ack=auto_ack)
+        CHANNEL.start_consuming()
+
+    def __enter__(self, **kwargs: Union[int, str]):
+        """
+        Keyword arguments for the AMQP connection. These do not need to be specified:
+        :param: hostname: Hostname to connect to
+        :param: port: Port to connect to
+        :param: connection_attempts: Number of connection attempts to try
+        :param: socket_timeout: Time to wait for a socket to connect
+        :param: retry_delay: Time to wait between retrying the connection
+        :param: username: Username to connect to as
+        :param: password: Password to use when connecting
+        :param: exchange: Exchange to use when connection
+        """
+        self.connect(**kwargs)
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.close()

shared/workspaces/src/workspaces/json.py

@@ -15,11 +15,11 @@ class WorkflowEventSchema(Schema):
     retval = fields.Integer(allow_none=True)
 
     # Enums apparently are not a first-class field type in Marshmallow
-    def get_type(self, obj: WorkflowEventType) -> int:
-        return obj.type.value
+    def get_type(self, obj: WorkflowEventType) -> str:
+        return obj.type.name.lower()
 
-    def load_type(self, value: int) -> WorkflowEventType:
-        return next(et for et in WorkflowEventType if et.value == value)
+    def load_type(self, name: str) -> WorkflowEventType:
+        return next(et for et in WorkflowEventType if et.name.lower() == name)
 
     @post_load
     def make_event(self, data, **kwargs):