From 941c024b2a385671e1b1da2ce1d097421b3b86dd Mon Sep 17 00:00:00 2001 From: thanks4opensource <31165662+thanks4opensource@users.noreply.github.com> Date: Wed, 7 Oct 2026 22:09:16 -0700 Subject: [PATCH 1/4] client: add get_address() and fd() Parses various input overloads such as 'client-name:port_name', ('client-name','port-name), etc., returns Address. Raises exception if Address(client_id, port_id) doesn't exist. Added new fd() public accessor. Added tests/test_sequencer_address.py. Modified examples/connect.py to accept 'client-name:port-name'. --- alsa_midi/client.py | 183 +++++++++++- examples/connect.py | 11 +- tests/test_sequencer_address.py | 485 ++++++++++++++++++++++++++++++++ 3 files changed, 674 insertions(+), 5 deletions(-) create mode 100644 tests/test_sequencer_address.py diff --git a/alsa_midi/client.py b/alsa_midi/client.py index 030559d..045ff8b 100644 --- a/alsa_midi/client.py +++ b/alsa_midi/client.py @@ -419,6 +419,10 @@ def _get_fds(self): assert (pfds[0].events & select.POLLIN) and (pfds[0].events & select.POLLOUT) self._fd = pfds[0].fd + def fd(self): + "Returns the ALSA sequencer file descriptor number." + return self._fd + def _get_event_parser(self): parser = self._event_parser if parser is None: @@ -1120,7 +1124,7 @@ def list_ports(self, *, The result is sorted in a way that the first returned entry should be the 'most usable' one for the selected purpose. E.g. when `output` = `True` then the first entry will be - a synthesizer input port rather than the dummy 'Midi Through' port. This is still a guess, + a synthesizer input port rather than the dummy 'Midi Through' port. Thisa is still a guess, though, so in the end the user should be able to choose. Wraps :alsa:`snd_seq_query_next_client` and :alsa:`snd_seq_query_next_port`. @@ -1232,6 +1236,183 @@ def list_ports(self, *, return result + + @overload + def get_address(self, arg1: str) -> 'Address': + ... + + @overload + def get_address(self, arg1: str, arg2: str = '') -> 'Address': + ... + + @overload + def get_address(self, arg1: int, arg2: int = 0) -> 'Address': + ... + + @overload + def get_address(self, arg1: tuple) -> 'Address': + ... + + @overload + def get_address(self, arg1: AddressType) -> 'Address': + ... + + + def get_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( "(%s, %s) is not one or two ints and/or " + "strings, or 2-tuple of same, or Address" + % (arg1, arg2)) + + if arg2 is None: + if isinstance(arg1, str): + # remove enclosing single or double quotes if any + if arg1.startswith("'"): + arg1 = arg1.strip("'") + elif arg1.startswith('"'): + arg1 = arg1.strip('"') + # 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 + elif isinstance(arg1, int): + client_name = arg1 + port_name = None + elif isinstance(arg1, tuple): + if arg2 is not None: + type_error() + if len(arg1) == 2: + (client_name,port_name) = arg1 + else: + type_error() + elif isinstance(arg1, Address): + if arg2 is not None: + type_error() + client_name = arg2.client_id # int will be converted below + port_name = arg2.port_id # " " " " " + else: + type_error() + else: + client_name = arg1 + port_name = arg2 + + + if not ( isinstance(client_name, (int,str )) + and isinstance(port_name, (int,str,type(None)))): + type_error() + + 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..01ef3b6 100755 --- a/examples/connect.py +++ b/examples/connect.py @@ -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.get_address(args.sender) + dest = client.get_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..a38d047 --- /dev/null +++ b/tests/test_sequencer_address.py @@ -0,0 +1,485 @@ +import argparse +import sys + +from collections import namedtuple + +import alsa_midi +from alsa_midi import Address + + + +"""Tests SequencerClientBase.get_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 +""" + + +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.get_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 + 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 + + 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: # and isinstance(address_args[0], tuple): + 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.get_address() + if expected_error is None: + try: + addr = tester.get_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.get_address(address_args) + except expected_error: + test_count += 1 + fail_count += 1 + if opts.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 opts.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" % expect_except) + 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 test_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 test_addresses(client_name, client_id, ports): + if opts.verbose: + sys.stdout.write( "# --- test_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 + + no_ports = not(tests_bits << 3) + + for port in ports: + test_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] + test_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 test_simple(): + if opts.verbose: sys.stdout.write("# --- test_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.get_address() overloads +def test_overloads(client_name, client_id, port_name, port_id): + if opts.verbose: sys.stdout.write("# --- test_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.get_address(overloads[0])) # address + + assert(all(addr == overloads[0] for addr in overloads)) + + + +def test_bad_arguments(): + if opts.verbose: sys.stdout.write("# --- test_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 test_bad_clients_ports(): + if opts.verbose: sys.stdout.write("# --- test_bad_clients_ports() ---\n") + for client in (-1, 256, 'bad_client'): + for port in (0, -1, 256, 'bad_port'): + test_variants(client, + port, + ':', + '', + client, + port, + FileNotFoundError, + has_port_0=False) + + + +def test_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) + + + +if __name__ == '__main__': + 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 test_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 + + # 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() + + clients_ports = {} # have to store ports somewhere else go out of scope + # and get deleted on return from create_ports() + create_ports(tester , tester_ports ) + create_ports(sequential, sequential_ports) + create_ports(prefix , prefix_ports ) + create_ports(explicit , explicit_ports ) + + test_simple() + test_bad_arguments() + test_overloads('explicit', expl_id, 'eleven', 11) # only need to do one + test_bad_clients_ports() + test_good_client_bad_ports('explicit', expl_id) # only need to do one + + test_addresses("tester", test_id, tester_ports) + test_addresses("sequential", seqn_id, sequential_ports) + test_addresses("seq", prfx_id, prefix_ports) # client,port substrings + test_addresses("explicit", expl_id, explicit_ports) + test_addresses("expl", expl_id, explicit_ports) # substrings + + 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") + + explicit.close() + sequential.close() + tester.close() + + sys.exit(0) From b16d7dfcd200297380598e7197865527d727cc3e Mon Sep 17 00:00:00 2001 From: thanks4opensource <31165662+thanks4opensource@users.noreply.github.com> Date: Wed, 14 Oct 2026 00:48:42 -0700 Subject: [PATCH 2/4] client: add resolve_address() and fd Parses various input overloads such as 'client-name:port_name', ('client-name','port-name'), etc., returns Address. Raises exception if Address(client_id, port_id) doesn't exist. Added new fd public getter. Added tests/test_sequencer_address.py. Modified examples/connect.py to accept 'client-name:port-name' Update: - Changed get_address() to resolve_address() - Added @property to fd() to make read-only fd - Fixed test_sequencer_address.py to run under tox/pytest - Changed whitespace,etc to pass tox style/flake8, except in test_sequencer_address.py --- alsa_midi/client.py | 53 +++++----- examples/connect.py | 6 +- tests/test_sequencer_address.py | 166 +++++++++++++++++--------------- 3 files changed, 118 insertions(+), 107 deletions(-) diff --git a/alsa_midi/client.py b/alsa_midi/client.py index 045ff8b..8ed5650 100644 --- a/alsa_midi/client.py +++ b/alsa_midi/client.py @@ -419,6 +419,7 @@ def _get_fds(self): assert (pfds[0].events & select.POLLIN) and (pfds[0].events & select.POLLOUT) self._fd = pfds[0].fd + @property def fd(self): "Returns the ALSA sequencer file descriptor number." return self._fd @@ -1236,29 +1237,27 @@ def list_ports(self, *, return result - @overload - def get_address(self, arg1: str) -> 'Address': + def resolve_address(self, arg1: str) -> 'Address': ... @overload - def get_address(self, arg1: str, arg2: str = '') -> 'Address': + def resolve_address(self, arg1: str, arg2: str = '') -> 'Address': ... @overload - def get_address(self, arg1: int, arg2: int = 0) -> 'Address': + def resolve_address(self, arg1: int, arg2: int = 0) -> 'Address': ... @overload - def get_address(self, arg1: tuple) -> 'Address': + def resolve_address(self, arg1: tuple) -> 'Address': ... @overload - def get_address(self, arg1: AddressType) -> 'Address': + def resolve_address(self, arg1: AddressType) -> 'Address': ... - - def get_address(self, + def resolve_address(self, arg1: Union[int, str, tuple, AddressType], arg2: Union[int, str, None] = None) -> Address: """Parse client:port address string @@ -1295,8 +1294,8 @@ def get_address(self, """ def type_error(): - raise TypeError( "(%s, %s) is not one or two ints and/or " - "strings, or 2-tuple of same, or Address" + raise TypeError("(%s, %s) is not one or two ints and/or " + "strings, or 2-tuple of same, or Address" % (arg1, arg2)) if arg2 is None: @@ -1312,34 +1311,33 @@ def type_error(): # split client and port client_port = arg1.split(':') if len(client_port) == 2: - (client_name,port_name) = client_port + (client_name, port_name) = client_port else: client_name = arg1 - port_name = None + port_name = None elif isinstance(arg1, int): client_name = arg1 - port_name = None + port_name = None elif isinstance(arg1, tuple): if arg2 is not None: type_error() if len(arg1) == 2: - (client_name,port_name) = arg1 + (client_name, port_name) = arg1 else: type_error() elif isinstance(arg1, Address): if arg2 is not None: type_error() client_name = arg2.client_id # int will be converted below - port_name = arg2.port_id # " " " " " + port_name = arg2.port_id # same else: type_error() else: client_name = arg1 - port_name = arg2 + port_name = arg2 - - if not ( isinstance(client_name, (int,str )) - and isinstance(port_name, (int,str,type(None)))): + if not (isinstance(client_name, (int, str)) + and isinstance(port_name, (int, str, type(None)))): type_error() ports = self.list_ports(include_system=True) @@ -1359,12 +1357,12 @@ def type_error(): 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 + 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 + client_id = port_info.client_id best_client = port_info else: for port_info in ports: @@ -1397,22 +1395,21 @@ def type_error(): break elif port_info.name == port_name: # exact match supercedes prefix matches - port_id = port_info.port_id + port_id = port_info.port_id best_port = port_info break - elif isinstance(port_name, str) \ - and port_name in port_info.name: + elif isinstance(port_name, str) \ + and port_name in port_info.name: # last of multiples will overwrite - port_id = port_info.port_id + 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) ) + 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 01ef3b6..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(): @@ -24,8 +24,8 @@ def main(): client = SequencerClient("dump_events.py") - sender = client.get_address(args.sender) - dest = client.get_address(args.dest) + sender = client.resolve_address(args.sender) + dest = client.resolve_address(args.dest) if args.disconnect: client.unsubscribe_port(sender, dest) diff --git a/tests/test_sequencer_address.py b/tests/test_sequencer_address.py index a38d047..378f079 100644 --- a/tests/test_sequencer_address.py +++ b/tests/test_sequencer_address.py @@ -1,5 +1,6 @@ import argparse import sys +import time from collections import namedtuple @@ -8,7 +9,7 @@ -"""Tests SequencerClientBase.get_address() +"""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)), @@ -25,11 +26,17 @@ """ +verbose = False + +clients_ports = {} # have to store ports somewhere else go out of scope + # and get deleted on return from create_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 # @@ -41,7 +48,7 @@ OUTPUT_CAPS = alsa_midi.READ_PORT IN_OUT_CAPS = INPUT_CAPS | OUTPUT_CAPS -# all tests use this for client.get_address(...) +# all tests use this for client.resolve_address(...) tester_ports = ( port_spec("zero", -1, 0, False, IN_OUT_CAPS), ) @@ -148,10 +155,10 @@ def arg_name(arg): else: return str(arg).replace(" ", '') - # test successful client.get_address() + # test successful client.resolve_address() if expected_error is None: try: - addr = tester.get_address(address_args) + addr = tester.resolve_address(address_args) except Exception as e: print_stats(sys.stderr) sys.stderr.write( "%s raises unexpected %s\n" @@ -159,11 +166,11 @@ def arg_name(arg): raise else: try: - addr = tester.get_address(address_args) + addr = tester.resolve_address(address_args) except expected_error: test_count += 1 fail_count += 1 - if opts.verbose: + if verbose: sys.stdout.write( "%s raises %s (as expected)\n" % (arg_name(address_args), expected_error.__name__)) @@ -195,7 +202,7 @@ def arg_name(arg): client_id, port_id ) ) - if opts.verbose: + if verbose: sys.stdout.write( "%s -> %d:%d" % (arg_name(address_args), addr.client_id, @@ -211,14 +218,14 @@ def arg_name(arg): # Tests permutations of "numeric-or-named-client:numeric-or-named-port", # "client:port", "client.port", "'client:port'", (client, port), # ((client, port)), etc. -def test_variants(client, - port, - separator, - quotes, - client_id, - port_id, - expected_error, - has_port_0=True): +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 @@ -247,9 +254,9 @@ def test_variants(client, # test all ports of client -def test_addresses(client_name, client_id, ports): - if opts.verbose: - sys.stdout.write( "# --- test_addresses(%s, %s) ---\n" +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 @@ -270,14 +277,14 @@ def test_addresses(client_name, client_id, ports): no_ports = not(tests_bits << 3) for port in ports: - test_variants(client_name, - port.name, - separator, - quotes, - client_id, - port.id, - None, - has_port_0) + 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): @@ -285,14 +292,14 @@ def test_addresses(client_name, client_id, ports): # 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] - test_variants(client_name, - substring, - separator, - quotes, - client_id, - port.id, - None, - has_port_0) + do_variants(client_name, + substring, + separator, + quotes, + client_id, + port.id, + None, + has_port_0) @@ -300,8 +307,8 @@ def test_addresses(client_name, client_id, ports): # # Note first client created by create_ports() (gets assigned i.d. 128) # must have port id 0 (explicit or default assigned) -def test_simple(): - if opts.verbose: sys.stdout.write("# --- test_simple() ---\n") +def do_simple(): + if verbose: sys.stdout.write("# --- do_simple() ---\n") address_parse( 128, 128, 0, None) address_parse( '128', 128, 0, None) @@ -315,9 +322,9 @@ def test_simple(): -# Test of client.get_address() overloads -def test_overloads(client_name, client_id, port_name, port_id): - if opts.verbose: sys.stdout.write("# --- test_overloads() ---\n") +# 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 = [] @@ -338,14 +345,14 @@ def test(*args): test('%s:%d' % (client_name, port_id )) # str:iin test('%s:%s' % (client_name, port_name)) # str:str - overloads.append(tester.get_address(overloads[0])) # address + overloads.append(tester.resolve_address(overloads[0])) # address assert(all(addr == overloads[0] for addr in overloads)) -def test_bad_arguments(): - if opts.verbose: sys.stdout.write("# --- test_bad_arguments() ---\n") +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) @@ -388,22 +395,22 @@ def test_bad_arguments(): -def test_bad_clients_ports(): - if opts.verbose: sys.stdout.write("# --- test_bad_clients_ports() ---\n") +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'): - test_variants(client, - port, - ':', - '', - client, - port, - FileNotFoundError, - has_port_0=False) + do_variants(client, + port, + ':', + '', + client, + port, + FileNotFoundError, + has_port_0=False) -def test_good_client_bad_ports(client_name, client_id): +def do_good_client_bad_ports(client_name, client_id): for port in (-1, 256, 'bad_port'): for separator in (':', '.'): for quotes in ('', "'", '"'): @@ -425,19 +432,45 @@ def test_good_client_bad_ports(client_name, client_id): -if __name__ == '__main__': +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 test_simple() + # 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', @@ -448,24 +481,9 @@ def test_good_client_bad_ports(client_name, client_id): help="Pause after tests") opts = parser.parse_args() - clients_ports = {} # have to store ports somewhere else go out of scope - # and get deleted on return from create_ports() - create_ports(tester , tester_ports ) - create_ports(sequential, sequential_ports) - create_ports(prefix , prefix_ports ) - create_ports(explicit , explicit_ports ) + verbose = opts.verbose - test_simple() - test_bad_arguments() - test_overloads('explicit', expl_id, 'eleven', 11) # only need to do one - test_bad_clients_ports() - test_good_client_bad_ports('explicit', expl_id) # only need to do one - - test_addresses("tester", test_id, tester_ports) - test_addresses("sequential", seqn_id, sequential_ports) - test_addresses("seq", prfx_id, prefix_ports) # client,port substrings - test_addresses("explicit", expl_id, explicit_ports) - test_addresses("expl", expl_id, explicit_ports) # substrings + test_sequencer_address() if opts.verbose or opts.pause: print_stats() @@ -478,8 +496,4 @@ def test_good_client_bad_ports(client_name, client_id): except KeyboardInterrupt: sys.stdout.write("exiting\n") - explicit.close() - sequential.close() - tester.close() - sys.exit(0) From 8779e02e040a21444bfeae9e57af159855c780f9 Mon Sep 17 00:00:00 2001 From: thanks4opensource <31165662+thanks4opensource@users.noreply.github.com> Date: Sat, 17 Oct 2026 20:18:25 -0700 Subject: [PATCH 3/4] client: add resolve_address() and fd Parses various input overloads such as 'client-name:port_name', ('client-name','port-name'), etc., returns Address. Raises exception if Address(client_id, port_id) doesn't exist. Added new fd public getter. Added tests/test_sequencer_address.py. Modified examples/connect.py to accept 'client-name:port-name' Update: - Changed get_address() to resolve_address() - Added @property to fd() to make read-only fd - Fixed test_sequencer_address.py to run under tox/pytest - Changed whitespace,etc to pass tox style/flake8, except in test_sequencer_address.py Update: - Whitespace/indentation changes, directives, and exclusions for for tox/pytest/isort. - All tests pass without error. --- alsa_midi/client.py | 4 ++-- tests/test_sequencer_address.py | 8 +++++--- tox.ini | 1 + 3 files changed, 8 insertions(+), 5 deletions(-) diff --git a/alsa_midi/client.py b/alsa_midi/client.py index 8ed5650..784bd09 100644 --- a/alsa_midi/client.py +++ b/alsa_midi/client.py @@ -1258,8 +1258,8 @@ def resolve_address(self, arg1: AddressType) -> 'Address': ... def resolve_address(self, - arg1: Union[int, str, tuple, AddressType], - arg2: Union[int, str, None] = None) -> Address: + 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 diff --git a/tests/test_sequencer_address.py b/tests/test_sequencer_address.py index 378f079..6860738 100644 --- a/tests/test_sequencer_address.py +++ b/tests/test_sequencer_address.py @@ -1,11 +1,13 @@ +# flake8: noqa + import argparse import sys import time -from collections import namedtuple +from collections import namedtuple # isort: skip -import alsa_midi -from alsa_midi import Address +import alsa_midi # isort: skip +from alsa_midi import Address # isort: skip diff --git a/tox.ini b/tox.ini index 9e7c555..c4caf40 100644 --- a/tox.ini +++ b/tox.ini @@ -44,3 +44,4 @@ commands = [flake8] max-line-length=99 +exclude = tests/test_sequencer_address.py From 31d56a15197328083a985b0c357b27c60e158fe8 Mon Sep 17 00:00:00 2001 From: thanks4opensource <31165662+thanks4opensource@users.noreply.github.com> Date: Sun, 18 Oct 2026 22:59:00 -0700 Subject: [PATCH 4/4] commit message: client: add resolve_address() and fileno() Parses various input overloads such as 'client-name:port_name', ('client-name','port-name'), etc., returns Address. Raises exception if Address(client_id, port_id) doesn't exist. Added new fileno() public accessor. Added tests/test_sequencer_address.py. Modified examples/connect.py to accept 'client-name:port-name' Update: - Changed get_address() to resolve_address() - Added @property to fd() to make read-only fd - Fixed test_sequencer_address.py to run under tox/pytest - Changed whitespace,etc to pass tox style/flake8, except in test_sequencer_address.py Update: - Whitespace/indentation changes, directives, and exclusions for for tox/pytest/isort. - All tests pass without error. Update: client.py: - Removed "@property\ndef fd(self):" - Replaced with Jajcus-provided "fileno(self)" - Removed typo/bug, redesigned/rewrote argument validation in resolve_address() - Fixed typo introduced in docstring test_sequencer_address.py - Fixed bug, use of non-existent variable - Removed unused variable - Removed "# flake8: noqa" - Reformatted code with autopep8 - Moved file-level docstring to top of file - Added "import pytest" and "@pytest.mark.require_alsa_seq" - Added more "# isort: skip" (imports in correct order, still produce errors) - Added "# noqua: F824" (global variables *are* used, flake8 flags incorrectly) tox.ini - Removed "-exclude = tests/test_sequencer_address.py" --- alsa_midi/client.py | 116 ++++++++++----- tests/test_sequencer_address.py | 256 ++++++++++++++++---------------- tox.ini | 1 - 3 files changed, 201 insertions(+), 172 deletions(-) diff --git a/alsa_midi/client.py b/alsa_midi/client.py index 784bd09..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`.""" @@ -419,11 +420,6 @@ def _get_fds(self): assert (pfds[0].events & select.POLLIN) and (pfds[0].events & select.POLLOUT) self._fd = pfds[0].fd - @property - def fd(self): - "Returns the ALSA sequencer file descriptor number." - return self._fd - def _get_event_parser(self): parser = self._event_parser if parser is None: @@ -435,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. @@ -1125,7 +1136,7 @@ def list_ports(self, *, The result is sorted in a way that the first returned entry should be the 'most usable' one for the selected purpose. E.g. when `output` = `True` then the first entry will be - a synthesizer input port rather than the dummy 'Midi Through' port. Thisa is still a guess, + a synthesizer input port rather than the dummy 'Midi Through' port. This is still a guess, though, so in the end the user should be able to choose. Wraps :alsa:`snd_seq_query_next_client` and :alsa:`snd_seq_query_next_port`. @@ -1294,52 +1305,75 @@ def resolve_address(self, """ def type_error(): - raise TypeError("(%s, %s) is not one or two ints and/or " - "strings, or 2-tuple of same, or Address" + 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)) - if arg2 is None: + 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): - # remove enclosing single or double quotes if any - if arg1.startswith("'"): - arg1 = arg1.strip("'") - elif arg1.startswith('"'): - arg1 = arg1.strip('"') - # 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 + 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 = None - elif isinstance(arg1, int): - client_name = arg1 - port_name = None - elif isinstance(arg1, tuple): - if arg2 is not None: - type_error() - if len(arg1) == 2: - (client_name, port_name) = arg1 - else: - type_error() - elif isinstance(arg1, Address): - if arg2 is not None: - type_error() - client_name = arg2.client_id # int will be converted below - port_name = arg2.port_id # same + port_name = arg2 # OK if None else: - type_error() - else: - client_name = arg1 - port_name = arg2 + 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 diff --git a/tests/test_sequencer_address.py b/tests/test_sequencer_address.py index 6860738..994f963 100644 --- a/tests/test_sequencer_address.py +++ b/tests/test_sequencer_address.py @@ -1,16 +1,3 @@ -# flake8: noqa - -import argparse -import sys -import time - -from collections import namedtuple # isort: skip - -import alsa_midi # isort: skip -from alsa_midi import Address # isort: skip - - - """Tests SequencerClientBase.resolve_address() Tests many permutations of "numeric-or-named-client:numeric-or-named-port", @@ -28,17 +15,29 @@ """ +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 -clients_ports = {} # have to store ports somewhere else go out of scope - # and get deleted on return from create_ports() +# 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 # @@ -46,7 +45,7 @@ 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 +INPUT_CAPS = alsa_midi.WRITE_PORT OUTPUT_CAPS = alsa_midi.READ_PORT IN_OUT_CAPS = INPUT_CAPS | OUTPUT_CAPS @@ -58,52 +57,50 @@ # 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 ), + 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 ), + 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 ), + 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 - stream.write( "%d tests\n" - "%d passed\n" - "%d failed with expected errors \n" + 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 + global clients_ports # noqa: F824 client_name = client.get_client_info().name @@ -118,7 +115,6 @@ def create_ports(client, port_descs): 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. @@ -127,17 +123,17 @@ def address_parse(*args): # and expected-error-or-None assert len(args) > 3, \ - "internal test program bug: address_parse(%s) <= 3\n" % (args,) + "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] + 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: # and isinstance(address_args[0], tuple): + if len(address_args) == 1: address_args = address_args[0] is_tuple = False # for arg_name() else: @@ -148,7 +144,7 @@ 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 + return str(arg)[1:-1].replace(" ", '') # strip('()') strips all elif isinstance(arg, str): if arg.startswith("'"): return '"%s"' % arg @@ -163,7 +159,7 @@ def arg_name(arg): addr = tester.resolve_address(address_args) except Exception as e: print_stats(sys.stderr) - sys.stderr.write( "%s raises unexpected %s\n" + sys.stderr.write("%s raises unexpected %s\n" % (arg_name(address_args), type(e).__name__)) raise else: @@ -173,21 +169,21 @@ def arg_name(arg): test_count += 1 fail_count += 1 if verbose: - sys.stdout.write( "%s raises %s (as expected)\n" + sys.stdout.write("%s raises %s (as expected)\n" % (arg_name(address_args), expected_error.__name__)) - return; + return except Exception as e: print_stats(sys.stderr) - sys.stderr.write( "%s raises %s (not expected %s)\n" + 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" + sys.stdout.write("%s -> %d:%d without raising " + "expected error %s\n" % (arg_name(address_args), addr.client_id, addr.port_id, @@ -198,19 +194,20 @@ def arg_name(arg): pass_count += 1 if addr.client_id != client_id or addr.port_id != port_id: - raise FileNotFoundError( "%s -> %s, not expected (%d,%d)" + raise FileNotFoundError("%s -> %s, not expected (%d,%d)" % (arg_name(address_args), addr, client_id, - port_id ) ) + port_id)) if verbose: - sys.stdout.write( "%s -> %d:%d" + 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" % expect_except) + sys.stdout.write(" with expected exception: %s\n" + % expected_error) else: sys.stdout.write('\n') @@ -237,28 +234,27 @@ def do_variants(client, # 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(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_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, 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) address_parse((client_id, port_id), client_id, port_id, expected_error) - if isinstance(client_id, int) and isinstance(port, int) : + 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" + sys.stdout.write("# --- do_addresses(%s, %s) ---\n" % (client_name, client_id)) # Not a rigorous test (ports could be explicit with first @@ -269,14 +265,15 @@ def do_addresses(client_name, client_id, ports): 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 = '.' + 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 - - no_ports = not(tests_bits << 3) + if quotes is None: + continue for port in ports: do_variants(client_name, @@ -289,7 +286,7 @@ def do_addresses(client_name, client_id, ports): has_port_0) # test that flagged ports can be accessed via substring - for (port_ndx,port) in enumerate(ports): + 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): @@ -304,57 +301,57 @@ def do_addresses(client_name, client_id, ports): 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) + 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) 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") + 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_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 + 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)) - + assert (all(addr == overloads[0] for addr in overloads)) def do_bad_arguments(): - if verbose: sys.stdout.write("# --- do_bad_arguments() ---\n") + if verbose: + sys.stdout.write("# --- do_bad_arguments() ---\n") address_parse(3.14, 0, 0, TypeError) address_parse([], 0, 0, TypeError) @@ -363,42 +360,42 @@ def do_bad_arguments(): 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((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((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), 0, 0, TypeError) - address_parse( 128, 2, 3, 4, 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((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, 3.14), 0, 0, TypeError) - address_parse( 128, 'explicit', 0, 0, FileNotFoundError) + 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', 128), 0, 0, FileNotFoundError) - address_parse( 'explicit', 3.14, 0, 0, TypeError) + 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((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((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) address_parse((128, Address(128, 0)), 0, 0, TypeError) - def do_bad_clients_ports(): - if verbose: sys.stdout.write("# --- do_bad_clients_ports() ---\n") + 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, @@ -411,7 +408,6 @@ def do_bad_clients_ports(): has_port_0=False) - def do_good_client_bad_ports(client_name, client_id): for port in (-1, 256, 'bad_port'): for separator in (':', '.'): @@ -427,20 +423,20 @@ def do_good_client_bad_ports(client_name, client_id): 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_name, port), client_id, 0, FileNotFoundError) - address_parse( client_id, 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 + tester = alsa_midi.SequencerClient("tester") # all tests use this sequential = alsa_midi.SequencerClient("sequential") - prefix = alsa_midi.SequencerClient("seq" ) - explicit = alsa_midi.SequencerClient("explicit" ) + 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() @@ -449,10 +445,10 @@ def test_sequencer_address(): prfx_id = prefix .get_client_info().client_id expl_id = explicit .get_client_info().client_id - create_ports(tester , tester_ports ) + create_ports(tester, tester_ports) create_ports(sequential, sequential_ports) - create_ports(prefix , prefix_ports ) - create_ports(explicit , explicit_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 @@ -466,10 +462,9 @@ def test_sequencer_address(): 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("seq", prfx_id, prefix_ports) # client,port substrings do_addresses("explicit", expl_id, explicit_ports) - do_addresses("expl", expl_id, explicit_ports) # substrings - + do_addresses("expl", expl_id, explicit_ports) # substrings if __name__ == '__main__': @@ -492,9 +487,10 @@ def test_sequencer_address(): if opts.pause: sys.stdout.write("`aconnect -oil` in separate shell for " - "clients+ports, type ^C here to exit\n") + "clients+ports, type ^C here to exit\n") try: - while True: pass + while True: + pass except KeyboardInterrupt: sys.stdout.write("exiting\n") diff --git a/tox.ini b/tox.ini index c4caf40..9e7c555 100644 --- a/tox.ini +++ b/tox.ini @@ -44,4 +44,3 @@ commands = [flake8] max-line-length=99 -exclude = tests/test_sequencer_address.py