diff --git a/alsa_midi/client.py b/alsa_midi/client.py index 030559d..fa77f01 100644 --- a/alsa_midi/client.py +++ b/alsa_midi/client.py @@ -404,6 +404,7 @@ def close(self): self._handle_p = None # type: ignore self.handle = None # type: ignore self._event_parser = None + self._fd = -1 def _get_fds(self): """Get the file descriptor number for the client connection into :data:`self._fd`.""" @@ -430,6 +431,21 @@ def _get_event_parser(self): self._event_parser = parser return parser + def fileno(self): + """Return file descriptor assigned to this connection. + + This file descriptor can be used with select, poll, or other event + loop mechanisms for monitoring ALSA sequencer events. + + Wraps :alsa:`snd_seq_poll_descriptors`. + + :return: file descriptor number + :raises OSError: if the connection is closed or file descriptor is not available + """ + if self._fd >= 0: + return self._fd + raise OSError("File descriptor not available") + def get_sequencer_name(self) -> str: """Get sequencer name. @@ -1232,6 +1248,202 @@ def list_ports(self, *, return result + @overload + def resolve_address(self, arg1: str) -> 'Address': + ... + + @overload + def resolve_address(self, arg1: str, arg2: str = '') -> 'Address': + ... + + @overload + def resolve_address(self, arg1: int, arg2: int = 0) -> 'Address': + ... + + @overload + def resolve_address(self, arg1: tuple) -> 'Address': + ... + + @overload + def resolve_address(self, arg1: AddressType) -> 'Address': + ... + + def resolve_address(self, + arg1: Union[int, str, tuple, AddressType], + arg2: Union[int, str, None] = None) -> Address: + """Parse client:port address string + + Similar to Address() but parses client and port by names + in addition numeric to ids. + + Raises exception if client:port does not exist. + + Note finds addresses in clients other than self (as per + ALSA semantics). + + As per alsa-util library's snd_seq_parse_address(): + - Accepts "client:port" and "client.port" interchangeably + - Returns client_id:0 if found for input without ":" or '.' + - Chooses last of possible multiple prefix-matches (if any) with + existing client names unless superceded by full exact match. + - Will use first of possible multiple exact matching client + names (yes, ALSA allows this). + - Likewise chooses last of possible multiple prefix-match + port names in chosen client unless superceded by exact + match, or first of possible multiple exact matches (ALSA + allows this, too, even if all are of the same READ, WRITE, + READ+WRITE gender). + - Does nothing fancy such as searching for the best port name + match within multiple candidate client matches. + - Raises TypeError for incorrect arguments, FileNotFoundError + if specified client:port doesn't exist. + + :param spec: address specification, 'client', 'client:port', 'client.port', etc. + :return: Address + + :exceptions: FileNotFoundError, TypeError + """ + + def type_error(): + raise TypeError("resolve_address(%s, %s) is not a single string " + "argument of form 'client:port' (or " + "'client.port') with string client:port names " + "or numeric IDs, or a single Address argument, " + "or two string or int client,port arguments " + "(no ':' 'or '.'), or a single 2-tuple argument " + "of same." + % (arg1, arg2)) + + def strip_quotes(arg): + # remove enclosing single or double quotes if any + if arg.startswith("'"): + arg = arg.strip("'") + elif arg.startswith('"'): + arg = arg.strip('"') + return arg + + # Validate arguments ... + args_parsed = False + + if isinstance(arg1, Address): + if arg2 is not None: + type_error() + client_name = arg1.client_id # will be tested and converted + port_name = arg1.port_id # to int below + args_parsed = True + elif isinstance(arg1, tuple): + if arg2 is not None: + type_error() + if len(arg1) == 2: + # treat as separate args so don't have to duplicate + # parsing below + (arg1, arg2) = arg1 + else: + type_error() + + if not args_parsed: + if isinstance(arg2, str): + if ':' in arg2 or '.' in arg2: + type_error() + arg2 = strip_quotes(arg2) + if isinstance(arg1, str): + arg1 = strip_quotes(arg1) + if ':' in arg1 or '.' in arg1: + if arg2 is not None: + type_error() + # normalize "client.port" to "client:port" + if '.' in arg1: + arg1 = arg1.replace('.', ':') + # split client and port + client_port = arg1.split(':') + if len(client_port) == 2: + (client_name, port_name) = client_port + else: + client_name = arg1 + port_name = None + else: + client_name = arg1 + port_name = arg2 # OK if None + else: + client_name = arg1 + port_name = arg2 # OK if None + + if not (isinstance(client_name, (int, str)) + and isinstance(port_name, (int, str, type(None)))): + type_error() + + # Arguments validated. Continue. + + ports = self.list_ports(include_system=True) + + # confirm client exists + client_id = None + if isinstance(client_name, int): + client_id = client_name + elif isinstance(client_name, str) and client_name.isdecimal(): + client_id = int(client_name) + else: + client_id = None + + best_client = None + if client_id is None: + for port_info in ports: + if port_info.client_name == client_name: + # exact match supercedes prefix matches + # but note can have multiple clients with same name + client_id = port_info.client_id + best_client = port_info + break + elif port_info.client_name.startswith(client_name): + # last of multiples will overwrite + client_id = port_info.client_id + best_client = port_info + else: + for port_info in ports: + if port_info.client_id == client_id: + best_client = port_info + break # can't have multiple clients with same numeric id + + # only return client if exists + if client_id is None: + raise FileNotFoundError("No ALSA client '%s'" % client_name) + elif best_client is None: + raise FileNotFoundError("No ALSA client %d" % client_id) + + # find specified port of specified client + if isinstance(port_name, int): + port_id = port_name + elif isinstance(port_name, str) and port_name.isdecimal(): + port_id = int(port_name) + elif port_name is None: + port_id = 0 # but still must check that exists + else: + port_id = None # find by name + + best_port = None + for port_info in ports: + if port_info.client_id == best_client.client_id: + if port_info.port_id == port_id: + # port id match supercedes name/prefix match + best_port = port_info + break + elif port_info.name == port_name: + # exact match supercedes prefix matches + port_id = port_info.port_id + best_port = port_info + break + elif isinstance(port_name, str) \ + and port_name in port_info.name: + # last of multiples will overwrite + port_id = port_info.port_id + best_port = port_info + + if best_port is None: + raise FileNotFoundError("No ALSA client:port %d:%s" + % (client_id, port_name)) + else: + return Address(client_id, best_port.port_id) + def _subunsub_port(self, func, sender: AddressType, dest: AddressType, *, queue: Optional[Union[Queue, int]] = None, diff --git a/examples/connect.py b/examples/connect.py index 1f52c63..e5cf4eb 100755 --- a/examples/connect.py +++ b/examples/connect.py @@ -2,7 +2,7 @@ from argparse import ArgumentParser -from alsa_midi import Address, SequencerClient +from alsa_midi import SequencerClient def main(): @@ -15,19 +15,22 @@ def main(): help="Use real-time-stamp") parser.add_argument("--tick", "-t", action="store_false", dest="real", help="Use tick-time-stamp") - parser.add_argument("sender", type=Address, metavar="CLIENT:PORT", + parser.add_argument("sender", metavar="CLIENT:PORT", help="Source side of the connection") - parser.add_argument("dest", type=Address, metavar="CLIENT:PORT", + parser.add_argument("dest", metavar="CLIENT:PORT", help="Sink side of the connection") args = parser.parse_args() client = SequencerClient("dump_events.py") + sender = client.resolve_address(args.sender) + dest = client.resolve_address(args.dest) + if args.disconnect: - client.unsubscribe_port(args.sender, args.dest) + client.unsubscribe_port(sender, dest) else: - client.subscribe_port(args.sender, args.dest, + client.subscribe_port(sender, dest, exclusive=args.exclusive, time_update=(args.real is not None), time_real=args.real) diff --git a/tests/test_sequencer_address.py b/tests/test_sequencer_address.py new file mode 100644 index 0000000..994f963 --- /dev/null +++ b/tests/test_sequencer_address.py @@ -0,0 +1,497 @@ +"""Tests SequencerClientBase.resolve_address() + +Tests many permutations of "numeric-or-named-client:numeric-or-named-port", +"client:port", "client.port", "'client:port'", (client, port), ((client, port)), +etc. + +Runs silently by default, returns 0 on all tests successful or raised +expected exception. Otherwise exits with uncaught exception. + +-v/--verbose option to output results and fail-with-expected-errors +-p/--pause to pause until killed (to use `aconnect -oil` to view test + client/ports + +Outputs test/pass/fail counts if either -v or -p +""" + + +import argparse +import sys +import time + +from collections import namedtuple # isort: skip +import pytest # isort: skip + +import alsa_midi # isort: skip +from alsa_midi import Address # isort: skip + + +verbose = False + +# Have to store ports somewhere else go out of scope +# and get deleted on return from create_ports() +clients_ports = {} + + +test_count = 0 # total number of permutations tested +pass_count = 0 # successful tests +fail_count = 0 # test which failed with expected error + + +# clients/ports for tests +# + +# port name, requested port id, actual port id, test substrings, capabilities +port_spec = namedtuple('port_spec', "name req_id id substr caps") + +# mixture of capabilities (not used, but to make sure don't affect results) +INPUT_CAPS = alsa_midi.WRITE_PORT +OUTPUT_CAPS = alsa_midi.READ_PORT +IN_OUT_CAPS = INPUT_CAPS | OUTPUT_CAPS + +# all tests use this for client.resolve_address(...) +tester_ports = ( + port_spec("zero", -1, 0, False, IN_OUT_CAPS), +) + +# default sequentially-numbered ports +# tests port can be accessed by substring of name +sequential_ports = ( + port_spec("zero", -1, 0, False, INPUT_CAPS), + port_spec("One", -1, 1, False, OUTPUT_CAPS), + port_spec("two", -1, 2, False, IN_OUT_CAPS), + port_spec("three", -1, 3, True, OUTPUT_CAPS), # test by substring + port_spec("FOUR", -1, 4, False, INPUT_CAPS), + port_spec("five", -1, 5, False, IN_OUT_CAPS), +) + +# find above ports by substrings of names +prefix_ports = ( + port_spec("er", -1, 0, False, IN_OUT_CAPS), + port_spec("On", -1, 1, False, INPUT_CAPS), + port_spec("wo", -1, 2, False, OUTPUT_CAPS), +) + +# explicit non-default, non-sequential port numbers +# tests exact match of port name overrides substring match +explicit_ports = ( + port_spec("THREE", 3, 3, False, OUTPUT_CAPS), + port_spec("six", 6, 6, False, INPUT_CAPS), + port_spec("Eight", 8, 8, False, IN_OUT_CAPS), + port_spec("ele", 12, 12, False, INPUT_CAPS), # exact match + port_spec("lev", 13, 13, False, INPUT_CAPS), # " " + port_spec("eleven", 11, 11, False, INPUT_CAPS), + port_spec("eve", 14, 14, False, INPUT_CAPS), # exact match + port_spec("ven", 15, 15, False, INPUT_CAPS), # " " + port_spec("twenTY", 20, 20, False, OUTPUT_CAPS), +) + + +# at program exit if -v/--verbose, -p/--pause, or uncaught exception +def print_stats(stream=sys.stdout): + global test_count, pass_count, fail_count # noqa: F824 + stream.write("%d tests\n" + "%d passed\n" + "%d failed with expected errors \n" + % (test_count, + pass_count, + fail_count)) + + +# create clients/ports ports from above tuples of port_specs +def create_ports(client, port_descs): + global clients_ports # noqa: F824 + + client_name = client.get_client_info().name + + for port_desc in port_descs: + if port_desc.req_id == -1: + port = client.create_port(name=port_desc.name, + caps=port_desc.caps) + else: + port = client.create_port(name=port_desc.name, + caps=port_desc.caps, + port_id=port_desc.req_id) + clients_ports.setdefault(client_name, []).append(port) + + +# Parse address of various forms ('client:port', (client, port), +# ((client, port),), etc. checking for success or expected error +# exception. +def address_parse(*args): + # one or two address arguments, then expected client_id, port_id, + # and expected-error-or-None + + assert len(args) > 3, \ + "internal test program bug: address_parse(%s) <= 3\n" % (args,) + + expected_error = args[-1] + port_id = args[-2] + client_id = args[-3] + address_args = args[:-3] + + global test_count, pass_count, fail_count + + # input of ((a,b),c,d,e) -> ((a,b),) not (a,b) + if len(address_args) == 1: + address_args = address_args[0] + is_tuple = False # for arg_name() + else: + is_tuple = True # for arg_name() + + # format address argument(s) into more readable form + def arg_name(arg): + if hasattr(arg, 'client_id'): + return 'Address(%s)' % str(arg) + elif is_tuple: + return str(arg)[1:-1].replace(" ", '') # strip('()') strips all + elif isinstance(arg, str): + if arg.startswith("'"): + return '"%s"' % arg + else: + return "'%s'" % arg + else: + return str(arg).replace(" ", '') + + # test successful client.resolve_address() + if expected_error is None: + try: + addr = tester.resolve_address(address_args) + except Exception as e: + print_stats(sys.stderr) + sys.stderr.write("%s raises unexpected %s\n" + % (arg_name(address_args), type(e).__name__)) + raise + else: + try: + addr = tester.resolve_address(address_args) + except expected_error: + test_count += 1 + fail_count += 1 + if verbose: + sys.stdout.write("%s raises %s (as expected)\n" + % (arg_name(address_args), + expected_error.__name__)) + return + except Exception as e: + print_stats(sys.stderr) + sys.stderr.write("%s raises %s (not expected %s)\n" + % (arg_name(address_args), + type(e).__name__, + expected_error.__name__)) + raise + else: + print_stats(sys.stderr) + sys.stdout.write("%s -> %d:%d without raising " + "expected error %s\n" + % (arg_name(address_args), + addr.client_id, + addr.port_id, + expected_error.__name__)) + raise AssertionError + + test_count += 1 + pass_count += 1 + + if addr.client_id != client_id or addr.port_id != port_id: + raise FileNotFoundError("%s -> %s, not expected (%d,%d)" + % (arg_name(address_args), + addr, + client_id, + port_id)) + + if verbose: + sys.stdout.write("%s -> %d:%d" + % (arg_name(address_args), + addr.client_id, + addr.port_id)) + if expected_error is not None: + sys.stdout.write(" with expected exception: %s\n" + % expected_error) + else: + sys.stdout.write('\n') + + return addr + + +# Tests permutations of "numeric-or-named-client:numeric-or-named-port", +# "client:port", "client.port", "'client:port'", (client, port), +# ((client, port)), etc. +def do_variants(client, + port, + separator, + quotes, + client_id, + port_id, + expected_error, + has_port_0=True): + + # just client + just_client_error = None if has_port_0 else FileNotFoundError + spec = "%s%s%s" % (quotes, client, quotes) + address_parse(spec, client_id, 0, just_client_error) + address_parse(client, client_id, 0, just_client_error) + + # client and port + spec = "%s%s%s%s%s" % (quotes, client, separator, port, quotes) + address_parse(spec, client_id, port_id, expected_error) + address_parse(client, port, client_id, port_id, expected_error) + address_parse((client, port), client_id, port_id, expected_error) + address_parse(client_id, port, client_id, port_id, expected_error) + address_parse((client_id, port), client_id, port_id, expected_error) + address_parse(client, port_id, client_id, port_id, expected_error) + address_parse((client, port_id), client_id, port_id, expected_error) + address_parse(client_id, port_id, client_id, port_id, expected_error) + address_parse((client_id, port_id), client_id, port_id, expected_error) + + if isinstance(client_id, int) and isinstance(port, int): + address_parse(Address(client_id, port_id), + client_id, + port_id, + expected_error) + + +# test all ports of client +def do_addresses(client_name, client_id, ports): + if verbose: + sys.stdout.write("# --- do_addresses(%s, %s) ---\n" + % (client_name, client_id)) + + # Not a rigorous test (ports could be explicit with first + # one not zero) but works for all test cases here where + # most are implicitly-ordered (so zero is first), and the + # one that isn't has no port 0 at all much less not + # as first port. + has_port_0 = (ports[0].id == 0) + + # permutations of address specifications + for tests_bits in range(1 << 4): + if (tests_bits & 1): + separator = ':' + else: + separator = '.' + + quotes = ("", "'", '"', None)[(tests_bits >> 1) & 0x3] + if quotes is None: + continue + + for port in ports: + do_variants(client_name, + port.name, + separator, + quotes, + client_id, + port.id, + None, + has_port_0) + + # test that flagged ports can be accessed via substring + for (port_ndx, port) in enumerate(ports): + if port.substr: + # test all 3 character substrings of port name + for char_ndx in range(0, len(port.name) - 3): + substring = port.name[char_ndx:char_ndx+3] + do_variants(client_name, + substring, + separator, + quotes, + client_id, + port.id, + None, + has_port_0) + + +# Test numeric client and port +# +# Note first client created by create_ports() (gets assigned i.d. 128) +# must have port id 0 (explicit or default assigned) +def do_simple(): + if verbose: + sys.stdout.write("# --- do_simple() ---\n") + + address_parse(128, 128, 0, None) + address_parse('128', 128, 0, None) + address_parse(128, 0, 128, 0, None) + address_parse('128:0', 128, 0, None) + address_parse((128, 0), 128, 0, None) + address_parse(Address(128, 0), 128, 0, None) + address_parse((Address(128, 0),), 128, 0, TypeError) + address_parse(128, Address(128, 0), 0, 0, TypeError) + address_parse((128, Address(128, 0)), 0, 0, TypeError) + + +# Test of client.resolve_address() overloads +def do_overloads(client_name, client_id, port_name, port_id): + if verbose: + sys.stdout.write("# --- do_overloads() ---\n") + + overloads = [] + + def test(*args): + overloads.append(address_parse(*(args + (client_id, port_id, None)))) + + test(client_id, port_id) # 2 ints + test((client_id, port_id)) # tuple, same + test(client_id, port_name) # int, str + test((client_id, port_name)) # tuple, same + test(client_name, port_id) # str, int + test((client_name, port_id)) # tuple, same + test(client_name, port_name) # 2 str + test((client_name, port_name)) # tuple, same + + test('%d:%d' % (client_id, port_id)) # int:int + test('%d:%s' % (client_id, port_name)) # int:str + test('%s:%d' % (client_name, port_id)) # str:iin + test('%s:%s' % (client_name, port_name)) # str:str + + overloads.append(tester.resolve_address(overloads[0])) # address + + assert (all(addr == overloads[0] for addr in overloads)) + + +def do_bad_arguments(): + if verbose: + sys.stdout.write("# --- do_bad_arguments() ---\n") + + address_parse(3.14, 0, 0, TypeError) + address_parse([], 0, 0, TypeError) + address_parse(tester, 0, 0, TypeError) + + address_parse((128,), 0, 0, TypeError) + address_parse(('explicit',), 0, 0, TypeError) + address_parse((3.14,), 0, 0, TypeError) + address_parse((Address(128, 0),), 0, 0, TypeError) + + address_parse(('explicit', None), 0, 0, FileNotFoundError) + address_parse((3.14, None), 0, 0, TypeError) + address_parse((Address(128, 0), None), 0, 0, TypeError) + + address_parse(128, 2, 3, 0, 0, TypeError) + address_parse((128, 2, 3), 0, 0, TypeError) + address_parse(128, 2, 3, 4, 0, 0, TypeError) + address_parse((128, 2, 3, 4), 0, 0, TypeError) + + address_parse(3.14, 128, 0, 0, TypeError) + address_parse((3.14, 128), 0, 0, TypeError) + address_parse(128, 3.14, 0, 0, TypeError) + address_parse((128, 3.14), 0, 0, TypeError) + + address_parse(128, 'explicit', 0, 0, FileNotFoundError) + address_parse((128, 'explicit'), 0, 0, FileNotFoundError) + address_parse('explicit', 128, 0, 0, FileNotFoundError) + address_parse(('explicit', 128), 0, 0, FileNotFoundError) + + address_parse('explicit', 3.14, 0, 0, TypeError) + address_parse(('explicit', 3.14), 0, 0, TypeError) + address_parse(3.14, 'explicit', 0, 0, TypeError) + address_parse((3.14, 'explicit'), 0, 0, TypeError) + + address_parse(Address(128, 0), 128, 0, 0, TypeError) + address_parse((Address(128, 0)), 128, 0, 0, TypeError) + + address_parse(128, Address(128, 0), 0, 0, TypeError) + address_parse((128, Address(128, 0)), 0, 0, TypeError) + + +def do_bad_clients_ports(): + if verbose: + sys.stdout.write("# --- do_bad_clients_ports() ---\n") + for client in (-1, 256, 'bad_client'): + for port in (0, -1, 256, 'bad_port'): + do_variants(client, + port, + ':', + '', + client, + port, + FileNotFoundError, + has_port_0=False) + + +def do_good_client_bad_ports(client_name, client_id): + for port in (-1, 256, 'bad_port'): + for separator in (':', '.'): + for quotes in ('', "'", '"'): + spec = "%s%s%s%s%s" % (quotes, + client_name, + separator, + port, + quotes) + address_parse(spec, + client_name, + port, + client_id, + 0, + TypeError) + address_parse(client_name, port, client_id, 0, FileNotFoundError) + address_parse((client_name, port), client_id, 0, FileNotFoundError) + address_parse(client_id, port, client_id, 0, FileNotFoundError) + address_parse((client_id, port), client_id, 0, FileNotFoundError) + + +@pytest.mark.require_alsa_seq +def test_sequencer_address(): + global tester + + tester = alsa_midi.SequencerClient("tester") # all tests use this + sequential = alsa_midi.SequencerClient("sequential") + prefix = alsa_midi.SequencerClient("seq") + explicit = alsa_midi.SequencerClient("explicit") + + # First one gets assigned client id 128, must have port + # with id 0 for do_simple() + test_id = tester .get_client_info().client_id + seqn_id = sequential.get_client_info().client_id + prfx_id = prefix .get_client_info().client_id + expl_id = explicit .get_client_info().client_id + + create_ports(tester, tester_ports) + create_ports(sequential, sequential_ports) + create_ports(prefix, prefix_ports) + create_ports(explicit, explicit_ports) + + # sometimes delay before clients/ports actually exist + # particularly if run in tox/pytest environment + time.sleep(2) + + do_simple() + do_bad_arguments() + do_overloads('explicit', expl_id, 'eleven', 11) # only need to do one + do_bad_clients_ports() + do_good_client_bad_ports('explicit', expl_id) # only need to do one + + do_addresses("tester", test_id, tester_ports) + do_addresses("sequential", seqn_id, sequential_ports) + do_addresses("seq", prfx_id, prefix_ports) # client,port substrings + do_addresses("explicit", expl_id, explicit_ports) + do_addresses("expl", expl_id, explicit_ports) # substrings + + +if __name__ == '__main__': + # commandline options + parser = argparse.ArgumentParser() + parser.add_argument('-v', '--verbose', + action="store_true", + help="Ouput results of each test (note: thousands)") + parser.add_argument('-p', '--pause', + action="store_true", + help="Pause after tests") + opts = parser.parse_args() + + verbose = opts.verbose + + test_sequencer_address() + + if opts.verbose or opts.pause: + print_stats() + + if opts.pause: + sys.stdout.write("`aconnect -oil` in separate shell for " + "clients+ports, type ^C here to exit\n") + try: + while True: + pass + except KeyboardInterrupt: + sys.stdout.write("exiting\n") + + sys.exit(0)