import array
import asyncio
import contextlib
import datetime
import functools
import random
import sys
from typing import Union

import netifaces
import pytest
import uvloop as uvloop

import conftest
import test_data

from aiolo import MultiCast, AioServer, Address, Message, PROTO_UDP, PROTO_UNIX, TimeTag, FRAC_PER_SEC, \
    NO_ARGS, Route, unix_timestamp_to_osc_timestamp, TT_IMMEDIATE, MultiCastAddress, Bundle, ANY_PATH, TypeSpec, \
    StartError, PROTO_DEFAULT, EPOCH_UTC, ANY_ARGS, JAN_1970, ThreadedServer, Midi, PROTO_TCP, INFINITY, \
    INFINITUM, TIMETAG, MIDI, NIL, FALSE, TRUE, BLOB, STRING, DOUBLE, INT64, Path, Sub, Subs, \
    compile_osc_address_pattern


def create_task(coro, cancel_timeout=conftest.CANCEL_TIMEOUT):
    loop = asyncio.get_event_loop()
    if sys.version_info[:2] >= (3, 7):
        task = asyncio.create_task(coro)
    else:
        task = loop.create_task(coro)
    if cancel_timeout >= 0:
        loop.call_later(cancel_timeout, task.cancel)
    return task


def get_ipv4(iface):
    try:
        return netifaces.ifaddresses(iface)[netifaces.AF_INET][0]['addr']
    except KeyError:
        return None


def now():
    return datetime.datetime.now(datetime.timezone.utc)


@pytest.fixture(params=[
    iface
    for iface in netifaces.interfaces()
    if get_ipv4(iface) and get_ipv4(iface) != '127.0.0.1'
])
def ip_interface(request):
    iface = request.param
    return get_ipv4(iface), iface


@pytest.fixture(params=[
    asyncio,
    uvloop
])
def event_loop(request):
    loop_mod = request.param
    loop = loop_mod.new_event_loop()
    asyncio.set_event_loop(loop)
    if loop_mod != uvloop:
        # uvloop in debug mode calls extract_stack, which results in "ValueError: call stack is not deep enough"
        # for Cython code
        loop.set_debug(True)
    with contextlib.closing(loop):
        yield loop


@pytest.fixture(params=[
    AioServer,
    ThreadedServer
])
def any_server_class(request, event_loop):
    return request.param


@pytest.fixture
async def any_server(any_server_class, unused_tcp_port_factory):
    tries = 0

    exc = None
    server = any_server_class(port=unused_tcp_port_factory())
    while tries < 3:
        try:
            server.start()
        except StartError as exc:
            await asyncio.sleep(0.1)
            server = any_server_class(port=unused_tcp_port_factory())
        else:
            break
        tries += 1
    if not server.running:
        raise StartError from exc
    yield server
    server.stop()


@pytest.fixture
async def server(event_loop, unused_tcp_port_factory):
    tries = 0
    while tries < 3:
        try:
            server = AioServer(port=unused_tcp_port_factory())
            server.start()
        except StartError:
            await asyncio.sleep(0.1)
        else:
            break
        tries += 1
    yield server
    server.stop()


@pytest.fixture(params=[
    lambda cls, port: cls(),
    lambda cls, port: cls(port=port()),
    lambda cls, port: cls(url='osc.udp://:%s' % port()),
    lambda cls, port: cls(url='osc.unix:///tmp/test-aiolo-%s.osc' % port()),
    lambda cls, port: cls(proto=PROTO_TCP, port=port()),
    lambda cls, port: cls(proto=PROTO_UDP, port=port()),
    lambda cls, port: cls(proto=PROTO_UNIX, port='/tmp/test-aiolo-%s.osc' % port()),
    lambda cls, port: cls(multicast=MultiCast('172.16.31.10', port=port())),
])
def server_init_factory(request, any_server_class, unused_tcp_port_factory):
    return functools.partial(request.param, any_server_class, unused_tcp_port_factory)


def test_server_init(server_init_factory):
    """
    Test that servers can be constructed and started with various call patterns.
    """
    server = server_init_factory()
    server.start()
    assert server.running
    assert server.port is not None
    server.stop()


def test_server_init_proto_default_udp_port(any_server_class, unused_tcp_port):
    server = any_server_class(proto=PROTO_DEFAULT, port=unused_tcp_port)
    server.start()
    assert server.running
    assert server.proto == PROTO_UDP
    assert server.port is not None
    server.stop()


def test_server_init_proto_default_unix_port(any_server_class, unused_tcp_port):
    server = any_server_class(proto=PROTO_DEFAULT, port='/tmp/test-aiolo-%s.osc' % unused_tcp_port)
    server.start()
    assert server.running
    assert server.proto == PROTO_UNIX
    assert server.port is not None
    server.stop()


@pytest.fixture(params=[
    lambda cls, port: cls(port='/foo'),
    lambda cls, port: cls(port=-1),
    lambda cls, port: cls(url='osc.foo://:%s' % port()),
])
def server_start_error_factory(request, any_server_class, unused_tcp_port_factory):
    return functools.partial(request.param, any_server_class, unused_tcp_port_factory)


def test_server_start_error(server_start_error_factory):
    server = server_start_error_factory()
    with pytest.raises(StartError):
        server.start()


def server_init_error_multicast_factory(any_server_class, unused_tcp_port_factory):
    port = unused_tcp_port_factory()
    return any_server_class(
        url='osc.tcp://:%s' % port,
        multicast=MultiCast('172.16.31.10', port=port()))


@pytest.fixture(params=[
    lambda cls, port: cls(multicast='foo'),
    lambda cls, port: cls(proto=-1, port=port()),
    lambda cls, port: cls(proto='foo', port=port()),
    lambda cls, port: cls(url='osc.tcp://:%s' % port(), port='10'),
    lambda cls, port: cls(url='osc.tcp://:%s' % port(), proto=PROTO_UDP),
    server_init_error_multicast_factory,
    lambda cls, port: cls(proto=PROTO_TCP, multicast=MultiCast('172.16.31.10', port=port())),
    lambda cls, port: cls(port='10', multicast=MultiCast('172.16.31.10', port=port())),
])
def server_init_error_factory(request, any_server_class, unused_tcp_port_factory):
    return functools.partial(request.param, any_server_class, unused_tcp_port_factory)


def test_server_init_error(server_init_error_factory):
    with pytest.raises((ValueError, TypeError)):
        server_init_error_factory()


@pytest.mark.asyncio
async def test_subs_unsub(event_loop):
    foo = Sub(Route('/foo', ANY_ARGS))
    bar = Sub(Route('/bar', ANY_ARGS))
    subs = foo | bar
    task = create_task(subscribe(subs, 3))
    await foo.pub('1')
    await bar.pub('2')
    await bar.unsub()
    await foo.pub('3')
    results = await task
    assert_results(results, {bar.route: ['2'], foo.route: ['1', '3']})


@pytest.mark.asyncio
async def test_multiple_addresses(any_server):
    """
    Test that multiple clients can send data to a single server
    """
    foo = any_server.route('/foo', int)
    addresses = [Address(port=any_server.port) for i in range(3)]
    task = create_task(subscribe(foo.sub(), 3))
    for i, address in enumerate(addresses):
        address.send(foo, i)
        # await asyncio.sleep(0.1)
    results = await task
    assert results == [[0], [1], [2]]


@pytest.fixture(params=[
    lambda port, ip, iface: (
        Address(url='osc.tcp://%s:%s' % (ip, port())),
        ip, iface),
    lambda port, ip, iface: (
        Address(url='osc.udp://%s:%s' % (ip, port())),
        ip, iface),
    lambda port, ip, iface: (
        Address(url='osc.unix:///aiolo-test-%s.osc' % port()),
        ip, iface),
    lambda port, ip, iface: (
        Address(proto=PROTO_TCP, host=ip, port=port()),
        ip, iface),
    lambda port, ip, iface: (
        Address(proto=PROTO_UDP, host=ip, port=port()),
        ip, iface),
    lambda port, ip, iface: (
        Address(proto=PROTO_UNIX, host=ip, port='/tmp/aiolo-test-%s.osc' % port()),
        ip, iface),
])
def address_init_factory(request, ip_interface, unused_tcp_port_factory):
    ip, iface = ip_interface
    return functools.partial(request.param, unused_tcp_port_factory, ip, iface)


def test_address_init(address_init_factory):
    """
    Test Address initialization and state.
    """
    address, ip, iface = address_init_factory()

    if iface is None or address.proto in (PROTO_UNIX, PROTO_DEFAULT):
        with pytest.raises(ValueError):
            address.interface = iface
    else:
        address.interface = iface
        assert address.interface == iface

    address, ip, iface = address_init_factory()
    if not ip or address.proto == PROTO_UNIX:
        with pytest.raises(ValueError):
            address.set_ip(ip)
    else:
        assert address.interface is None
        address.set_ip(ip)
        assert address.interface == iface

    address, ip, iface = address_init_factory()
    with pytest.raises(ValueError):
        address.interface = 'foobar0'
    with pytest.raises(ValueError):
        address.set_ip('foo.bar')
    with pytest.raises(ValueError):
        address.set_ip('172.16.31.10')


@pytest.mark.asyncio
async def test_multicast(any_server_class, unused_tcp_port):
    """
    Test multicast send/receive.
    """
    cluster = []
    multicast = MultiCast('172.16.31.10', port=unused_tcp_port)
    for i in range(3):
        server = any_server_class(multicast=multicast)
        server.start()
        cluster.append(server)

    address = MultiCastAddress(server=random.choice(cluster))
    foo = Route('/foo', str)

    for s in cluster:
        s.route(foo)

    task = create_task(
        subscribe(foo.sub(), 3 * len(cluster)),
        cancel_timeout=conftest.CANCEL_TIMEOUT * len(cluster))

    for d in ['foo', 'bar', 'baz']:
        address.send(foo, d)
        # await asyncio.sleep(1)

    results = await task

    assert results.count(['foo']) == len(cluster)
    assert results.count(['bar']) == len(cluster)
    assert results.count(['baz']) == len(cluster)
    for s in cluster:
        s.stop()


@pytest.mark.asyncio
async def test_multiple_servers(any_server_class, unused_tcp_port_factory):
    """
    Test that multiple servers can run okay in the same process.
    """
    servers = []
    addresses = []
    foo = Route('/foo', int)

    for i in range(3):
        server = any_server_class(url='osc.tcp://:%s' % unused_tcp_port_factory())
        server.route(foo)
        server.start()
        servers.append(server)
        address = Address(url=server.url)
        addresses.append(address)

    task = create_task(subscribe(foo.sub(), 3))
    for i, address in enumerate(addresses):
        address.send(foo, i)

    try:
        results = await task
        assert set([r[0] for r in results]) == {0, 1, 2}
    finally:
        for server in servers:
            server.stop()


def unpack_message_data():
    for t in test_data.ARGDEF_TEST_DATA:
        for typespec in t.typespecs:
            for value, expected in t.valid:
                if isinstance(typespec, int) and not isinstance(typespec, bool):
                    # Test unicode typespec like 'h'
                    yield t.path, chr(typespec), value, [expected]
                yield t.path, typespec, value, [expected]
                yield t.path, [typespec, typespec], [value, value], [expected, expected]


@pytest.mark.parametrize('path, typespec, data, expected', unpack_message_data())
def test_unpack_message(path, typespec, data, expected):
    """
    Test that types get parsed correctly
    """
    route = Route(path, typespec)
    message = Message(route, data)
    assert message.unpack() == expected

def unpack_message_type_error_data():
    for typespec_test_data in test_data.ARGDEF_TEST_DATA:
        for typespec in typespec_test_data.typespecs:
            for value in typespec_test_data.type_error:
                yield typespec_test_data.path, typespec, [value]


@pytest.mark.parametrize('path, typespec, data', unpack_message_type_error_data())
def test_unpack_message_type_error(path, typespec, data):
    """
    Test that TypeError is raised for invalid data types
    """
    route = Route(path, typespec)
    with pytest.raises(TypeError):
        message = Message(route, data)
        message.unpack()


def unpack_message_overflow_error_data():
    for t in test_data.ARGDEF_TEST_DATA:
        for typespec in t.typespecs:
            for value in t.overflow_error:
                yield t.path, typespec, [value]


@pytest.mark.parametrize('path, typespec, data', unpack_message_overflow_error_data())
def test_unpack_message_overflow_error(path, typespec, data):
    """
    Test that OverflowError is raised for invalid data types
    """
    route = Route(path, typespec)
    with pytest.raises(OverflowError):
        message = Message(route, data)
        message.unpack()


def unpack_message_value_error_data():
    for t in test_data.ARGDEF_TEST_DATA:
        for typespec in t.typespecs:
            for value in t.value_error:
                yield t.path, typespec, [value]


@pytest.mark.parametrize('path, typespec, data', unpack_message_value_error_data())
def test_unpack_message_value_error(path, typespec, data):
    """
    Test that ValueError is raised for invalid data values
    """
    route = Route(path, typespec)
    with pytest.raises(ValueError):
        message = Message(route, data)
        message.unpack()


@pytest.mark.parametrize('value, typespec', [
    [1, INT64],
    [1.0, DOUBLE],
    [0, INT64],
    [0.0, DOUBLE],
    [42, INT64],
    [42.0, DOUBLE],
    ['foo', STRING],
    [b'foo', BLOB],
    [array.array('b', b'foo'), BLOB],
    [True, TRUE],
    [False, FALSE],
    [None, NIL],
    [Midi(1, 2, 3, 4), MIDI],
    [now(), TIMETAG],
    [TimeTag(), TIMETAG],
    [INFINITY, INFINITUM],
])
def test_typespec_guess(value, typespec):
    assert TypeSpec.guess([value]) == TypeSpec(typespec)


@pytest.mark.parametrize('value', [
    {},
    [],
])
def test_typespec_guess_type_error(value):
    with pytest.raises(TypeError):
        TypeSpec.guess([value])


@pytest.mark.asyncio
async def test_reuse_subs(any_server):
    address = Address(url=any_server.url)
    foo = any_server.route('/foo', 's')
    tasks = asyncio.gather(*[
        create_task(subscribe(foo.sub(), 1))
        for i in range(3)
    ])
    address.send(foo, 'bar')
    with pytest.raises(asyncio.CancelledError):
        await tasks


@pytest.mark.asyncio
async def test_unroute(server):
    address = Address(url=server.url)
    foo = server.route('/foo', 's')
    task = create_task(subscribe(foo.sub(), 1))
    server.unroute(foo)
    address.send(foo, 'bar')
    with pytest.raises(asyncio.CancelledError):
        await task


@pytest.mark.asyncio
async def test_bundle(server):
    address = Address(url=server.url)
    foo = server.route('/foo', 's')
    bar = server.route('/bar', 's')
    baz = server.route('/baz', 's')
    subs = foo.sub() | bar.sub() | baz.sub()
    task = create_task(subscribe(subs, 3))
    address.bundle([
        Message(foo, 'foo'),
        Message(bar, 'bar'),
        Message(baz, 'baz'),
    ])
    results = await task
    assert_results(results, {foo: [['foo']], bar: [['bar']], baz: [['baz']]})


@pytest.mark.asyncio
async def test_bundle_delayed(server):
    address = Address(url=server.url)
    foo = server.route('/foo', 's')
    task = create_task(subscribe(foo.sub(), 1))
    address.bundle([
        Message(foo, 'now'),
        Bundle(
            [Message(foo, 'later')],
            timetag=now() + datetime.timedelta(seconds=0.5))
    ])
    address.delay(1, foo, 'later than later')
    results = list(await task)
    assert results == [['now']]
    assert server.events_pending
    assert 0 < server.next_event_delay < 1
    task = create_task(subscribe(foo.sub(), 1))
    results = list(await task)
    assert results == [['later']]
    assert server.events_pending
    assert 0 < server.next_event_delay < 1
    task = create_task(subscribe(foo.sub(), 1))
    results = list(await task)
    assert results == [['later than later']]
    assert not server.events_pending


def test_bundle_and_message_ops():
    foo = Route('/foo', 's')
    bar = Route('/bar', 's')
    baz = Route('/baz', 's')
    spaz = Route('/spaz', 's')
    bundle = Bundle()
    assert len(bundle) == 0
    assert bundle == Bundle()

    with pytest.raises(IndexError):
        assert bundle[0]

    message = Message(foo, 'foo')
    assert message == Message(foo, 'foo')
    assert bundle != Bundle(message)

    # Test hashable
    items = {bundle, message}
    assert len(items) == 2
    assert bundle in items
    assert message in items
    items.add(Bundle())
    assert len(items) == 2
    items.add(Message(foo, 'foo'))
    assert len(items) == 2
    items.add(Bundle(timetag=now()))
    assert len(items) == 3
    items.add(Message(bar, 'bar'))
    assert len(items) == 4

    bundle += Message(foo, 'foo')
    assert len(bundle) == 1
    assert bundle[0] == Message(foo, 'foo')

    # Test adding several messages at once
    bundle += [Message(bar, 'bar'), Message(baz, 'baz')]
    assert len(bundle) == 3
    assert bundle[1] == Message(bar, 'bar')
    assert bundle[2] == Message(baz, 'baz')

    # Test adding a nested bundle
    bundle += Bundle(Message(spaz, 'spaz'))
    assert len(bundle) == 4
    assert bundle[3] == Bundle(Message(spaz, 'spaz'))

    # Test __add__ instead of __iadd__
    other = Message(foo, 'foo') + Message(bar, 'bar') + Message(baz, 'baz')
    assert len(other) == 3
    assert bundle != other
    assert bundle != other + Message(spaz, 'spaz')
    assert bundle == other + Bundle(Message(spaz, 'spaz'))

    # test sorting bundles
    timetag = TimeTag(now())
    bundles = (
        Bundle(timetag=timetag + 2),
        Bundle(timetag=timetag + 1),
        Bundle(timetag=timetag + 4),
        Bundle(timetag=timetag + 3))
    bundle2, bundle1, bundle4, bundle3 = bundles

    assert sorted(bundles) == [bundle1, bundle2, bundle3, bundle4]


def test_message_arglength_mismatch():
    foo = Route('/foo', 's')
    with pytest.raises(ValueError, match=r'Argument length does not match typespec .*'):
        Message(foo, 'foo', 'bar')


@pytest.mark.asyncio
async def test_route_pattern(server):
    address = Address(url=server.url)
    foo = server.route('/aaa/foo', 's')
    bar = server.route('/bbb/foo', 's')
    sub = foo.sub() | bar.sub()
    task = create_task(subscribe(sub, 4))
    address.send(Route('//foo', 's'), ['xpath'])
    address.send(Route('/{aaa,bbb}/foo', 's'), ['array'])
    results = await task
    assert_results(results, {foo: [['xpath'], ['array']], bar: [['xpath'], ['array']]})


@pytest.mark.parametrize('path,pattern,expect_match', [
    ('/x1y', '/x{1,10,11}y', True),
    ('/x10y', '/x{1,10,11}y', True),
    ('/x11y', '/x{1,10,11}y', True),
    ('/x2y', '/x{1,10,11}y', False),
    ('/x12y', '/x{1,10,11}y', False),
    ('/x12y', '/x{11}y', False),
    ('/x12y', '/x{12}y', True),
    ('/x1,2y', r'/x{1\,2}y', True),
    ('/xy', r'/x{,}y', True),
    ('/aaa/foo', '/{aaa,bbb}/foo', True),

    # Test [] pattern
    ('/x1y', '/x[321]y', True),
    ('/x4y', '/x[321]y', False),
    ('/x2y', '/x[1-3]y', True),
    ('/x3y', '/x[2-3]y', True),
    ('/xzy', '/x[2-3z]y', True),
    ('/x1y', '/x[!a-z]y', True),
    ('/x1y', '/x[a-z]y', False),
    ('/xby', '/x[a-z]y', True),
    # spec 1.0: "A - at the end of the string has no special meaning"
    ('/x-y', '/x[23-]y', True),
    # spec 1.0: "An ! anywhere besides the first character after the
    # open bracket has no special meaning"
    ('/x!y', '/x[a-z!]y', True),

    # Test * pattern
    ('/x123y', '/x*y', True),
    ('/x123y', '/x*z', False),

    # Test ? pattern
    ('/x1y', '/x?y', True),
    ('/x?y', '/x?y', True),

    # Test ?, *, and [] with multiple /
    ('/xy/z', '/*/?', True),
    ('/xy/z', '/?/*', False),
    ('/w/xy/z', '/[xyzw]/*/?', True),

    # Test '//' from spec 1.1
    ('/z', '//z', True),
    ('/xy/z', '//z', True),
    ('/xy/z/w/u', '///w/u', True),
    ('/xy/z/w/u', '///z/w', False),

    # Now some more tests for special pattern characters
    ('/xy/z/w/u?', '///w/u[?]', True),
    ('/xy/z/w/u?', '///w/u[!?]', False),
    ('/xy/z/w/u{1,2}', r'///w/u\{1,2\}', True),
    ('/xy/z/w/u2', r'///w/u\{1,2\}', False),
    (r'/xy/z/w/u-', r'///w/u[-]', True),
    (r'/xy/z/w/u-', r'///w/u[\\-]', True),
    (r'/xy/z/w/ua-', r'///w/u[a-][-]', True),
])
def test_compile_osc_address_pattern(path, pattern, expect_match):
    regex = compile_osc_address_pattern(pattern)
    if expect_match:
        assert regex.match(path) is not None
    else:
        assert regex.match(path) is None


@pytest.mark.parametrize('char', '#{}[]!?*,-^\\'.split())
def test_cannot_serve_pattern(server, char):
    with pytest.raises(ValueError):
        server.route(char, 's')
    with pytest.raises(ValueError):
        route = Route(char, 's')
        assert route.is_pattern
        server.route(route)


@pytest.mark.asyncio
async def test_route_any_path(server):
    address = Address(url=server.url)
    any_path = server.route(ANY_PATH, 's')
    task = create_task(subscribe(any_path.sub(), 1))
    with pytest.raises(ValueError):
        address.send(any_path, ['foo'])
    address.send(Route('/foo', 's'), ['foo'])
    results = list(await task)
    assert results == [['foo']]


@pytest.mark.asyncio
async def test_route_no_args(server):
    address = Address(url=server.url)
    foo = server.route('/foo', NO_ARGS)
    task = create_task(subscribe(foo.sub(), 1))
    address.send(foo)
    results = list(await task)
    assert results == [[]]


@pytest.mark.asyncio
async def test_route_any_args(server):
    address = Address(url=server.url)
    foo = server.route('/foo', ANY_ARGS)
    task = create_task(subscribe(foo.sub(), 1))
    address.send(foo, 'foo')
    results = list(await task)
    assert results == [['foo']]


@pytest.mark.asyncio
async def test_sub_join(server):
    address = Address(url=server.url)
    foo = server.route('/foo', 's')
    bar = server.route('/bar', 's')
    baz = server.route('/baz', 's')
    foo_sub = foo.sub()
    bar_sub = bar.sub()
    baz_sub = baz.sub()
    sub = foo_sub | bar_sub
    assert foo in sub
    assert foo_sub in sub
    assert baz not in sub
    sub |= baz_sub
    sub |= sub  # no-op
    assert len(sub) == 3
    assert sub == foo_sub | bar_sub | baz_sub
    assert sub == foo.sub() | bar.sub() | baz.sub()
    assert sub != foo.sub() | bar.sub()
    assert sub not in foo.sub() | bar.sub()
    assert sub in foo.sub() | bar.sub() | baz.sub()
    assert (foo.sub() | bar.sub()) in sub

    task = create_task(subscribe(sub, 6))

    for route, data in {
        foo: ['foo1', 'foo2'],
        bar: ['bar1', 'bar2'],
        baz: ['baz1', 'baz2']
    }.items():
        for d in data:
            address.send(route, d)

    results = await task
    assert_results(results, {
        foo: [['foo1'], ['foo2']],
        bar: [['bar1'], ['bar2']],
        baz: [['baz1'], ['baz2']],
    })


def test_typespec_ops():
    typespec = TypeSpec(str)
    assert typespec == 's'
    assert typespec == (str, )
    typespec = TypeSpec(str) + TypeSpec(int)
    assert typespec == 'sh'
    assert typespec == (str, int)
    typespec += 'h'
    assert typespec == 'shh'
    assert typespec == (str, int, int)
    assert typespec == TypeSpec('shh')
    with pytest.raises(ValueError):
        typespec += ANY_ARGS
    with pytest.raises(ValueError):
        typespec += NO_ARGS
    assert ANY_ARGS + ANY_ARGS == ANY_ARGS
    assert NO_ARGS + NO_ARGS == NO_ARGS
    assert NO_ARGS == NO_ARGS
    assert NO_ARGS != ANY_ARGS
    assert ANY_ARGS != NO_ARGS
    assert typespec in ANY_ARGS
    assert typespec not in NO_ARGS
    assert NO_ARGS in NO_ARGS
    assert NO_ARGS not in typespec
    assert NO_ARGS in ANY_ARGS
    assert ANY_ARGS not in NO_ARGS
    assert ANY_ARGS in ANY_ARGS


def test_path_ops():
    foo = Path('/foo')
    assert not foo.is_pattern
    assert foo in foo
    assert foo in ANY_PATH
    assert ANY_PATH not in foo
    barfoo = Path('/bar/foo')
    pattern = Path('//foo')
    assert pattern.is_pattern
    assert barfoo in pattern
    assert foo in pattern


def test_timetag():
    # assert that constants are sane
    assert TT_IMMEDIATE == (0, 1)
    assert TimeTag(EPOCH_UTC) == (JAN_1970, 0)

    # Test TT_IMMEDIATE is frozen
    tt = TT_IMMEDIATE
    assert tt is TT_IMMEDIATE
    tt += 1
    # TimeTag does not implement __iadd__ so a new instance is created for the name tt
    assert tt is not TT_IMMEDIATE
    assert tt == (1, 1)
    assert TT_IMMEDIATE == (0, 1)

    # Test operations
    tt = TimeTag((0, 1))
    tt = tt + 3
    assert tt == (3, 1)
    tt = tt - 2.1
    assert tt == (0, .9 * FRAC_PER_SEC)

    # Test inplace operations
    tt = TimeTag((0, 1))
    tt += 3
    assert tt == (3, 1)
    tt -= 2.1
    assert tt == (0, .9 * FRAC_PER_SEC)

    for dt in (
            EPOCH_UTC,
            datetime.datetime(1905, 5, 5, 5, 5, 5, 5, datetime.timezone.utc),
            datetime.datetime(2026, 6, 6, 6, 6, 6, 6, datetime.timezone.utc),
    ):
        tt = TimeTag(dt)
        assert tt == dt and tt <= dt <= tt
        assert tt.dt == dt and tt <= dt <= tt
        assert tt.osc_timestamp == unix_timestamp_to_osc_timestamp(dt.timestamp())
        assert int(dt.timestamp()) == int(tt.unix_timestamp)
        assert pytest.approx(float(dt.timestamp()), rel=1e-6) == tt.unix_timestamp
        assert tuple(tt) == (tt.sec, tt.frac)

        for other in (
                TimeTag(tt),
                dt,
                int(tt),
                float(tt),
                tuple(tt)
        ):
            # test comparison operators
            if isinstance(other, int):
                assert int(tt) == other
                assert tt >= other
                assert other <= tt
            else:
                assert tt == other
                assert tt <= other
                assert tt >= other
                assert not tt < other
                assert not other > tt
                assert not other != tt

        # frac precision is lost when comparing to datetime or int
        assert (tt + 0.000000001).unix_timestamp == tt.unix_timestamp
        assert (tt + 0.000000001).dt == tt.dt

        # frac precision is not lost when comparing to tuple or timetag
        assert tt + 0.1 != tt
        assert tt + 0.1 != tuple(tt)


def assert_results(results_dict, expected):
    assert set(results_dict.keys()) == set(expected.keys())
    for route, results in expected.items():
        for result in results:
            assert result in results_dict[route]


async def subscribe(sub: Union[Sub, Subs], count: int):
    if isinstance(sub, Subs):
        results = {}
        async for route, item in sub:
            results.setdefault(route, []).append(item)
            if len([v for values in results.values() for v in values]) == count:
                await sub.unsub()
                break
        return results
    else:
        items = []
        async for item in sub:
            items.append(item)
            if len(items) == count:
                await sub.unsub()
                break
        return items


if __name__ == '__main__':
    pytest.main()

import array
import asyncio
import contextlib
import datetime
import functools
import random
import sys
from typing import Union

import netifaces
import pytest
import uvloop as uvloop

import conftest
import test_data

from aiolo import MultiCast, AioServer, Address, Message, PROTO_UDP, PROTO_UNIX, TimeTag, FRAC_PER_SEC, \
    NO_ARGS, Route, unix_timestamp_to_osc_timestamp, TT_IMMEDIATE, MultiCastAddress, Bundle, ANY_PATH, TypeSpec, \
    StartError, PROTO_DEFAULT, EPOCH_UTC, ANY_ARGS, JAN_1970, ThreadedServer, Midi, PROTO_TCP, INFINITY, \
    INFINITUM, TIMETAG, MIDI, NIL, FALSE, TRUE, BLOB, STRING, DOUBLE, INT64, Path, Sub, Subs, \
    compile_osc_address_pattern


def create_task(coro, cancel_timeout=conftest.CANCEL_TIMEOUT):
    loop = asyncio.get_event_loop()
    if sys.version_info[:2] >= (3, 7):
        task = asyncio.create_task(coro)
    else:
        task = loop.create_task(coro)
    if cancel_timeout >= 0:
        loop.call_later(cancel_timeout, task.cancel)
    return task


def get_ipv4(iface):
    try:
        return netifaces.ifaddresses(iface)[netifaces.AF_INET][0]['addr']
    except KeyError:
        return None


def now():
    return datetime.datetime.now(datetime.timezone.utc)


@pytest.fixture(params=[
    iface
    for iface in netifaces.interfaces()
    if get_ipv4(iface) and get_ipv4(iface) != '127.0.0.1'
])
def ip_interface(request):
    iface = request.param
    return get_ipv4(iface), iface


@pytest.fixture(params=[
    asyncio,
    uvloop
])
def event_loop(request):
    loop_mod = request.param
    loop = loop_mod.new_event_loop()
    asyncio.set_event_loop(loop)
    if loop_mod != uvloop:
        # uvloop in debug mode calls extract_stack, which results in "ValueError: call stack is not deep enough"
        # for Cython code
        loop.set_debug(True)
    with contextlib.closing(loop):
        yield loop


@pytest.fixture(params=[
    AioServer,
    ThreadedServer
])
def any_server_class(request, event_loop):
    return request.param


@pytest.fixture
async def any_server(any_server_class, unused_tcp_port_factory):
    tries = 0

    exc = None
    server = any_server_class(port=unused_tcp_port_factory())
    while tries < 3:
        try:
            server.start()
        except StartError as exc:
            await asyncio.sleep(0.1)
            server = any_server_class(port=unused_tcp_port_factory())
        else:
            break
        tries += 1
    if not server.running:
        raise StartError from exc
    yield server
    server.stop()


@pytest.fixture
async def server(event_loop, unused_tcp_port_factory):
    tries = 0
    while tries < 3:
        try:
            server = AioServer(port=unused_tcp_port_factory())
            server.start()
        except StartError:
            await asyncio.sleep(0.1)
        else:
            break
        tries += 1
    yield server
    server.stop()


@pytest.fixture(params=[
    lambda cls, port: cls(),
    lambda cls, port: cls(port=port()),
    lambda cls, port: cls(url='osc.udp://:%s' % port()),
    lambda cls, port: cls(url='osc.unix:///tmp/test-aiolo-%s.osc' % port()),
    lambda cls, port: cls(proto=PROTO_TCP, port=port()),
    lambda cls, port: cls(proto=PROTO_UDP, port=port()),
    lambda cls, port: cls(proto=PROTO_UNIX, port='/tmp/test-aiolo-%s.osc' % port()),
    lambda cls, port: cls(multicast=MultiCast('172.16.31.10', port=port())),
])
def server_init_factory(request, any_server_class, unused_tcp_port_factory):
    return functools.partial(request.param, any_server_class, unused_tcp_port_factory)


def test_server_init(server_init_factory):
    """
    Test that servers can be constructed and started with various call patterns.
    """
    server = server_init_factory()
    server.start()
    assert server.running
    assert server.port is not None
    server.stop()


def test_server_init_proto_default_udp_port(any_server_class, unused_tcp_port):
    server = any_server_class(proto=PROTO_DEFAULT, port=unused_tcp_port)
    server.start()
    assert server.running
    assert server.proto == PROTO_UDP
    assert server.port is not None
    server.stop()


def test_server_init_proto_default_unix_port(any_server_class, unused_tcp_port):
    server = any_server_class(proto=PROTO_DEFAULT, port='/tmp/test-aiolo-%s.osc' % unused_tcp_port)
    server.start()
    assert server.running
    assert server.proto == PROTO_UNIX
    assert server.port is not None
    server.stop()


@pytest.fixture(params=[
    lambda cls, port: cls(port='/foo'),
    lambda cls, port: cls(port=-1),
    lambda cls, port: cls(url='osc.foo://:%s' % port()),
])
def server_start_error_factory(request, any_server_class, unused_tcp_port_factory):
    return functools.partial(request.param, any_server_class, unused_tcp_port_factory)


def test_server_start_error(server_start_error_factory):
    server = server_start_error_factory()
    with pytest.raises(StartError):
        server.start()


def server_init_error_multicast_factory(any_server_class, unused_tcp_port_factory):
    port = unused_tcp_port_factory()
    return any_server_class(
        url='osc.tcp://:%s' % port,
        multicast=MultiCast('172.16.31.10', port=port()))


@pytest.fixture(params=[
    lambda cls, port: cls(multicast='foo'),
    lambda cls, port: cls(proto=-1, port=port()),
    lambda cls, port: cls(proto='foo', port=port()),
    lambda cls, port: cls(url='osc.tcp://:%s' % port(), port='10'),
    lambda cls, port: cls(url='osc.tcp://:%s' % port(), proto=PROTO_UDP),
    server_init_error_multicast_factory,
    lambda cls, port: cls(proto=PROTO_TCP, multicast=MultiCast('172.16.31.10', port=port())),
    lambda cls, port: cls(port='10', multicast=MultiCast('172.16.31.10', port=port())),
])
def server_init_error_factory(request, any_server_class, unused_tcp_port_factory):
    return functools.partial(request.param, any_server_class, unused_tcp_port_factory)


def test_server_init_error(server_init_error_factory):
    with pytest.raises((ValueError, TypeError)):
        server_init_error_factory()


@pytest.mark.asyncio
async def test_subs_unsub(event_loop):
    foo = Sub(Route('/foo', ANY_ARGS))
    bar = Sub(Route('/bar', ANY_ARGS))
    subs = foo | bar
    task = create_task(subscribe(subs, 3))
    await foo.pub('1')
    await bar.pub('2')
    await bar.unsub()
    await foo.pub('3')
    results = await task
    assert_results(results, {bar.route: ['2'], foo.route: ['1', '3']})


@pytest.mark.asyncio
async def test_multiple_addresses(any_server):
    """
    Test that multiple clients can send data to a single server
    """
    foo = any_server.route('/foo', int)
    addresses = [Address(port=any_server.port) for i in range(3)]
    task = create_task(subscribe(foo.sub(), 3))
    for i, address in enumerate(addresses):
        address.send(foo, i)
        # await asyncio.sleep(0.1)
    results = await task
    assert results == [[0], [1], [2]]


@pytest.fixture(params=[
    lambda port, ip, iface: (
        Address(url='osc.tcp://%s:%s' % (ip, port())),
        ip, iface),
    lambda port, ip, iface: (
        Address(url='osc.udp://%s:%s' % (ip, port())),
        ip, iface),
    lambda port, ip, iface: (
        Address(url='osc.unix:///aiolo-test-%s.osc' % port()),
        ip, iface),
    lambda port, ip, iface: (
        Address(proto=PROTO_TCP, host=ip, port=port()),
        ip, iface),
    lambda port, ip, iface: (
        Address(proto=PROTO_UDP, host=ip, port=port()),
        ip, iface),
    lambda port, ip, iface: (
        Address(proto=PROTO_UNIX, host=ip, port='/tmp/aiolo-test-%s.osc' % port()),
        ip, iface),
])
def address_init_factory(request, ip_interface, unused_tcp_port_factory):
    ip, iface = ip_interface
    return functools.partial(request.param, unused_tcp_port_factory, ip, iface)


def test_address_init(address_init_factory):
    """
    Test Address initialization and state.
    """
    address, ip, iface = address_init_factory()

    if iface is None or address.proto in (PROTO_UNIX, PROTO_DEFAULT):
        with pytest.raises(ValueError):
            address.interface = iface
    else:
        address.interface = iface
        assert address.interface == iface

    address, ip, iface = address_init_factory()
    if not ip or address.proto == PROTO_UNIX:
        with pytest.raises(ValueError):
            address.set_ip(ip)
    else:
        assert address.interface is None
        address.set_ip(ip)
        assert address.interface == iface

    address, ip, iface = address_init_factory()
    with pytest.raises(ValueError):
        address.interface = 'foobar0'
    with pytest.raises(ValueError):
        address.set_ip('foo.bar')
    with pytest.raises(ValueError):
        address.set_ip('172.16.31.10')


@pytest.mark.asyncio
async def test_multicast(any_server_class, unused_tcp_port):
    """
    Test multicast send/receive.
    """
    cluster = []
    multicast = MultiCast('172.16.31.10', port=unused_tcp_port)
    for i in range(3):
        server = any_server_class(multicast=multicast)
        server.start()
        cluster.append(server)

    address = MultiCastAddress(server=random.choice(cluster))
    foo = Route('/foo', str)

    for s in cluster:
        s.route(foo)

    task = create_task(
        subscribe(foo.sub(), 3 * len(cluster)),
        cancel_timeout=conftest.CANCEL_TIMEOUT * len(cluster))

    for d in ['foo', 'bar', 'baz']:
        address.send(foo, d)
        # await asyncio.sleep(1)

    results = await task

    assert results.count(['foo']) == len(cluster)
    assert results.count(['bar']) == len(cluster)
    assert results.count(['baz']) == len(cluster)
    for s in cluster:
        s.stop()


@pytest.mark.asyncio
async def test_multiple_servers(any_server_class, unused_tcp_port_factory):
    """
    Test that multiple servers can run okay in the same process.
    """
    servers = []
    addresses = []
    foo = Route('/foo', int)

    for i in range(3):
        server = any_server_class(url='osc.tcp://:%s' % unused_tcp_port_factory())
        server.route(foo)
        server.start()
        servers.append(server)
        address = Address(url=server.url)
        addresses.append(address)

    task = create_task(subscribe(foo.sub(), 3))
    for i, address in enumerate(addresses):
        address.send(foo, i)

    try:
        results = await task
        assert set([r[0] for r in results]) == {0, 1, 2}
    finally:
        for server in servers:
            server.stop()


def unpack_message_data():
    for t in test_data.ARGDEF_TEST_DATA:
        for typespec in t.typespecs:
            for value, expected in t.valid:
                if isinstance(typespec, int) and not isinstance(typespec, bool):
                    # Test unicode typespec like 'h'
                    yield t.path, chr(typespec), value, [expected]
                yield t.path, typespec, value, [expected]
                yield t.path, [typespec, typespec], [value, value], [expected, expected]


@pytest.mark.parametrize('path, typespec, data, expected', unpack_message_data())
def test_unpack_message(path, typespec, data, expected):
    """
    Test that types get parsed correctly
    """
    route = Route(path, typespec)
    message = Message(route, data)
    assert message.unpack() == expected

def unpack_message_type_error_data():
    for typespec_test_data in test_data.ARGDEF_TEST_DATA:
        for typespec in typespec_test_data.typespecs:
            for value in typespec_test_data.type_error:
                yield typespec_test_data.path, typespec, [value]


@pytest.mark.parametrize('path, typespec, data', unpack_message_type_error_data())
def test_unpack_message_type_error(path, typespec, data):
    """
    Test that TypeError is raised for invalid data types
    """
    route = Route(path, typespec)
    with pytest.raises(TypeError):
        message = Message(route, data)
        message.unpack()


def unpack_message_overflow_error_data():
    for t in test_data.ARGDEF_TEST_DATA:
        for typespec in t.typespecs:
            for value in t.overflow_error:
                yield t.path, typespec, [value]


@pytest.mark.parametrize('path, typespec, data', unpack_message_overflow_error_data())
def test_unpack_message_overflow_error(path, typespec, data):
    """
    Test that OverflowError is raised for invalid data types
    """
    route = Route(path, typespec)
    with pytest.raises(OverflowError):
        message = Message(route, data)
        message.unpack()


def unpack_message_value_error_data():
    for t in test_data.ARGDEF_TEST_DATA:
        for typespec in t.typespecs:
            for value in t.value_error:
                yield t.path, typespec, [value]


@pytest.mark.parametrize('path, typespec, data', unpack_message_value_error_data())
def test_unpack_message_value_error(path, typespec, data):
    """
    Test that ValueError is raised for invalid data values
    """
    route = Route(path, typespec)
    with pytest.raises(ValueError):
        message = Message(route, data)
        message.unpack()


@pytest.mark.parametrize('value, typespec', [
    [1, INT64],
    [1.0, DOUBLE],
    [0, INT64],
    [0.0, DOUBLE],
    [42, INT64],
    [42.0, DOUBLE],
    ['foo', STRING],
    [b'foo', BLOB],
    [array.array('b', b'foo'), BLOB],
    [True, TRUE],
    [False, FALSE],
    [None, NIL],
    [Midi(1, 2, 3, 4), MIDI],
    [now(), TIMETAG],
    [TimeTag(), TIMETAG],
    [INFINITY, INFINITUM],
])
def test_typespec_guess(value, typespec):
    assert TypeSpec.guess([value]) == TypeSpec(typespec)


@pytest.mark.parametrize('value', [
    {},
    [],
])
def test_typespec_guess_type_error(value):
    with pytest.raises(TypeError):
        TypeSpec.guess([value])


@pytest.mark.asyncio
async def test_reuse_subs(any_server):
    address = Address(url=any_server.url)
    foo = any_server.route('/foo', 's')
    tasks = asyncio.gather(*[
        create_task(subscribe(foo.sub(), 1))
        for i in range(3)
    ])
    address.send(foo, 'bar')
    with pytest.raises(asyncio.CancelledError):
        await tasks


@pytest.mark.asyncio
async def test_unroute(server):
    address = Address(url=server.url)
    foo = server.route('/foo', 's')
    task = create_task(subscribe(foo.sub(), 1))
    server.unroute(foo)
    address.send(foo, 'bar')
    with pytest.raises(asyncio.CancelledError):
        await task


@pytest.mark.asyncio
async def test_bundle(server):
    address = Address(url=server.url)
    foo = server.route('/foo', 's')
    bar = server.route('/bar', 's')
    baz = server.route('/baz', 's')
    subs = foo.sub() | bar.sub() | baz.sub()
    task = create_task(subscribe(subs, 3))
    address.bundle([
        Message(foo, 'foo'),
        Message(bar, 'bar'),
        Message(baz, 'baz'),
    ])
    results = await task
    assert_results(results, {foo: [['foo']], bar: [['bar']], baz: [['baz']]})


@pytest.mark.asyncio
async def test_bundle_delayed(server):
    address = Address(url=server.url)
    foo = server.route('/foo', 's')
    task = create_task(subscribe(foo.sub(), 1))
    address.bundle([
        Message(foo, 'now'),
        Bundle(
            [Message(foo, 'later')],
            timetag=now() + datetime.timedelta(seconds=0.5))
    ])
    address.delay(1, foo, 'later than later')
    results = list(await task)
    assert results == [['now']]
    assert server.events_pending
    assert 0 < server.next_event_delay < 1
    task = create_task(subscribe(foo.sub(), 1))
    results = list(await task)
    assert results == [['later']]
    assert server.events_pending
    assert 0 < server.next_event_delay < 1
    task = create_task(subscribe(foo.sub(), 1))
    results = list(await task)
    assert results == [['later than later']]
    assert not server.events_pending


def test_bundle_and_message_ops():
    foo = Route('/foo', 's')
    bar = Route('/bar', 's')
    baz = Route('/baz', 's')
    spaz = Route('/spaz', 's')
    bundle = Bundle()
    assert len(bundle) == 0
    assert bundle == Bundle()

    with pytest.raises(IndexError):
        assert bundle[0]

    message = Message(foo, 'foo')
    assert message == Message(foo, 'foo')
    assert bundle != Bundle(message)

    # Test hashable
    items = {bundle, message}
    assert len(items) == 2
    assert bundle in items
    assert message in items
    items.add(Bundle())
    assert len(items) == 2
    items.add(Message(foo, 'foo'))
    assert len(items) == 2
    items.add(Bundle(timetag=now()))
    assert len(items) == 3
    items.add(Message(bar, 'bar'))
    assert len(items) == 4

    bundle += Message(foo, 'foo')
    assert len(bundle) == 1
    assert bundle[0] == Message(foo, 'foo')

    # Test adding several messages at once
    bundle += [Message(bar, 'bar'), Message(baz, 'baz')]
    assert len(bundle) == 3
    assert bundle[1] == Message(bar, 'bar')
    assert bundle[2] == Message(baz, 'baz')

    # Test adding a nested bundle
    bundle += Bundle(Message(spaz, 'spaz'))
    assert len(bundle) == 4
    assert bundle[3] == Bundle(Message(spaz, 'spaz'))

    # Test __add__ instead of __iadd__
    other = Message(foo, 'foo') + Message(bar, 'bar') + Message(baz, 'baz')
    assert len(other) == 3
    assert bundle != other
    assert bundle != other + Message(spaz, 'spaz')
    assert bundle == other + Bundle(Message(spaz, 'spaz'))

    # test sorting bundles
    timetag = TimeTag(now())
    bundles = (
        Bundle(timetag=timetag + 2),
        Bundle(timetag=timetag + 1),
        Bundle(timetag=timetag + 4),
        Bundle(timetag=timetag + 3))
    bundle2, bundle1, bundle4, bundle3 = bundles

    assert sorted(bundles) == [bundle1, bundle2, bundle3, bundle4]


def test_message_arglength_mismatch():
    foo = Route('/foo', 's')
    with pytest.raises(ValueError, match=r'Argument length does not match typespec .*'):
        Message(foo, 'foo', 'bar')


@pytest.mark.asyncio
async def test_route_pattern(server):
    address = Address(url=server.url)
    foo = server.route('/aaa/foo', 's')
    bar = server.route('/bbb/foo', 's')
    sub = foo.sub() | bar.sub()
    task = create_task(subscribe(sub, 4))
    address.send(Route('//foo', 's'), ['xpath'])
    address.send(Route('/{aaa,bbb}/foo', 's'), ['array'])
    results = await task
    assert_results(results, {foo: [['xpath'], ['array']], bar: [['xpath'], ['array']]})


@pytest.mark.parametrize('path,pattern,expect_match', [
    ('/x1y', '/x{1,10,11}y', True),
    ('/x10y', '/x{1,10,11}y', True),
    ('/x11y', '/x{1,10,11}y', True),
    ('/x2y', '/x{1,10,11}y', False),
    ('/x12y', '/x{1,10,11}y', False),
    ('/x12y', '/x{11}y', False),
    ('/x12y', '/x{12}y', True),
    ('/x1,2y', r'/x{1\,2}y', True),
    ('/xy', r'/x{,}y', True),
    ('/aaa/foo', '/{aaa,bbb}/foo', True),

    # Test [] pattern
    ('/x1y', '/x[321]y', True),
    ('/x4y', '/x[321]y', False),
    ('/x2y', '/x[1-3]y', True),
    ('/x3y', '/x[2-3]y', True),
    ('/xzy', '/x[2-3z]y', True),
    ('/x1y', '/x[!a-z]y', True),
    ('/x1y', '/x[a-z]y', False),
    ('/xby', '/x[a-z]y', True),
    # spec 1.0: "A - at the end of the string has no special meaning"
    ('/x-y', '/x[23-]y', True),
    # spec 1.0: "An ! anywhere besides the first character after the
    # open bracket has no special meaning"
    ('/x!y', '/x[a-z!]y', True),

    # Test * pattern
    ('/x123y', '/x*y', True),
    ('/x123y', '/x*z', False),

    # Test ? pattern
    ('/x1y', '/x?y', True),
    ('/x?y', '/x?y', True),

    # Test ?, *, and [] with multiple /
    ('/xy/z', '/*/?', True),
    ('/xy/z', '/?/*', False),
    ('/w/xy/z', '/[xyzw]/*/?', True),

    # Test '//' from spec 1.1
    ('/z', '//z', True),
    ('/xy/z', '//z', True),
    ('/xy/z/w/u', '///w/u', True),
    ('/xy/z/w/u', '///z/w', False),

    # Now some more tests for special pattern characters
    ('/xy/z/w/u?', '///w/u[?]', True),
    ('/xy/z/w/u?', '///w/u[!?]', False),
    ('/xy/z/w/u{1,2}', r'///w/u\{1,2\}', True),
    ('/xy/z/w/u2', r'///w/u\{1,2\}', False),
    (r'/xy/z/w/u-', r'///w/u[-]', True),
    (r'/xy/z/w/u-', r'///w/u[\\-]', True),
    (r'/xy/z/w/ua-', r'///w/u[a-][-]', True),
])
def test_compile_osc_address_pattern(path, pattern, expect_match):
    regex = compile_osc_address_pattern(pattern)
    if expect_match:
        assert regex.match(path) is not None
    else:
        assert regex.match(path) is None


@pytest.mark.parametrize('char', '#{}[]!?*,-^\\'.split())
def test_cannot_serve_pattern(server, char):
    with pytest.raises(ValueError):
        server.route(char, 's')
    with pytest.raises(ValueError):
        route = Route(char, 's')
        assert route.is_pattern
        server.route(route)


@pytest.mark.asyncio
async def test_route_any_path(server):
    address = Address(url=server.url)
    any_path = server.route(ANY_PATH, 's')
    task = create_task(subscribe(any_path.sub(), 1))
    with pytest.raises(ValueError):
        address.send(any_path, ['foo'])
    address.send(Route('/foo', 's'), ['foo'])
    results = list(await task)
    assert results == [['foo']]


@pytest.mark.asyncio
async def test_route_no_args(server):
    address = Address(url=server.url)
    foo = server.route('/foo', NO_ARGS)
    task = create_task(subscribe(foo.sub(), 1))
    address.send(foo)
    results = list(await task)
    assert results == [[]]


@pytest.mark.asyncio
async def test_route_any_args(server):
    address = Address(url=server.url)
    foo = server.route('/foo', ANY_ARGS)
    task = create_task(subscribe(foo.sub(), 1))
    address.send(foo, 'foo')
    results = list(await task)
    assert results == [['foo']]


@pytest.mark.asyncio
async def test_sub_join(server):
    address = Address(url=server.url)
    foo = server.route('/foo', 's')
    bar = server.route('/bar', 's')
    baz = server.route('/baz', 's')
    foo_sub = foo.sub()
    bar_sub = bar.sub()
    baz_sub = baz.sub()
    sub = foo_sub | bar_sub
    assert foo in sub
    assert foo_sub in sub
    assert baz not in sub
    sub |= baz_sub
    sub |= sub  # no-op
    assert len(sub) == 3
    assert sub == foo_sub | bar_sub | baz_sub
    assert sub == foo.sub() | bar.sub() | baz.sub()
    assert sub != foo.sub() | bar.sub()
    assert sub not in foo.sub() | bar.sub()
    assert sub in foo.sub() | bar.sub() | baz.sub()
    assert (foo.sub() | bar.sub()) in sub

    task = create_task(subscribe(sub, 6))

    for route, data in {
        foo: ['foo1', 'foo2'],
        bar: ['bar1', 'bar2'],
        baz: ['baz1', 'baz2']
    }.items():
        for d in data:
            address.send(route, d)

    results = await task
    assert_results(results, {
        foo: [['foo1'], ['foo2']],
        bar: [['bar1'], ['bar2']],
        baz: [['baz1'], ['baz2']],
    })


def test_typespec_ops():
    typespec = TypeSpec(str)
    assert typespec == 's'
    assert typespec == (str, )
    typespec = TypeSpec(str) + TypeSpec(int)
    assert typespec == 'sh'
    assert typespec == (str, int)
    typespec += 'h'
    assert typespec == 'shh'
    assert typespec == (str, int, int)
    assert typespec == TypeSpec('shh')
    with pytest.raises(ValueError):
        typespec += ANY_ARGS
    with pytest.raises(ValueError):
        typespec += NO_ARGS
    assert ANY_ARGS + ANY_ARGS == ANY_ARGS
    assert NO_ARGS + NO_ARGS == NO_ARGS
    assert NO_ARGS == NO_ARGS
    assert NO_ARGS != ANY_ARGS
    assert ANY_ARGS != NO_ARGS
    assert typespec in ANY_ARGS
    assert typespec not in NO_ARGS
    assert NO_ARGS in NO_ARGS
    assert NO_ARGS not in typespec
    assert NO_ARGS in ANY_ARGS
    assert ANY_ARGS not in NO_ARGS
    assert ANY_ARGS in ANY_ARGS


def test_path_ops():
    foo = Path('/foo')
    assert not foo.is_pattern
    assert foo in foo
    assert foo in ANY_PATH
    assert ANY_PATH not in foo
    barfoo = Path('/bar/foo')
    pattern = Path('//foo')
    assert pattern.is_pattern
    assert barfoo in pattern
    assert foo in pattern


def test_timetag():
    # assert that constants are sane
    assert TT_IMMEDIATE == (0, 1)
    assert TimeTag(EPOCH_UTC) == (JAN_1970, 0)

    # Test TT_IMMEDIATE is frozen
    tt = TT_IMMEDIATE
    assert tt is TT_IMMEDIATE
    tt += 1
    # TimeTag does not implement __iadd__ so a new instance is created for the name tt
    assert tt is not TT_IMMEDIATE
    assert tt == (1, 1)
    assert TT_IMMEDIATE == (0, 1)

    # Test operations
    tt = TimeTag((0, 1))
    tt = tt + 3
    assert tt == (3, 1)
    tt = tt - 2.1
    assert tt == (0, .9 * FRAC_PER_SEC)

    # Test inplace operations
    tt = TimeTag((0, 1))
    tt += 3
    assert tt == (3, 1)
    tt -= 2.1
    assert tt == (0, .9 * FRAC_PER_SEC)

    for dt in (
            EPOCH_UTC,
            datetime.datetime(1905, 5, 5, 5, 5, 5, 5, datetime.timezone.utc),
            datetime.datetime(2026, 6, 6, 6, 6, 6, 6, datetime.timezone.utc),
    ):
        tt = TimeTag(dt)
        assert tt == dt and tt <= dt <= tt
        assert tt.dt == dt and tt <= dt <= tt
        assert tt.osc_timestamp == unix_timestamp_to_osc_timestamp(dt.timestamp())
        assert int(dt.timestamp()) == int(tt.unix_timestamp)
        assert pytest.approx(float(dt.timestamp()), rel=1e-6) == tt.unix_timestamp
        assert tuple(tt) == (tt.sec, tt.frac)

        for other in (
                TimeTag(tt),
                dt,
                int(tt),
                float(tt),
                tuple(tt)
        ):
            # test comparison operators
            if isinstance(other, int):
                assert int(tt) == other
                assert tt >= other
                assert other <= tt
            else:
                assert tt == other
                assert tt <= other
                assert tt >= other
                assert not tt < other
                assert not other > tt
                assert not other != tt

        # frac precision is lost when comparing to datetime or int
        assert (tt + 0.000000001).unix_timestamp == tt.unix_timestamp
        assert (tt + 0.000000001).dt == tt.dt

        # frac precision is not lost when comparing to tuple or timetag
        assert tt + 0.1 != tt
        assert tt + 0.1 != tuple(tt)


def assert_results(results_dict, expected):
    assert set(results_dict.keys()) == set(expected.keys())
    for route, results in expected.items():
        for result in results:
            assert result in results_dict[route]


async def subscribe(sub: Union[Sub, Subs], count: int):
    if isinstance(sub, Subs):
        results = {}
        async for route, item in sub:
            results.setdefault(route, []).append(item)
            if len([v for values in results.values() for v in values]) == count:
                await sub.unsub()
                break
        return results
    else:
        items = []
        async for item in sub:
            items.append(item)
            if len(items) == count:
                await sub.unsub()
                break
        return items


if __name__ == '__main__':
    pytest.main()

# uvloop in debug mode calls extract_stack, which results in "ValueError: call stack is not deep enough" # for Cython code Test that servers can be constructed and started with various call patterns. Test that multiple clients can send data to a single server # await asyncio.sleep(0.1) Test Address initialization and state. Test multicast send/receive. # await asyncio.sleep(1) Test that multiple servers can run okay in the same process. # Test unicode typespec like 'h' Test that types get parsed correctly Test that TypeError is raised for invalid data types Test that OverflowError is raised for invalid data types Test that ValueError is raised for invalid data values # Test hashable # Test adding several messages at once # Test adding a nested bundle # Test __add__ instead of __iadd__ # test sorting bundles # Test [] pattern # spec 1.0: "A - at the end of the string has no special meaning" # spec 1.0: "An ! anywhere besides the first character after the # open bracket has no special meaning" # Test * pattern # Test ? pattern # Test ?, *, and [] with multiple / # Test '//' from spec 1.1 # Now some more tests for special pattern characters # no-op # assert that constants are sane # Test TT_IMMEDIATE is frozen # TimeTag does not implement __iadd__ so a new instance is created for the name tt # Test operations # Test inplace operations # test comparison operators # frac precision is lost when comparing to datetime or int # frac precision is not lost when comparing to tuple or timetag