tests: Update helper functions for new virtual-image features

This also changes the code to keep the connection open and adds
automatic mainloop iteration to ensure the driver processes the request.
This is important so we will not deadlock when we send multiple
requests.
This commit is contained in:
Benjamin Berg 2019-12-04 19:58:26 +01:00 committed by Benjamin Berg
parent 3e958ab7b4
commit 50a837573d

View file

@ -24,20 +24,6 @@ if wrapper:
os.unsetenv('LIBFPRINT_TEST_WRAPPER') os.unsetenv('LIBFPRINT_TEST_WRAPPER')
sys.exit(subprocess.check_call(wrap_cmd)) sys.exit(subprocess.check_call(wrap_cmd))
class Connection:
def __init__(self, addr):
self.addr = addr
def __enter__(self):
self.con = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.con.connect(self.addr)
return self.con
def __exit__(self, exc_type, exc_val, exc_tb):
self.con.close()
del self.con
def load_image(img): def load_image(img):
png = cairo.ImageSurface.create_from_png(img) png = cairo.ImageSurface.create_from_png(img)
@ -101,24 +87,51 @@ class VirtualImage(unittest.TestCase):
def setUp(self): def setUp(self):
self.dev.open_sync() self.dev.open_sync()
self.con = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.con.connect(self.sockaddr)
def tearDown(self): def tearDown(self):
self.con.close()
del self.con
self.dev.close_sync() self.dev.close_sync()
def report_finger(self, state): def send_retry(self, retry_error=1, iterate=True):
with Connection(self.sockaddr) as con: # The default (1) is too-short
con.write(struct.pack('ii', -1, 1 if state else 0)) self.sendall(struct.pack('ii', -1, retry_error))
while iterate and ctx.pending():
ctx.iteration(False)
def send_image(self, image): def send_error(self, device_error=0, iterate=True):
# The default (0) is a generic error
self.sendall(struct.pack('ii', -1, retry_error))
while iterate and ctx.pending():
ctx.iteration(False)
def send_finger_automatic(self, automatic, iterate=True):
# Set whether finger on/off is reported around images
self.con.sendall(struct.pack('ii', -3, 1 if automatic else 0))
while iterate and ctx.pending():
ctx.iteration(False)
def send_finger_report(self, has_finger, iterate=True):
# Send finger on/off
self.con.sendall(struct.pack('ii', -4, 1 if has_finger else 0))
while iterate and ctx.pending():
ctx.iteration(False)
def send_image(self, image, iterate=True):
img = self.prints[image] img = self.prints[image]
with Connection(self.sockaddr) as con:
mem = img.get_data()
mem = mem.tobytes()
assert len(mem) == img.get_width() * img.get_height()
encoded_img = struct.pack('ii', img.get_width(), img.get_height()) mem = img.get_data()
encoded_img += mem mem = mem.tobytes()
assert len(mem) == img.get_width() * img.get_height()
con.sendall(encoded_img) encoded_img = struct.pack('ii', img.get_width(), img.get_height())
encoded_img += mem
self.con.sendall(encoded_img)
while iterate and ctx.pending():
ctx.iteration(False)
def test_capture_prevents_close(self): def test_capture_prevents_close(self):
cancel = Gio.Cancellable() cancel = Gio.Cancellable()