Initial testing framework

Change-Id: I6dffda68221faae3960230e5e3f9856fc0caba47

requirements.txt

@@ -0,0 +1,3 @@
+falcon==1.2.0
+python-dateutil==2.6.1
+requests==2.18.3

setup.py

@@ -24,10 +24,9 @@ setup(name='shipyard_airflow',
       packages=['shipyard_airflow',
                 'shipyard_airflow.control'],
       install_requires=[
-        'falcon',
+          'falcon',
-        'requests',
+          'requests',
-        'configparser',
+          'configparser',
-        'uwsgi>1.4'
+          'uwsgi>1.4',
-      ]
+          'python-dateutil'
-     )
+      ])
-

shipyard_airflow/__init__.py

@@ -10,4 +10,4 @@
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
-# limitations under the License.
+# limitations under the License.

shipyard_airflow/airflow_client.py

@@ -0,0 +1,17 @@
+import requests
+
+from shipyard_airflow.errors import AirflowError
+
+
+class AirflowClient(object):
+    def __init__(self, url):
+        self.url = url
+
+    def get(self):
+        response = requests.get(self.url).json()
+
+        # This gives us more freedom to handle the responses from airflow
+        if response["output"]["stderr"]:
+            raise AirflowError(response["output"]["stderr"])
+        else:
+            return response["output"]["stdout"]

shipyard_airflow/control/airflow_connections.py

@@ -11,12 +11,11 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-import falcon
+from urllib.parse import urlunsplit
-import json
+from falcon import HTTPInvalidParam
-import requests
-import urllib.parse
 
 from .base import BaseResource
+from shipyard_airflow.airflow_client import AirflowClient
 
 # We need to be able to add/delete connections so that we can create/delete
 # connection endpoints that Airflow needs to connect to
@@ -25,35 +24,21 @@ class AirflowAddConnectionResource(BaseResource):
     authorized_roles = ['user']
 
     def on_get(self, req, resp, action, conn_id, protocol, host, port):
-        # Retrieve URL
         web_server_url = self.retrieve_config('base', 'web_server')
+        if action != 'add':
+            raise HTTPInvalidParam(
+                'Invalid Paremeters for Adding Airflow Connection', 'action')
 
-        if 'Error' in web_server_url:
+        # Concatenate to form the connection URL
-            resp.status = falcon.HTTP_500
+        netloc = ''.join([host, ':', port])
-            raise falcon.HTTPInternalServerError("Internal Server Error", "Missing Configuration File")
+        url = (protocol, netloc, '', '', '')
-        else:
+        conn_uri = urlunsplit(url)
-            if action == 'add':
+        # Form the request URL towards Airflow
-                # Concatenate to form the connection URL
+        req_url = ('{}/admin/rest_api/api?api=connections&add=true&conn_id'
-                netloc = ''.join([host, ':', port])
+                   '={}&conn_uri={}'.format(web_server_url, conn_id, conn_uri))
-                url = (protocol, netloc, '','','')
-                conn_uri = urlparse.urlunsplit(url)
 
-                # Form the request URL towards Airflow
+        airflow_client = AirflowClient(req_url)
-                req_url = '{}/admin/rest_api/api?api=connections&add=true&conn_id={}&conn_uri={}'.format(web_server_url, conn_id, conn_uri)
+        self.on_success(resp, airflow_client.get())
-            else:
-                self.return_error(resp, falcon.HTTP_400, 'Invalid Paremeters for Adding Airflow Connection')
-                return
-
-            response = requests.get(req_url).json()
-
-            # Return output
-            if response["output"]["stderr"]:
-                resp.status = falcon.HTTP_400
-                resp.body = response["output"]["stderr"]
-                return
-            else:
-                resp.status = falcon.HTTP_200
-                resp.body = response["output"]["stdout"]
 
 
 # Delete a particular connection endpoint
@@ -64,28 +49,15 @@ class AirflowDeleteConnectionResource(BaseResource):
     def on_get(self, req, resp, action, conn_id):
         # Retrieve URL
         web_server_url = self.retrieve_config('base', 'web_server')
+        if action != 'delete':
+            raise HTTPInvalidParam(
+                'Invalid Paremeters for Deleting Airflow Connection', 'action')
 
-        if 'Error' in web_server_url:
+        # Form the request URL towards Airflow
-            resp.status = falcon.HTTP_500
+        req_url = ('{}/admin/rest_api/api?api=connections&delete=true&conn_id'
-            raise falcon.HTTPInternalServerError("Internal Server Error", "Missing Configuration File")
+                   '={}'.format(web_server_url, conn_id))
-        else:
+        airflow_client = AirflowClient(req_url)
-            if action == 'delete':
+        self.on_success(resp, airflow_client.get())
-                # Form the request URL towards Airflow
-                req_url = '{}/admin/rest_api/api?api=connections&delete=true&conn_id={}'.format(web_server_url, conn_id)
-            else:
-                self.return_error(resp, falcon.HTTP_400, 'Invalid Paremeters for Deleting Airflow Connection')
-                return
-
-            response = requests.get(req_url).json()
-
-            # Return output
-            if response["output"]["stderr"]:
-                resp.status = falcon.HTTP_400
-                resp.body = response["output"]["stderr"]
-                return
-            else:
-                resp.status = falcon.HTTP_200
-                resp.body = response["output"]["stdout"]
 
 
 # List all current connection endpoints
@@ -94,28 +66,13 @@ class AirflowListConnectionsResource(BaseResource):
     authorized_roles = ['user']
 
     def on_get(self, req, resp, action):
-        # Retrieve URL
         web_server_url = self.retrieve_config('base', 'web_server')
+        if action != 'list':
+            raise HTTPInvalidParam(
+                'Invalid Paremeters for listing Airflow Connections', 'action')
 
-        if 'Error' in web_server_url:
+        req_url = '{}/admin/rest_api/api?api=connections&list=true'.format(
-            resp.status = falcon.HTTP_500
+            web_server_url)
-            raise falcon.HTTPInternalServerError("Internal Server Error", "Missing Configuration File")
-        else:
-            if action == 'list':
-                # Form the request URL towards Airflow
-                req_url = '{}/admin/rest_api/api?api=connections&list=true'.format(web_server_url)
-            else:
-                self.return_error(resp, falcon.HTTP_400, 'Invalid Paremeters for listing Airflow Connections')
-                return
-
-            response = requests.get(req_url).json()
-
-            # Return output
-            if response["output"]["stderr"]:
-                resp.status = falcon.HTTP_400
-                resp.body = response["output"]["stderr"]
-                return
-            else:
-                resp.status = falcon.HTTP_200
-                resp.body = response["output"]["stdout"]
 
+        airflow_client = AirflowClient(req_url)
+        self.on_success(resp, airflow_client.get())

shipyard_airflow/control/api.py

@@ -29,11 +29,17 @@ from .airflow_connections import AirflowDeleteConnectionResource
 from .airflow_connections import AirflowListConnectionsResource
 from .airflow_get_version import GetAirflowVersionResource
 from .middleware import AuthMiddleware, ContextMiddleware, LoggingMiddleware
+from shipyard_airflow.errors import AppError
+
 
 def start_api():
-
+    middlewares = [
-    control_api = falcon.API(request_type=ShipyardRequest,
+        AuthMiddleware(),
-                             middleware=[AuthMiddleware(), ContextMiddleware(), LoggingMiddleware()])
+        ContextMiddleware(),
+        LoggingMiddleware(),
+    ]
+    control_api = falcon.API(
+        request_type=ShipyardRequest, middleware=middlewares)
 
     control_api.add_route('/versions', VersionsResource())
 
@@ -45,13 +51,19 @@ def start_api():
         ('/dags/{dag_id}/tasks/{task_id}', TaskResource()),
         ('/dags/{dag_id}/dag_runs', DagRunResource()),
         ('/list_dags', ListDagsResource()),
-        ('/task_state/dags/{dag_id}/tasks/{task_id}/execution_date/{execution_date}', GetTaskStatusResource()),
+        ('/task_state/dags/{dag_id}/tasks/{task_id}/execution_date/'
-        ('/dag_state/dags/{dag_id}/execution_date/{execution_date}', GetDagStateResource()),
+         '{execution_date}', GetTaskStatusResource()),
+        ('/dag_state/dags/{dag_id}/execution_date/{execution_date}',
+            GetDagStateResource()),
         ('/list_tasks/dags/{dag_id}', ListTasksResource()),
-        ('/trigger_dag/dags/{dag_id}/run_id/{run_id}', TriggerDagRunResource()),
+        ('/trigger_dag/dags/{dag_id}/run_id/{run_id}',
-        ('/trigger_dag/dags/{dag_id}/run_id/{run_id}/poll', TriggerDagRunPollResource()),
+            TriggerDagRunResource()),
-        ('/connections/{action}/conn_id/{conn_id}/protocol/{protocol}/host/{host}/port/{port}', AirflowAddConnectionResource()),
+        ('/trigger_dag/dags/{dag_id}/run_id/{run_id}/poll',
-        ('/connections/{action}/conn_id/{conn_id}', AirflowDeleteConnectionResource()),
+            TriggerDagRunPollResource()),
+        ('/connections/{action}/conn_id/{conn_id}/protocol/{protocol}'
+         '/host/{host}/port/{port}', AirflowAddConnectionResource()),
+        ('/connections/{action}/conn_id/{conn_id}',
+            AirflowDeleteConnectionResource()),
         ('/connections/{action}', AirflowListConnectionsResource()),
         ('/airflow/version', GetAirflowVersionResource()),
     ]
@@ -59,6 +71,7 @@ def start_api():
     for path, res in v1_0_routes:
         control_api.add_route('/api/v1.0' + path, res)
 
+    control_api.add_error_handler(AppError, AppError.handle)
     return control_api
 
 class VersionsResource(BaseResource):
@@ -66,9 +79,9 @@ class VersionsResource(BaseResource):
     authorized_roles = ['anyone']
 
     def on_get(self, req, resp):
-        resp.body = json.dumps({'v1.0': {
+        resp.body = json.dumps({
-                                    'path': '/api/v1.0',
+            'v1.0': {
-                                    'status': 'stable'
+                'path': '/api/v1.0',
-                                }})
+                'status': 'stable'
+            }})
         resp.status = falcon.HTTP_200
-

shipyard_airflow/control/base.py

@@ -11,11 +11,22 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-import falcon, falcon.request as request
+import falcon
+import falcon.request as request
 import uuid
 import json
 import configparser
 import os
+try:
+    from collections import OrderedDict
+except ImportError:
+    OrderedDict = dict
+
+from shipyard_airflow.errors import (
+    AppError,
+    ERR_UNKNOWN,
+)
+
 
 class BaseResource(object):
 
@@ -43,17 +54,30 @@ class BaseResource(object):
         else:
             return True
 
+    def to_json(self, body_dict):
+        return json.dumps(body_dict)
+
+    def on_success(self, res, message=None):
+        res.status = falcon.HTTP_200
+        response_dict = OrderedDict()
+        response_dict['type'] = 'success'
+        response_dict['message'] = message
+        res.body = self.to_json(response_dict)
+
     # Error Handling
     def return_error(self, resp, status_code, message="", retry=False):
         """
-        Write a error message body and throw a Falcon exception to trigger an HTTP status
+        Write a error message body and throw a Falcon exception to trigger
+        an HTTP status
 
         :param resp: Falcon response object to update
         :param status_code: Falcon status_code constant
         :param message: Optional error message to include in the body
-        :param retry: Optional flag whether client should retry the operation. Can ignore if we rely solely on 4XX vs 5xx status codes
+        :param retry: Optional flag whether client should retry the operation.
+        Can ignore if we rely solely on 4XX vs 5xx status codes
         """
-        resp.body = json.dumps({'type': 'error', 'message': message, 'retry': retry})
+        resp.body = self.to_json(
+            {'type': 'error', 'message': message, 'retry': retry})
         resp.content_type = 'application/json'
         resp.status = status_code
 
@@ -62,7 +86,7 @@ class BaseResource(object):
 
         # Shipyard config will be located at /etc/shipyard/shipyard.conf
         path = '/etc/shipyard/shipyard.conf'
-        
+
         # Check that shipyard.conf exists
         if os.path.isfile(path):
             config = configparser.ConfigParser()
@@ -73,7 +97,7 @@ class BaseResource(object):
 
             return query_data
         else:
-            return 'Error - Missing Configuration File'
+            raise AppError(ERR_UNKNOWN, "Missing Configuration File")
 
 
 class ShipyardRequestContext(object):
@@ -107,4 +131,3 @@ class ShipyardRequestContext(object):
 
 class ShipyardRequest(request.Request):
     context_type = ShipyardRequestContext
-

shipyard_airflow/errors.py

@@ -0,0 +1,52 @@
+# -*- coding: utf-8 -*-
+
+import json
+import falcon
+
+try:
+    from collections import OrderedDict
+except ImportError:
+    OrderedDict = dict
+
+ERR_UNKNOWN = {
+    'status': falcon.HTTP_500,
+    'title': 'Internal Server Error'
+}
+
+ERR_AIRFLOW_RESPONSE = {
+    'status': falcon.HTTP_400,
+    'title': 'Error response from Airflow'
+}
+
+
+class AppError(Exception):
+    def __init__(self, error=ERR_UNKNOWN, description=None):
+        self.error = error
+        self.error['description'] = description
+
+    @property
+    def title(self):
+        return self.error['title']
+
+    @property
+    def status(self):
+        return self.error['status']
+
+    @property
+    def description(self):
+        return self.error['description']
+
+    @staticmethod
+    def handle(exception, req, res, error=None):
+        res.status = exception.status
+        meta = OrderedDict()
+        meta['message'] = exception.title
+        if exception.description:
+            meta['description'] = exception.description
+        res.body = json.dumps(meta)
+
+
+class AirflowError(AppError):
+    def __init__(self, description=None):
+        super().__init__(ERR_AIRFLOW_RESPONSE)
+        self.error['description'] = description

shipyard_airflow/setup.py

@@ -24,10 +24,9 @@ setup(name='shipyard_airflow',
       packages=['shipyard_airflow',
                 'shipyard_airflow.control'],
       install_requires=[
-        'falcon',
+          'falcon',
-        'requests',
+          'requests',
-        'configparser',
+          'configparser',
-        'uwsgi>1.4'
+          'uwsgi>1.4',
-      ]
+          'python-dateutil'
-     )
+      ])
-

test-requirements.txt

@@ -0,0 +1,6 @@
+# Testing
+pytest==3.2.1
+mock==2.0.0
+
+# Linting
+flake8==3.3.0

tests/unit/control/test_airflow_connections.py

@@ -0,0 +1,267 @@
+# Copyright 2017 AT&T Intellectual Property.  All other rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an 'AS IS' BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import falcon
+from falcon import testing
+import mock
+
+from shipyard_airflow.control import api
+from shipyard_airflow.control.airflow_connections import (
+    AirflowAddConnectionResource,
+    AirflowDeleteConnectionResource,
+    AirflowListConnectionsResource,
+)
+
+
+class BaseTesting(testing.TestCase):
+    def setUp(self):
+        super().setUp()
+        self.app = api.start_api()
+        self.conn_id = 1
+        self.protocol = 'http'
+        self.host = '10.0.0.1'
+        self.port = '3000'
+
+    @property
+    def _headers(self):
+        return {
+            'X-Auth-Token': '10'
+        }
+
+
+class AirflowAddConnectionResourceTestCase(BaseTesting):
+    def setUp(self):
+        super().setUp()
+        self.action = 'add'
+
+    @property
+    def _url(self):
+        return ('/api/v1.0/connections/{}/conn_id/{}/'
+                'protocol/{}/host/{}/port/{}'.format(
+                    self.action, self.conn_id,
+                    self.protocol, self.host, self.port))
+
+    def test_on_get_missing_config_file(self):
+        doc = {
+            'description': 'Missing Configuration File',
+            'message': 'Internal Server Error'
+        }
+        result = self.simulate_get(self._url, headers=self._headers)
+        assert result.json == doc
+        assert result.status == falcon.HTTP_500
+
+    @mock.patch.object(AirflowAddConnectionResource, 'retrieve_config')
+    def test_on_get_invalid_action(self, mock_config):
+        self.action = 'invalid_action'
+        doc = {
+            'title': 'Invalid parameter',
+            'description': ('The "action" parameter is invalid.'
+                            ' Invalid Paremeters for Adding Airflow'
+                            ' Connection')
+        }
+        mock_config.return_value = 'some_url'
+        result = self.simulate_get(self._url, headers=self._headers)
+        assert result.json == doc
+        assert result.status == falcon.HTTP_400
+        mock_config.assert_called_once_with('base', 'web_server')
+
+    @mock.patch('shipyard_airflow.airflow_client.requests')
+    @mock.patch.object(AirflowAddConnectionResource, 'retrieve_config')
+    def test_on_get_airflow_error(self, mock_config, mock_requests):
+        doc = {
+            'message': 'Error response from Airflow',
+            'description': "can't add connections in airflow"
+        }
+        mock_response = {
+            'output': {
+                'stderr': "can't add connections in airflow"
+            }
+        }
+        mock_requests.get.return_value.json.return_value = mock_response
+        mock_config.return_value = 'some_url'
+        result = self.simulate_get(self._url, headers=self._headers)
+        assert result.json == doc
+        assert result.status == falcon.HTTP_400
+        mock_config.assert_called_once_with('base', 'web_server')
+
+    @mock.patch('shipyard_airflow.airflow_client.requests')
+    @mock.patch.object(AirflowAddConnectionResource, 'retrieve_config')
+    def test_on_get_airflow_success(self, mock_config, mock_requests):
+        doc = {
+            'type': 'success',
+            'message': 'Airflow Success',
+        }
+        mock_response = {
+            'output': {
+                'stderr': None,
+                'stdout': 'Airflow Success'
+            }
+        }
+        mock_requests.get.return_value.json.return_value = mock_response
+        mock_config.return_value = 'some_url'
+        result = self.simulate_get(self._url, headers=self._headers)
+        assert result.json == doc
+        assert result.status == falcon.HTTP_200
+        mock_config.assert_called_once_with('base', 'web_server')
+
+
+class AirflowDeleteConnectionResource(BaseTesting):
+    def setUp(self):
+        self.action = 'delete'
+        super().setUp()
+
+    @property
+    def _url(self):
+        return '/api/v1.0/connections/{}/conn_id/{}'.format(
+            self.action, self.conn_id)
+
+    def test_on_get_missing_config_file(self):
+        doc = {
+            'description': 'Missing Configuration File',
+            'message': 'Internal Server Error'
+        }
+        result = self.simulate_get(self._url, headers=self._headers)
+        assert result.json == doc
+        assert result.status == falcon.HTTP_500
+
+    @mock.patch.object(AirflowDeleteConnectionResource, 'retrieve_config')
+    def test_on_get_invalid_action(self, mock_config):
+        self.action = 'invalid_action'
+        doc = {
+            'title': 'Invalid parameter',
+            'description': ('The "action" parameter is invalid.'
+                            ' Invalid Paremeters for Deleting Airflow'
+                            ' Connection')
+        }
+        mock_config.return_value = 'some_url'
+        result = self.simulate_get(self._url, headers=self._headers)
+        assert result.json == doc
+        assert result.status == falcon.HTTP_400
+        mock_config.assert_called_once_with('base', 'web_server')
+
+    @mock.patch('shipyard_airflow.airflow_client.requests')
+    @mock.patch.object(AirflowDeleteConnectionResource, 'retrieve_config')
+    def test_on_get_airflow_error(self, mock_config, mock_requests):
+        doc = {
+            'message': 'Error response from Airflow',
+            'description': "can't delete connections in airflow"
+        }
+        mock_response = {
+            'output': {
+                'stderr': "can't delete connections in airflow"
+            }
+        }
+        mock_requests.get.return_value.json.return_value = mock_response
+        mock_config.return_value = 'some_url'
+        result = self.simulate_get(self._url, headers=self._headers)
+
+        assert result.json == doc
+        assert result.status == falcon.HTTP_400
+        mock_config.assert_called_once_with('base', 'web_server')
+
+    @mock.patch('shipyard_airflow.airflow_client.requests')
+    @mock.patch.object(AirflowDeleteConnectionResource, 'retrieve_config')
+    def test_on_get_airflow_success(self, mock_config, mock_requests):
+        doc = {
+            'type': 'success',
+            'message': 'Airflow Success',
+        }
+        mock_response = {
+            'output': {
+                'stderr': None,
+                'stdout': 'Airflow Success'
+            }
+        }
+        mock_requests.get.return_value.json.return_value = mock_response
+        mock_config.return_value = 'some_url'
+        result = self.simulate_get(self._url, headers=self._headers)
+        assert result.json == doc
+        assert result.status == falcon.HTTP_200
+        mock_config.assert_called_once_with('base', 'web_server')
+
+
+class AirflowListConnectionsResource(BaseTesting):
+    def setUp(self):
+        self.action = 'list'
+        super().setUp()
+
+    @property
+    def _url(self):
+        return '/api/v1.0/connections/{}'.format(self.action)
+
+    def test_on_get_missing_config_file(self):
+        doc = {
+            'description': 'Missing Configuration File',
+            'message': 'Internal Server Error'
+        }
+        result = self.simulate_get(self._url, headers=self._headers)
+        assert result.json == doc
+        assert result.status == falcon.HTTP_500
+
+    @mock.patch.object(AirflowListConnectionsResource, 'retrieve_config')
+    def test_on_get_invalid_action(self, mock_config):
+        self.action = 'invalid_action'
+        doc = {
+            'title': 'Invalid parameter',
+            'description': ('The "action" parameter is invalid.'
+                            ' Invalid Paremeters for listing Airflow'
+                            ' Connections')
+        }
+        mock_config.return_value = 'some_url'
+        result = self.simulate_get(self._url, headers=self._headers)
+
+        assert result.json == doc
+        assert result.status == falcon.HTTP_400
+        mock_config.assert_called_once_with('base', 'web_server')
+
+    @mock.patch('shipyard_airflow.airflow_client.requests')
+    @mock.patch.object(AirflowListConnectionsResource, 'retrieve_config')
+    def test_on_get_airflow_error(self, mock_config, mock_requests):
+        doc = {
+            'message': 'Error response from Airflow',
+            'description': "can't list connections in airlfow"
+        }
+        mock_response = {
+            'output': {
+                'stderr': "can't list connections in airlfow"
+            }
+        }
+        mock_requests.get.return_value.json.return_value = mock_response
+        mock_config.return_value = 'some_url'
+        result = self.simulate_get(self._url, headers=self._headers)
+
+        assert result.json == doc
+        assert result.status == falcon.HTTP_400
+        mock_config.assert_called_once_with('base', 'web_server')
+
+    @mock.patch('shipyard_airflow.airflow_client.requests')
+    @mock.patch.object(AirflowListConnectionsResource, 'retrieve_config')
+    def test_on_get_airflow_success(self, mock_config, mock_requests):
+        doc = {
+            'type': 'success',
+            'message': 'Airflow Success',
+        }
+        mock_response = {
+            'output': {
+                'stderr': None,
+                'stdout': 'Airflow Success'
+            }
+        }
+        mock_requests.get.return_value.json.return_value = mock_response
+        mock_config.return_value = 'some_url'
+        result = self.simulate_get(self._url, headers=self._headers)
+
+        assert result.json == doc
+        assert result.status == falcon.HTTP_200
+        mock_config.assert_called_once_with('base', 'web_server')

tox.ini

@@ -0,0 +1,17 @@
+[tox]
+envlist = py35, pep8
+
+[testenv]
+deps = -r{toxinidir}/requirements.txt
+       -r{toxinidir}/test-requirements.txt
+setenv=
+  PYTHONWARNING=all
+commands=
+  pytest \
+    {posargs}
+
+[testenv:pep8]
+commands = flake8 {posargs}
+
+[flake8]
+ignore=E302,H306,D100,D101,D102