senpy/tests/test_extensions.py

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

try:
    from unittest import mock
except ImportError:
    import mock

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


class ExtensionsTest(TestCase):
    def setUp(self):
        self.app = Flask('test_extensions')
        self.dir = os.path.join(os.path.dirname(__file__))
        self.senpy = Senpy(plugin_folder=self.dir,
                           app=self.app,
                           default_plugins=False)
        self.senpy.activate_plugin("Dummy", sync=True)

    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
        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 """
        info = {
            'name': 'TestPip',
            'module': 'dummy',
            'description': None,
            'requirements': ['noop'],
            'version': 0
        }
        root = os.path.join(self.dir, 'plugins', 'dummy_plugin')
        name, module = self.senpy._load_plugin_from_info(info, root=root)
        assert name == 'TestPip'
        assert module
        import noop
        dir(noop)
        self.senpy.install_deps()

    def test_installing(self):
        """ Enabling a plugin """
        self.senpy.activate_all(sync=True)
        assert len(self.senpy.plugins) >= 3
        assert self.senpy.plugins["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
        }
        root = os.path.join(self.dir, 'plugins', 'dummy_plugin')
        with self.assertRaises(Error):
            name, module = self.senpy._load_plugin_from_info(info, root=root)

    def test_disabling(self):
        """ Disabling a plugin """
        self.senpy.deactivate_all(sync=True)
        assert not self.senpy.plugins["Dummy"].is_activated
        assert not self.senpy.plugins["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(self.senpy.analyse, input="tupni"))
        self.assertRaises(Error,
                          partial(
                              self.senpy.analyse,
                              input="tupni",
                              algorithm='Dummy'))

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

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

    def test_analyse_error(self):
        mm = mock.MagicMock()
        mm.id = 'magic_mock'
        mm.analyse_entries.side_effect = Error('error on analysis', status=500)
        self.senpy.plugins['MOCK'] = mm
        try:
            self.senpy.analyse(input='nothing', algorithm='MOCK')
            assert False
        except Error as ex:
            assert ex['message'] == 'error on analysis'
            assert ex['status'] == 500

        mm.analyse.side_effect = Exception('generic exception on analysis')
        mm.analyse_entries.side_effect = Exception(
            'generic exception on analysis')

        try:
            self.senpy.analyse(input='nothing', algorithm='MOCK')
            assert False
        except Error as ex:
            assert ex['message'] == 'generic exception on analysis'
            assert ex['status'] == 500

    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", is_activated=True)
        self.senpy.deactivate_plugin("Dummy", sync=True)
        assert not len(
            self.senpy.filter_plugins(name="Dummy", is_activated=True))

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

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

    # def test_async_plugin(self):
    #     """ We should accept multiprocessing plugins with async=False"""
    #     thread1 = self.senpy.activate_plugin("Async", sync=False)
    #     thread1.join(timeout=1)
    #     assert len(self.senpy.plugins['Async'].value) == 4

    #     resp = self.senpy.analyse(input='nothing', algorithm='Async')

    #     assert len(resp.entries[0].async_values) == 2
    #     self.senpy.activate_plugin("Async", sync=True)