How to use the pyee.BaseEventEmitter function in pyee

To help you get started, we’ve selected a few pyee examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github jfhbrook / pyee / tests / test_sync.py View on Github external
def test_once():
    """Test that `once()` method works propers.
    """

    # very similar to "test_emit" but also makes sure that the event
    # gets removed afterwards

    call_me = Mock()
    ee = BaseEventEmitter()

    def once_handler(data):
        assert data == 'emitter is emitted!'
        call_me()

    # Tests to make sure that after event is emitted that it's gone.
    ee.once('event', once_handler)

    ee.emit('event', 'emitter is emitted!')

    call_me.assert_called_once()

    assert ee._events['event'] == OrderedDict()
github jfhbrook / pyee / tests / test_sync.py View on Github external
def test_listener_removal_on_emit():
    """Test that a listener removed during an emit is called inside the current
    emit cycle.
    """

    call_me = Mock()
    ee = BaseEventEmitter()

    def should_remove():
        ee.remove_listener('remove', call_me)

    ee.on('remove', should_remove)
    ee.on('remove', call_me)

    ee.emit('remove')

    call_me.assert_called_once()

    call_me.reset_mock()

    # Also test with the listeners added in the opposite order
    ee = BaseEventEmitter()
    ee.on('remove', call_me)
github jfhbrook / pyee / tests / test_sync.py View on Github external
    BaseEventEmitter,
    EventEmitter
])
def test_emit_sync(cls):
    """Basic synchronous emission works"""

    call_me = Mock()
    ee = cls()

    @ee.on('event')
    def event_handler(data, **kwargs):
        call_me()
        assert data == 'emitter is emitted!'

    # Making sure data is passed propers
    ee.emit('event', 'emitter is emitted!', error=False)
github jfhbrook / pyee / tests / test_sync.py View on Github external
def test_properties_preserved():
    """Test that the properties of decorated functions are preserved."""

    call_me = Mock()
    call_me_also = Mock()
    ee = BaseEventEmitter()

    @ee.on('always')
    def always_event_handler():
        """An event handler."""
        call_me()

    @ee.once('once')
    def once_event_handler():
        """Another event handler."""
        call_me_also()

    assert always_event_handler.__doc__ == 'An event handler.'
    assert once_event_handler.__doc__ == 'Another event handler.'

    always_event_handler()
    call_me.assert_called_once()
github jfhbrook / pyee / tests / test_sync.py View on Github external
def test_inheritance():
    """Test that inheritance is preserved from object"""
    assert object in getmro(BaseEventEmitter)

    class example(BaseEventEmitter):
        def __init__(self):
            super(example, self).__init__()

    assert BaseEventEmitter in getmro(example)
    assert object in getmro(example)
github jfhbrook / pyee / tests / test_sync.py View on Github external
def test_once_removal():
    """Removal of once functions works
    """

    ee = BaseEventEmitter()

    def once_handler(data):
        pass

    handle = ee.once('event', once_handler)

    assert handle == once_handler

    ee.remove_listener('event', handle)

    assert ee._events['event'] == OrderedDict()
github jfhbrook / pyee / tests / test_sync.py View on Github external
def test_listener_removal():
    """Removing listeners removes the correct listener from an event."""

    ee = BaseEventEmitter()

    # Some functions to pass to the EE
    def first():
        return 1

    ee.on('event', first)

    @ee.on('event')
    def second():
        return 2

    @ee.on('event')
    def third():
        return 3

    def fourth():
github MycroftAI / adapt / adapt / parser.py View on Github external
# 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 pyee
import time
from adapt.expander import BronKerboschExpander
from adapt.tools.text.trie import Trie

__author__ = 'seanfitz'


class Parser(pyee.BaseEventEmitter):
    """
    Coordinate a tagger and expander to yield valid parse results.
    """
    def __init__(self, tokenizer, tagger):
        pyee.BaseEventEmitter.__init__(self)
        self._tokenizer = tokenizer
        self._tagger = tagger

    def parse(self, utterance, context=None, N=1):
        """Used to find tags within utterance with a given confidence

        Args:
            utterance(str): conversational piece given by the user
            context(list): a list of entities
            N(int): number of results
        Returns: yield an object with the following fields