Commit eaf65f0c authored by J. Fernando Sánchez's avatar J. Fernando Sánchez
Browse files

First tests

parent a5e79bea
from senpy.plugins import SenpyPlugin
class Prueba(SenpyPlugin):
def __init__(self, **kwargs):
super(Prueba, self).__init__(name="prueba",
plugin = Prueba()
......@@ -18,7 +18,7 @@
Sentiment analysis server in Python
VERSION = "0.2.5"
VERSION = "0.2.6"
import extensions
import blueprints
......@@ -17,8 +17,11 @@
Simple Sentiment Analysis server
from flask import Blueprint, render_template, request, jsonify, current_app
import json
import logging
logger = logging.getLogger(__name__)
from flask import Blueprint, render_template, request, jsonify, current_app
nif_blueprint = Blueprint("NIF Sentiment Analysis Server", __name__)
......@@ -86,10 +89,12 @@ def home(entries=None):
algo = get_params(request).get("algorithm", None)
specific_params = current_app.senpy.parameters(algo)
params = get_params(request, specific_params)
except ValueError as ex:
return ex.message
response = current_app.senpy.analyse(**params)
return jsonify(response)
except ValueError as ex:
return ex.message
except Exception as ex:
return jsonify(status="400", message=ex.message)
def default():
import os
import sys
import imp
import logging
logger = logging.getLogger(__name__)
from flask import current_app
from collections import defaultdict
from .plugins import SentimentPlugin, EmotionPlugin
......@@ -11,29 +12,33 @@ try:
except ImportError:
from flask import _request_ctx_stack as stack
from blueprints import nif_blueprint
from .blueprints import nif_blueprint
from git import Repo, InvalidGitRepositoryError
class Senpy(object):
""" Default Senpy extension for Flask """
def __init__(self, app=None, plugin_folder="plugins"): = app
base_folder = os.path.join(os.path.dirname(__file__), "plugins")
self.search_folders = (folder for folder in (base_folder, plugin_folder, '/tmp/plugins')
if folder and os.path.isdir(folder))
self._search_folders = set()
self._outdated = True
for folder in (base_folder, plugin_folder):
if app is not None:
def init_app(self, app):
""" Initialise a flask app to add plugins to its context """
Note: I'm not particularly fond of adding and app.senpy, but
I can't think of a better way to do it.
def init_app(self, app, plugin_folder="plugins"):
app.senpy = self
#app.config.setdefault('SQLITE3_DATABASE', ':memory:')
# Use the newstyle teardown_appcontext if it's available,
# otherwise fall back to the request context
if hasattr(app, 'teardown_appcontext'):
......@@ -42,9 +47,18 @@ class Senpy(object):
def add_folder(self, folder):
if os.path.isdir(folder):
self._outdated = True
return True
return False
def analyse(self, **params):
algo = None
print("analysing with params: {}".format(params))
logger.debug("analysing with params: {}".format(params))
if "algorithm" in params:
algo = params["algorithm"]
elif self.plugins:
......@@ -55,36 +69,39 @@ class Senpy(object):
return resp
return {"status": 500, "message": "No valid algorithm"}
return {"status": 400, "message": "The algorithm '{}' is not valid".format(algo) }
def default_plugin(self):
if self.plugins:
candidate = self.filter_plugins(enabled=True).keys()[0]
print("Default: {}".format(candidate))
candidates = self.filter_plugins(enabled=True)
if len(candidates)>1:
candidate = candidates.keys()[0]
logger.debug("Default: {}".format(candidate))
return candidate
return Exception("No algorithm")
return None
def parameters(self, algo):
return getattr(self.plugins.get(algo or self.default_plugin), "params", {})
def enable_plugin(self, plugin):
def disable_plugin(self, plugin):
def reload_plugin(self, plugin):
print("Reloading {}".format(plugin))
logger.debug("Reloading {}".format(plugin))
plug = self.plugins[plugin]
nplug = self._load_plugin(plug.module, plug.path)
del self.plugins[plugin]
self.plugins[] = nplug
def _load_plugin(self, plugin, search_folder, enabled=True):
logger.debug("Loading plugins")
(fp, pathname, desc) = imp.find_module(plugin)
tmp = imp.load_module(plugin, fp, pathname, desc).plugin
tmp.path = search_folder
......@@ -96,19 +113,24 @@ class Senpy(object):
if not hasattr(tmp, "enabled"):
tmp.enabled = enabled
tmp.module = plugin
except Exception as ex:
tmp = None
logger.debug("Exception importing {}: {}".format(plugin, ex))
return tmp
def _load_plugins(self):
plugins = {}
for search_folder in self.search_folders:
for search_folder in self._search_folders:
for item in os.listdir(search_folder):
if os.path.isdir(os.path.join(search_folder, item)) \
and os.path.exists(
os.path.join(search_folder, item, "")):
plugin = self._load_plugin(item, search_folder)
if plugin:
plugins[] = plugin
self._outdated = False
return plugins
def teardown(self, exception):
......@@ -126,30 +148,36 @@ class Senpy(object):
def plugins(self):
""" Return the plugins registered for a given application. """
ctx =
if ctx is not None:
if not hasattr(self, '_plugins'):
self._plugins = self._load_plugins()
return self._plugins
if not hasattr(ctx, 'senpy_plugins') or self._outdated:
ctx.senpy_plugins = self._load_plugins()
return ctx.senpy_plugins
def filter_plugins(self, **kwargs):
""" Filter plugins by different criteria """
def matches(plug):
res = all(getattr(plug, k, None)==v for (k,v) in kwargs.items())
print("matching {} with {}: {}".format(, kwargs, res))
res = all(getattr(plug, k, None) == v for (k, v) in kwargs.items())
logger.debug("matching {} with {}: {}".format(,
return res
if not kwargs:
return self.plugins
return {n:p for n,p in self.plugins.items() if matches(p)}
return {n:p for n, p in self.plugins.items() if matches(p)}
def sentiment_plugins(self):
return (plugin for plugin in self.plugins if
isinstance(plugin, SentimentPlugin))
""" Return only the sentiment plugins """
return {p:plugin for p, plugin in self.plugins.items() if
isinstance(plugin, SentimentPlugin)}
if __name__ == '__main__':
from flask import Flask
app = Flask(__name__)
sp = Senpy()
with app.app_context():
with APP.app_context():
import logging
logger = logging.getLogger(__name__)
PARAMS = {"input": {"aliases": ["i", "input"],
"required": True,
"help": "Input text"
......@@ -30,7 +34,7 @@ PARAMS = {"input": {"aliases": ["i", "input"],
class SenpyPlugin(object):
def __init__(self, name=None, version=None, extraparams=None, params=None):
print("Initing {}".format(name))
logger.debug("Initialising {}".format(name)) = name
self.version = version
if params:
* Method: POST
* Parameters: JSON Object (that is copied to the result)
* text
* query
* language
* topic
* Example response:
{"data": [{"text": "I love Titanic.", "id":1234, "polarity": 4},
{"text": "I hate Titanic.", "id":4567, "polarity": 0}]}
import requests
import json
def analyse(texts):
parameters = {"data": []}
if isinstance(texts, list):
for text in texts:
parameters["data"].append({"text": text})
parameters["data"].append({"text": texts})
res =, json.dumps(parameters))
return res.json()
def test():
print analyse("I love Titanic")
print analyse(["I love Titanic", "I hate Titanic"])
if __name__ == "__main__":
import os
import logging
import unittest.mock as mock
except ImportError:
import mock
from senpy.extensions import Senpy
from flask import Flask
from flask.ext.testing import TestCase
def check_dict(indic, template):
return all(item in indic.items() for item in template.items())
class Blueprints_Test(TestCase):
def create_app(self): = Flask("test_extensions")
self.senpy = Senpy()
self.dir = os.path.join(os.path.dirname(__file__), "..")
def test_home(self):
""" Calling with no arguments should ask the user for more arguments """
resp = self.client.get("/")
assert resp.json["status"] == "failed"
atleast = {
"status": "failed",
"message": "Missing or invalid parameters",
assert check_dict(resp.json, atleast)
from senpy.plugins import SenpyPlugin
plugin = SenpyPlugin("dummy")
import os
import logging
import unittest.mock as mock
except ImportError:
import mock
from senpy.extensions import Senpy
from flask import Flask
from flask.ext.testing import TestCase
class Extensions_Test(TestCase):
def create_app(self): = Flask("test_extensions")
self.senpy = Senpy()
self.dir = os.path.join(os.path.dirname(__file__), "..")
def test_init(self):
""" Initialising the app with the extension. """
assert hasattr(, "senpy")
tapp = Flask("temp app")
tsen = Senpy(tapp)
assert hasattr(tapp, "senpy")
def test_discovery(self):
""" Discovery of plugins in given folders. """
assert self.dir in self.senpy._search_folders
print self.senpy.plugins
assert "dummy" in self.senpy.plugins
def test_enabling(self):
""" Enabling a plugin """
assert self.senpy.plugins["dummy"].enabled
def test_disabling(self):
""" Disabling a plugin """
assert self.senpy.plugins["dummy"].enabled == False
def test_default(self):
""" Default plugin should be set """
assert self.senpy.default_plugin
assert self.senpy.default_plugin == "dummy"
def test_analyse(self):
""" Using a plugin """
with mock.patch.object(self.senpy.plugins["dummy"], "analyse") as mocked:
self.senpy.analyse(algorithm="dummy", input="tupni", output="tuptuo")
self.senpy.analyse(input="tupni", output="tuptuo")
mocked.assert_any_call(input="tupni", output="tuptuo", algorithm="dummy")
mocked.assert_any_call(input="tupni", output="tuptuo")
for plug in self.senpy.plugins:
resp = self.senpy.analyse(input="tupni")
logging.debug("Response: {}".format(resp))
assert resp["status"] == 400
def test_filtering(self):
""" Filtering plugins """
assert len(self.senpy.filter_plugins(name="dummy"))>0
assert not len(self.senpy.filter_plugins(name="notdummy"))
assert self.senpy.filter_plugins(name="dummy", enabled=True)
assert not len(self.senpy.filter_plugins(name="dummy", enabled=True))
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment