onyx_online/vban-cmd-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/vban-cmd-python.git synced 2024-11-15 17:10:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

make better use of pattern matching features

Browse Source 
error test updated
This commit is contained in:
onyx-and-iris 2023-08-10 19:12:52 +01:00
parent a61e09b075
commit a635109308
2 changed files with 8 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/test_errors.py
View File

@ -32,5 +32,5 @@ class TestErrors:
"unknown-0": {"state": True}, "unknown-0": {"state": True},
"vban-out-1": {"name": "streamname"}, "vban-out-1": {"name": "streamname"},
} }
with pytest.raises(ValueError, match="invalid config key 'unknown'"): with pytest.raises(ValueError, match="invalid config key 'unknown-0'"):
vban.apply(CONFIG) vban.apply(CONFIG)

17
vban_cmd/vbancmd.py
View File

@ -183,17 +183,13 @@ class VbanCmd(metaclass=ABCMeta):
""" """
def target(key): def target(key):
kls, m2, *rem = key.split("-") match key.split("-"):
match kls: case ["strip" | "bus" | "button" as kls, index]:
case "strip" | "bus" | "button":
index = m2
target = getattr(self, kls) target = getattr(self, kls)
case "vban": case ["vban", direction, index]:
dir = f"{m2.rstrip('stream')}stream" target = getattr(self.vban, f"{direction.rstrip('stream')}stream")
index = rem[0]
target = getattr(self.vban, dir)
case _: case _:
ERR_MSG = f"invalid config key '{kls}'" ERR_MSG = f"invalid config key '{key}'"
self.logger.error(ERR_MSG) self.logger.error(ERR_MSG)
raise ValueError(ERR_MSG) raise ValueError(ERR_MSG)
return target[int(index)] return target[int(index)]
@ -229,7 +225,8 @@ class VbanCmd(metaclass=ABCMeta):
if not self.stopped(): if not self.stopped():
self.logger.debug("events thread shutdown started") self.logger.debug("events thread shutdown started")
self.stop_event.set() self.stop_event.set()
self.subscriber.join() # wait for subscriber thread to complete cycle for t in (self.producer, self.subscriber):
t.join()
[sock.close() for sock in self.socks] [sock.close() for sock in self.socks]
self.logger.info(f"{type(self).__name__}: Successfully logged out of {self}") self.logger.info(f"{type(self).__name__}: Successfully logged out of {self}")