senpy/tests/test_extensions.py

from __future__ import print_function
import os
from copy import deepcopy
import logging


from functools import partial
from senpy.extensions import Senpy
from senpy import plugins
from senpy.models import Error, Results, Entry, EmotionSet, Emotion, Plugin
from senpy import api
from flask import Flask
from unittest import TestCase


def analyse(instance, **kwargs):
    basic = api.parse_params(kwargs, api.API_PARAMS)
    request = api.parse_call(basic)
    return instance.analyse(request)


class ExtensionsTest(TestCase):
    def setUp(self):
        self.app = Flask('test_extensions')
        self.examples_dir = os.path.join(os.path.dirname(__file__), '..', 'example-plugins')
        self.senpy = Senpy(plugin_folder=self.examples_dir,
                           app=self.app,
                           default_plugins=False)
        self.senpy.deactivate_all()
        self.senpy.activate_plugin("Dummy", sync=True)
        self.app.config['TESTING'] = True  # Tell Flask not to catch Exceptions

    def test_init(self):
        """ Initialising the app with the extension.  """
        assert hasattr(self.app, "senpy")
        tapp = Flask("temp app")
        self.senpy.init_app(tapp)
        assert hasattr(tapp, "senpy")

    def test_discovery(self):
        """ Discovery of plugins in given folders.  """
        # noinspection PyProtectedMember
        print(self.senpy.plugins())
        assert self.senpy.get_plugin("dummy")

    def test_add_delete(self):
        '''Should be able to add and delete new plugins. '''
        new = plugins.Analyser(name='new', description='new', version=0)
        self.senpy.add_plugin(new)
        assert new in self.senpy.plugins(is_activated=False)
        self.senpy.delete_plugin(new)
        assert new not in self.senpy.plugins(is_activated=False)

    def test_adding_folder(self):
        """ It should be possible for senpy to look for plugins in more folders. """
        senpy = Senpy(plugin_folder=None,
                      app=self.app,
                      default_plugins=False)
        assert not senpy.analysis_plugins()
        senpy.add_folder(self.examples_dir)
        assert senpy.plugins(plugin_type=plugins.Analyser, is_activated=False)
        self.assertRaises(AttributeError, senpy.add_folder, 'DOES NOT EXIST')

    def test_installing(self):
        """ Installing a plugin """
        info = {
            'name': 'TestPip',
            'module': 'mynoop',
            'description': None,
            'requirements': ['noop'],
            'version': 0
        }
        module = plugins.from_info(info, root=self.examples_dir, install=True)
        assert module.name == 'TestPip'
        assert module
        import noop
        dir(noop)

    def test_enabling(self):
        """ Enabling a plugin """
        self.senpy.activate_all(sync=True)
        assert len(self.senpy.plugins()) >= 3
        assert self.senpy.get_plugin("Sleep").is_activated

    def test_installing_nonexistent(self):
        """ Fail if the dependencies cannot be met """
        info = {
            'name': 'TestPipFail',
            'module': 'dummy',
            'description': None,
            'requirements': ['IAmMakingThisPackageNameUpToFail'],
            'version': 0
        }
        with self.assertRaises(Error):
            plugins.install_deps(info)

    def test_disabling(self):
        """ Disabling a plugin """
        self.senpy.deactivate_all(sync=True)
        assert not self.senpy.get_plugin("dummy").is_activated
        assert not self.senpy.get_plugin("sleep").is_activated

    def test_default(self):
        """ Default plugin should be set """
        assert self.senpy.default_plugin
        assert self.senpy.default_plugin.name == "dummy"
        self.senpy.deactivate_all(sync=True)
        logging.debug("Default: {}".format(self.senpy.default_plugin))
        assert self.senpy.default_plugin is None

    def test_noplugin(self):
        """ Don't analyse if there isn't any plugin installed """
        self.senpy.deactivate_all(sync=True)
        self.assertRaises(Error, partial(analyse, self.senpy, input="tupni"))

    def test_analyse(self):
        """ Using a plugin """
        # I was using mock until plugin started inheriting
        # Leaf (defaultdict with  __setattr__ and __getattr__.
        r1 = analyse(self.senpy, algorithm="Dummy", input="tupni", output="tuptuo")
        r2 = analyse(self.senpy, input="tupni", output="tuptuo")
        assert r1.activities[0].algorithm == "endpoint:plugins/dummy_0.1"
        assert r2.activities[0].algorithm == "endpoint:plugins/dummy_0.1"
        assert r1.entries[0]['nif:isString'] == 'input'

    def test_analyse_empty(self):
        """ Trying to analyse when no plugins are installed should raise an error."""
        senpy = Senpy(plugin_folder=None,
                      app=self.app,
                      default_plugins=False)
        self.assertRaises(Error, senpy.analyse, Results(), [])

    def test_analyse_wrong(self):
        """ Trying to analyse with a non-existent plugin should raise an error."""
        self.assertRaises(Error, analyse, self.senpy, algorithm='DOES NOT EXIST', input='test')

    def test_analyse_jsonld(self):
        """ Using a plugin with JSON-LD input"""
        js_input = '''{
        "@id": "prueba",
        "@type": "results",
        "entries": [
          {"@id": "entry1",
           "nif:isString": "tupni",
           "@type": "entry"
          }
        ]
        }'''
        r1 = analyse(self.senpy,
                     algorithm="Dummy",
                     input=js_input,
                     informat="json-ld",
                     output="tuptuo")
        r2 = analyse(self.senpy,
                     input="tupni",
                     output="tuptuo")
        assert r1.activities[0].algorithm == "endpoint:plugins/dummy_0.1"
        assert r2.activities[0].algorithm == "endpoint:plugins/dummy_0.1"
        assert r1.entries[0]['nif:isString'] == 'input'

    def test_analyse_error(self):
        class ErrorPlugin(plugins.Analyser):
            author = 'nobody'
            version = 0
            ex = Error()

            def process(self, *args, **kwargs):
                raise self.ex

        m = ErrorPlugin(ex=Error('error in analysis', status=500))
        self.senpy.add_plugin(m)
        try:
            analyse(self.senpy, input='nothing', algorithm='ErrorPlugin')
            assert False
        except Error as ex:
            assert 'error in analysis' in ex['message']
            assert ex['status'] == 500

        m.ex = Exception('generic exception on analysis')

        try:
            analyse(self.senpy, input='nothing', algorithm='ErrorPlugin')
            assert False
        except Exception as ex:
            assert 'generic exception on analysis' in str(ex)

    def test_filtering(self):
        """ Filtering plugins """
        assert len(self.senpy.plugins(name="Dummy")) > 0
        assert not len(self.senpy.plugins(name="NotDummy"))
        assert self.senpy.plugins(name="Dummy", is_activated=True)
        self.senpy.deactivate_plugin("Dummy", sync=True)
        assert not len(self.senpy.plugins(name="Dummy",
                                          is_activated=True))

    def test_load_default_plugins(self):
        senpy = Senpy(plugin_folder=self.examples_dir, default_plugins=True)
        assert len(senpy.plugins(is_activated=False)) > 1

    def test_convert_emotions(self):
        self.senpy.activate_all(sync=True)
        plugin = Plugin({
            'id': 'imaginary',
            'onyx:usesEmotionModel': 'emoml:fsre-dimensions'
        })
        eSet1 = EmotionSet()
        activity = plugin.activity()
        eSet1.prov(activity)
        eSet1['onyx:hasEmotion'].append(Emotion({
            'emoml:arousal': 1,
            'emoml:potency': 0,
            'emoml:valence': 0
        }))
        response = Results({
            'activities': [activity],
            'entries': [Entry({
                'nif:isString': 'much ado about nothing',
                'onyx:hasEmotionSet': [eSet1]
            })]
        })
        params = {'emotion-model': 'emoml:big6',
                  'algorithm': ['conversion'],
                  'conversion': 'full'}
        r1 = deepcopy(response)
        r1.parameters = params
        self.senpy.analyse(r1)
        assert len(r1.entries[0].emotions) == 2
        params['conversion'] = 'nested'
        r2 = deepcopy(response)
        r2.parameters = params
        self.senpy.analyse(r2)
        assert len(r2.entries[0].emotions) == 1
        assert r2.entries[0].emotions[0]['prov:wasDerivedFrom'] == eSet1
        params['conversion'] = 'filtered'
        r3 = deepcopy(response)
        r3.parameters = params
        self.senpy.analyse(r3)
        assert len(r3.entries[0].emotions) == 1
        r3.jsonld()