<commit_before>import asyncio
import threading
import pytest

from pydispatch import Property

class AsyncListener:
    def __init__(self, loop, mainloop):
        self.loop = loop
        self.mainloop = mainloop
        self.received_events = []
        self.received_event_data = []
        self.received_event_loops = []
        self.received_event_queue = asyncio.Queue(loop=mainloop)
        self.property_events = []
        self.property_event_kwargs = []
        self.property_event_map = {}
        self.property_event_loops = []
        self.property_queue = asyncio.Queue(loop=mainloop)
    async def on_event(self, *args, **kwargs):
        self.received_event_data.append({'args':args, 'kwargs':kwargs})
        name = kwargs.get('triggered_event')
        self.received_events.append(name)
        cur_loop = asyncio.get_event_loop()
        self.received_event_loops.append(cur_loop)
        asyncio.run_coroutine_threadsafe(self.event_queue_put({
            'name':name,
            'args':args,
            'kwargs':kwargs,
            'loop':cur_loop,
        }), loop=self.mainloop)
    async def event_queue_put(self, item):
        await self.received_event_queue.put(item)
    async def event_queue_get(self):
        item = await self.received_event_queue.get()
        self.received_event_queue.task_done()
        return item
    async def on_prop(self, obj, value, **kwargs):
        self.property_events.append(value)
        self.property_event_kwargs.append(kwargs)
        prop = kwargs['property']
        if prop.name not in self.property_event_map:
            self.property_event_map[prop.name] = []
        self.property_event_map[prop.name].append({'value':value, 'kwargs':kwargs})
        cur_loop = asyncio.get_event_loop()
        self.property_event_loops.append(cur_loop)
        asyncio.run_coroutine_threadsafe(self.prop_queue_put({
            'name':prop.name,
            'value':value,
            'kwargs':kwargs,
            'loop':cur_loop,
        }), loop=self.mainloop)
    async def prop_queue_put(self, item):
        await self.property_queue.put(item)
    async def prop_queue_get(self):
        item = await self.property_queue.get()
        self.property_queue.task_done()
        return item

class LoopThread(threading.Thread):
    def __init__(self, mainloop, sender):
        super().__init__()
        self.mainloop = mainloop
        self.sender = sender
        self.running = threading.Event()
        self.stopped = threading.Event()
    def run(self):
        loop = self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        listener = AsyncListener(loop, self.mainloop)
        sender = self.sender
        ev_names = sender._Dispatcher__events.keys()
        sender.bind_async(loop, **{name:listener.on_event for name in ev_names})
        prop_names = sender._Dispatcher__property_events.keys()
        sender.bind_async(loop, **{name:listener.on_prop for name in prop_names})
        self.listener = listener
        self.running.set()
        loop.run_until_complete(self.run_loop())
        self.stopped.set()
    async def run_loop(self):
        while self.running.is_set():
            await asyncio.sleep(.1)
    def stop(self):
        self.running.clear()
        self.stopped.wait()


@pytest.mark.asyncio
async def test_aio_listeners(sender_cls):

    class Sender(sender_cls):
        prop_a = Property()
        prop_b = Property()
        _events_ = ['on_test_a', 'on_test_b', 'on_test_c']

    sender = Sender()
    mainloop = asyncio.get_event_loop()

    async def send_event(name, *listener_threads):
        sender.trigger_event(name)
        for t in listener_threads:
            data = await t.listener.event_queue_get()
            assert data['name'] == name
            assert data['loop'] is t.loop
            assert data['loop'] is t.listener.loop
<commit_msg>Add a more basic aio test<commit_after>import asyncio
import threading
import pytest

from pydispatch import Property

class AsyncListener:
    def __init__(self, loop, mainloop):
        self.loop = loop
        self.mainloop = mainloop
        self.received_events = []
        self.received_event_data = []
        self.received_event_loops = []
        self.received_event_queue = asyncio.Queue(loop=mainloop)
        self.property_events = []
        self.property_event_kwargs = []
        self.property_event_map = {}
        self.property_event_loops = []
        self.property_queue = asyncio.Queue(loop=mainloop)
    async def on_event(self, *args, **kwargs):
        self.received_event_data.append({'args':args, 'kwargs':kwargs})
        name = kwargs.get('triggered_event')
        self.received_events.append(name)
        cur_loop = asyncio.get_event_loop()
        self.received_event_loops.append(cur_loop)
        asyncio.run_coroutine_threadsafe(self.event_queue_put({
            'name':name,
            'args':args,
            'kwargs':kwargs,
            'loop':cur_loop,
        }), loop=self.mainloop)
    async def event_queue_put(self, item):
        await self.received_event_queue.put(item)
    async def event_queue_get(self):
        item = await self.received_event_queue.get()
        self.received_event_queue.task_done()
        return item
    async def on_prop(self, obj, value, **kwargs):
        self.property_events.append(value)
        self.property_event_kwargs.append(kwargs)
        prop = kwargs['property']
        if prop.name not in self.property_event_map:
            self.property_event_map[prop.name] = []
        self.property_event_map[prop.name].append({'value':value, 'kwargs':kwargs})
        cur_loop = asyncio.get_event_loop()
        self.property_event_loops.append(cur_loop)
        asyncio.run_coroutine_threadsafe(self.prop_queue_put({
            'name':prop.name,
            'value':value,
            'kwargs':kwargs,
            'loop':cur_loop,
        }), loop=self.mainloop)
    async def prop_queue_put(self, item):
        await self.property_queue.put(item)
    async def prop_queue_get(self):
        item = await self.property_queue.get()
        self.property_queue.task_done()
        return item

class LoopThread(threading.Thread):
    def __init__(self, mainloop, sender):
        super().__init__()
        self.mainloop = mainloop
        self.sender = sender
        self.running = threading.Event()
        self.stopped = threading.Event()
    def run(self):
        loop = self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        listener = AsyncListener(loop, self.mainloop)
        sender = self.sender
        ev_names = sender._Dispatcher__events.keys()
        sender.bind_async(loop, **{name:listener.on_event for name in ev_names})
        prop_names = sender._Dispatcher__property_events.keys()
        sender.bind_async(loop, **{name:listener.on_prop for name in prop_names})
        self.listener = listener
        self.running.set()
        loop.run_until_complete(self.run_loop())
        self.stopped.set()
    async def run_loop(self):
        while self.running.is_set():
            await asyncio.sleep(.1)
    def stop(self):
        self.running.clear()
        self.stopped.wait()

@pytest.mark.asyncio
async def test_simple(sender_cls):

    class Sender(sender_cls):
        prop_a = Property()
        prop_b = Property()
        _events_ = ['on_test_a', 'on_test_b', 'on_test_c']

    loop = asyncio.get_event_loop()

    sender = Sender()
    listener = AsyncListener(loop, loop)

    ev_names = sender._Dispatcher__events.keys()
    sender.bind_async(loop, **{name:listener.on_event for name in ev_names})
    prop_names = sender._Dispatcher__property_events.keys()
    sender.bind_async(loop, **{name:listener.on_prop for name in prop_names})

    for name in ev_names:
        sender.trigger_event(name)
        data = await listener.event_queue_get()
        assert data['name'] == name
        assert data['loop'] is loop

    for name in prop_names:
        for i in range(10):
            setattr(sender, name, i)
            data = await listener.prop_queue_get()
            assert data['name'] == name
            assert data['value'] == i
            assert data['loop'] is loop

@pytest.mark.asyncio
async def test_multiple_loops(sender_cls):

    class Sender(sender_cls):
        prop_a = Property()
        prop_b = Property()
        _events_ = ['on_test_a', 'on_test_b', 'on_test_c']

    sender = Sender()
    mainloop = asyncio.get_event_loop()

    async def send_event(name, *listener_threads):
        sender.trigger_event(name)
        for t in listener_threads:
            data = await t.listener.event_queue_get()
            assert data['name'] == name
            assert data['loop'] is t.loop
            assert data['loop'] is t.listener.loop

{'fraction_non_alphanumeric': 0.10877472152761991, 'fraction_numerical': 0.0004546487838145033, 'mean_word_length': 2.6464981351015333, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4199225', 'n_tokens_mistral': 2684, 'n_tokens_neox': 2658, 'n_words': 493}