[gnome-keysign: 51/65] test_bluetooth: add corrupted key test

[Gitlab System User <gitlab src gnome org>]

commit 03c9c295292219922faca9dc3e22360247cf7f0e
Author: RyuzakiKK <aasonykk gmail com>
Date:   Mon Aug 28 10:39:19 2017 +0200

    test_bluetooth: add corrupted key test

 tests/test_bluetooth.py | 54 +++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 54 insertions(+)
---
diff --git a/tests/test_bluetooth.py b/tests/test_bluetooth.py
index 51b9569..3d7ff3a 100644
--- a/tests/test_bluetooth.py
+++ b/tests/test_bluetooth.py
@@ -1,10 +1,13 @@
 import os
 import logging
+import select
+import socket
 import gi
 gi.require_version('Gtk', '3.0')
 
 from nose.twistedtools import deferred
 from nose.tools import *
+from twisted.internet import threads
 from twisted.internet.defer import inlineCallbacks
 
 from keysign.bluetoothoffer import BluetoothOffer
@@ -91,3 +94,54 @@ def test_bt_wrong_mac():
     assert_is_none(downloaded_key_data)
     assert_false(success)
     assert_equal(error.args[0], "(112, 'Host is down')")
+
+
+@deferred(timeout=15)
+@inlineCallbacks
+def test_bt_corrupted_key():
+    """This test requires two working Bluetooth devices"""
+
+    @inlineCallbacks
+    def start(bo):
+        success = False
+        try:
+            while not success:
+                # server_socket.accept() is not stoppable. So with select we can call accept()
+                # only when we are sure that there is already a waiting connection
+                ready_to_read, ready_to_write, in_error = yield threads.deferToThread(
+                    select.select, [bo.server_socket], [], [], 0.5)
+                if ready_to_read:
+                    # We are sure that a connection is available, so we can call
+                    # accept() without deferring it to a thread
+                    client_socket, address = bo.server_socket.accept()
+                    key_data = get_public_key_data(bo.key.fingerprint)
+                    kd_decoded = key_data.decode('utf-8')
+                    # We send only a part of the key. In this way we can simulate the case
+                    # where the connection has been lost
+                    half = len(kd_decoded)/2
+                    kd_corrupted = kd_decoded[:half]
+                    yield threads.deferToThread(client_socket.sendall, kd_corrupted)
+                    client_socket.shutdown(socket.SHUT_RDWR)
+                    client_socket.close()
+                    success = True
+        except Exception as e:
+            log.error("An error occurred: %s" % e)
+
+    data = read_fixture_file("seckey-no-pw-1.asc")
+    key = openpgpkey_from_data(data)
+    log.info("Running with key %r", key)
+    file_key_data = get_public_key_data(key.fingerprint)
+    hmac = mac_generate(key.fingerprint.encode('ascii'), file_key_data)
+    # Start offering the key
+    offer = BluetoothOffer(key)
+    data = yield offer.allocate_code()
+    # getting the code from "BT=code;...."
+    code = data.split("=", 1)[1]
+    code = code.split(";", 1)[0]
+    port = int(data.rsplit("=", 1)[1])
+    start(offer)
+    receive = BluetoothReceive(port)
+    msg_tuple = yield receive.find_key(code, hmac)
+    downloaded_key_data, result, error = msg_tuple
+    assert_false(result)
+    assert_equal(type(error), ValueError)