Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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()
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)
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)
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()
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)
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()
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():
# 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