initial data route module & unit test

inventory_provider/__init__.py

+"""
+automatically invoked app factory
+"""
+import logging
+import os
+from flask import Flask
+
+
+def create_app():
+    """
+    overrides default settings with those found
+    in the file read from env var SETTINGS_FILENAME
+
+    :return: a new flask app instance
+    """
+
+    app = Flask(__name__)
+    app.secret_key = "super secret session key"
+
+    from inventory_provider import data_routes
+    app.register_blueprint(data_routes.routes, url_prefix='/data')
+
+    if "SETTINGS_FILENAME" not in os.environ:
+        assert False, \
+            "environment variable SETTINGS_FILENAME' must be defined"
+    app.config.from_envvar("SETTINGS_FILENAME")
+
+    assert "INVENTORY_PROVIDER_CONFIG_FILENAME" in app.config, (
+        "INVENTORY_PROVIDER_CONFIG_FILENAME not defined in %s"
+        % os.environ["SETTINGS_FILENAME"])
+
+    assert os.path.isfile(app.config["INVENTORY_PROVIDER_CONFIG_FILENAME"]), (
+        "config file '%s' not found" %
+        app.config["INVENTORY_PROVIDER_CONFIG_FILENAME"])
+
+    from inventory_provider import config
+    with open(app.config["INVENTORY_PROVIDER_CONFIG_FILENAME"]) as f:
+        # test the config file can be loaded
+        config.load(f)
+
+    logging.debug(app.config)
+
+    return app

inventory_provider/data_routes.py

+import functools
+import json
+
+from flask import Blueprint, request, Response
+#render_template, url_for
+
+routes = Blueprint("python-utils-ui-routes", __name__)
+
+VERSION = {
+    "api": "0.1",
+    "module": "0.1"
+}
+
+
+def require_accepts_json(f):
+    """
+    used as a route handler decorator to return an error
+    unless the request allows responses with type "application/json"
+    :param f: the function to be decorated
+    :return: the decorated function
+    """
+    @functools.wraps(f)
+    def decorated_function(*args, **kwargs):
+        # TODO: use best_match to disallow */* ...?
+        if not request.accept_mimetypes.accept_json:
+            return Response(
+                response="response will be json",
+                status=406,
+                mimetype="text/html")
+        return f(*args, **kwargs)
+    return decorated_function
+
+
+@routes.route("/version", methods=['GET', 'POST'])
+@require_accepts_json
+def version():
+    return Response(
+        json.dumps(VERSION),
+        mimetype="application/json"
+    )

test/test_data_routes.py

+import json
+import logging
+import os
+import tempfile
+
+import pytest
+
+import inventory_provider
+
+logging.basicConfig(level=logging.DEBUG)
+
+DEFAULT_REQUEST_HEADERS = {
+    "Content-type": "application/json",
+    "Accept": ["application/json"]
+}
+
+MODULE_DIR = os.path.realpath(os.path.join(
+    os.path.dirname(__file__),
+    "..",
+    "inventory_provider"))
+
+DATA_CONFIG = {
+    "alarms-db": {
+        "hostname": "test-dashboard01.geant.org",
+        "dbname": "alarms",
+        "username": "invprov",
+        "password": "invprov-password"
+    },
+    "oid_list.conf": os.path.join(
+        MODULE_DIR,
+        "oid_list.conf"),
+    "routers_community.conf": os.path.join(
+        MODULE_DIR,
+        "routers_community.conf"),
+    "ssh": {
+        "private-key": os.path.join(
+            MODULE_DIR,
+            "ssh",
+            "monitor_dsa"),
+        "known-hosts": os.path.join(
+            MODULE_DIR,
+            "ssh",
+            "monitor_known_hosts")
+    }
+}
+
+
+@pytest.fixture
+def app_config():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        data_config_filename = os.path.join(tmpdir, "config.json")
+        with open(data_config_filename, "w") as f:
+            f.write(json.dumps(DATA_CONFIG))
+
+        app_config_filename = os.path.join(tmpdir, "app.config")
+        with open(app_config_filename, "w") as f:
+            f.write("%s = '%s'\n" % (
+                "INVENTORY_PROVIDER_CONFIG_FILENAME",
+                data_config_filename))
+
+        yield app_config_filename
+
+
+@pytest.fixture
+def client(app_config):
+    os.environ["SETTINGS_FILENAME"] = app_config
+    # with release_webservice.create_app().test_client() as c:
+    #         yield c
+    with inventory_provider.create_app().test_client() as c:
+        yield c
+
+
+def test_version_request(client):
+    rv = client.post(
+        "data/version",
+        headers=DEFAULT_REQUEST_HEADERS)
+    assert rv.status_code == 200
+    logging.debug("rv.data: '%s'" % rv.data.decode("utf-8"))
+    print(rv.data.decode("utf-8"))
+        # validate(json.loads(rv.data.decode("utf-8")), SERVICES_SCHEMA)