edits to tests

add more tests to higher.

added opts to tests.

removed delay, max_polls from runmany, not required for these tests.
This commit is contained in:
onyx-and-iris
2022-02-28 18:14:31 +00:00
parent 314ecbe6da
commit 83429e17ad
8 changed files with 68 additions and 41 deletions

View File

@@ -72,7 +72,7 @@ class OutputBus(Channel):
self.setter('Label', val)
@property
def gain(self):
def gain(self) -> float:
def fget():
val = self.public_packet.busgain[self.index]
if val < 10000:

View File

@@ -53,6 +53,7 @@ class Modes:
_sel: hex=0x10000000
_monitor: hex=0x20000000
class Channel(abc.ABC):
""" Base class for InputStrip and OutputBus. """
def __init__(self, remote, index):

View File

@@ -158,7 +158,7 @@ class GainLayer(InputStrip):
self._i = i
@property
def gain(self):
def gain(self) -> float:
def fget():
val = getattr(self.public_packet, f'stripgainlayer{self._i+1}')[self.index]
if val < 10000:
@@ -170,7 +170,7 @@ class GainLayer(InputStrip):
return round((fget() * 0.01), 1)
@gain.setter
def gain(self, val):
def gain(self, val: float):
self.setter(f'GainLayer[{self._i}]', val)

View File

@@ -2,7 +2,6 @@ import abc
import select
import socket
from time import sleep
import sys
from threading import Thread
from typing import NamedTuple, NoReturn
@@ -40,7 +39,6 @@ class VbanCmd(abc.ABC):
)
self._register_rt_header = RegisterRTHeader()
self.expected_packet = VBAN_VMRT_Packet_Header()
self.buff = None
self._rt_register_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self._rt_packet_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@@ -52,12 +50,6 @@ class VbanCmd(abc.ABC):
self.ready_to_read, self.ready_to_write, in_error = select.select(is_readable, is_writable, is_error, 60)
def __enter__(self):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
res = s.connect_ex((self._ip, self._port))
s.close()
if res:
raise VMCMDErrors('Could not connect to remote Voicemeeter')
self._rt_packet_socket.bind((socket.gethostbyname(socket.gethostname()), self._port))
worker = Thread(target=self._send_register_rt, daemon=True)
worker.start()
@@ -107,8 +99,7 @@ class VbanCmd(abc.ABC):
@property
def public_packet(self):
self.buff = self._get_rt()
return self.buff
return self._get_rt()
def _get_rt(self):
def fget():
@@ -116,8 +107,7 @@ class VbanCmd(abc.ABC):
while not data:
data = self._fetch_rt_packet()
return data
private_packet = fget()
return private_packet if private_packet == self.buff else fget()
return fget()
def set_rt(self, id_, param, val):
cmd = f'{id_}.{param}={val}'
@@ -150,9 +140,9 @@ class VbanCmd(abc.ABC):
self.set_rt('Command', 'Restart', 1)
def close(self):
sleep(self._delay)
self._rt_register_socket.close()
self._sendrequest_string_socket.close()
self._rt_packet_socket.close()
sys.exit()
def __exit__(self, exc_type, exc_value, exc_traceback):
self.close()