# retoor <retoor@molodetz.nl>

import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor
import time

from devplacepy.services.nodes.wake import WakeHub


def run(coro):
    with ThreadPoolExecutor(max_workers=1) as pool:
        return pool.submit(asyncio.run, coro).result()


def test_a_disabled_or_detached_hub_just_sleeps():
    async def scenario():
        hub = WakeHub()
        started = time.monotonic()
        assert await hub.wait_queued(0.05) is False
        assert time.monotonic() - started >= 0.04
        hub.enabled = False
        hub.notify_queued()
        assert hub.job_event("x") is None

    run(scenario())


def test_notify_wakes_the_dispatcher_immediately():
    async def scenario():
        hub = WakeHub()
        hub.attach()
        asyncio.get_running_loop().call_later(0.02, hub.notify_queued)
        started = time.monotonic()
        assert await hub.wait_queued(5) is True
        assert time.monotonic() - started < 1

    run(scenario())


def test_a_notification_before_the_wait_is_not_lost():
    async def scenario():
        hub = WakeHub()
        hub.attach()
        hub.notify_queued()
        assert await hub.wait_queued(5) is True
        assert await hub.wait_queued(0.02) is False

    run(scenario())


def test_job_events_are_per_job_and_only_for_the_attached_loop():
    async def scenario():
        hub = WakeHub()
        assert hub.job_event("a") is None
        hub.attach()
        first, second = hub.job_event("a"), hub.job_event("b")
        assert first is not second and hub.job_event("a") is first
        hub.job_done("a")
        assert first.is_set() and not second.is_set()
        hub.forget("a")
        assert hub.job_event("a") is not first

    run(scenario())


def test_another_loop_gets_no_event():
    hub = WakeHub()
    seen = []

    async def attach():
        hub.attach()

    run(attach())

    async def foreign():
        seen.append(hub.job_event("z"))

    run(foreign())
    assert seen == [None]


def test_notifying_from_another_thread_wakes_the_loop():
    async def scenario():
        hub = WakeHub()
        hub.attach()
        threading.Timer(0.02, hub.notify_queued).start()
        assert await hub.wait_queued(5) is True

    run(scenario())