Candidate conversion of test_01_con_discon_success to py.test

Signed-off-by: NJames Myatt <james@jamesmyatt.co.uk>

tests/test_client.py

+import os
+import sys
+import inspect
+
+import paho.mqtt.client as mqtt
+
+# From http://stackoverflow.com/questions/279237/python-import-a-module-from-a-folder
+cmd_subfolder = os.path.realpath(
+    os.path.abspath(
+        os.path.join(
+            os.path.split(
+                inspect.getfile(inspect.currentframe()))[0],
+            '..', 'test')))
+if cmd_subfolder not in sys.path:
+    sys.path.insert(0, cmd_subfolder)
+import paho_test
+
+# Import test fixture
+from testsupport.broker import fake_broker
+
+
+class Test_connect:
+    def test_01_con_discon_success(self, fake_broker):
+        mqttc = mqtt.Client(
+            "01-con-discon-success", protocol=mqtt.MQTTv31)
+
+        def on_connect(mqttc, obj, flags, rc):
+            assert rc == 0
+            mqttc.disconnect()
+
+        mqttc.on_connect = on_connect
+
+        mqttc.connect_async("localhost", 1888)
+        mqttc.loop_start()
+
+        fake_broker.start()
+
+        connect_packet = paho_test.gen_connect(
+            "01-con-discon-success", keepalive=60,
+            proto_name="MQIsdp", proto_ver=3)
+        fake_broker.expect_packet(connect_packet)
+
+        connack_packet = paho_test.gen_connack(rc=0)
+        fake_broker.send_packet(connack_packet)
+
+        disconnect_packet = paho_test.gen_disconnect()
+        fake_broker.expect_packet(disconnect_packet)
+
+        mqttc.loop_stop()

tests/testsupport/broker.py

+import socket
+
+import pytest
+
+
+class FakeBroker:
+    def __init__(self):
+        # Bind to "localhost" for maximum performance, as described in:
+        # http://docs.python.org/howto/sockets.html#ipc
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        sock.settimeout(30)
+        sock.bind(("localhost", 1888))
+        sock.listen(1)
+
+        self._sock = sock
+        self._conn = None
+
+    def start(self):
+        if self._sock is None:
+            raise ValueError('Socket is not open')
+
+        (conn, address) = self._sock.accept()
+        conn.settimeout(10)
+        self._conn = conn
+
+    def finish(self):
+        if self._conn is not None:
+            self._conn.close()
+            self._conn = None
+
+        if self._sock is not None:
+            self._sock.close()
+            self._sock = None
+
+    def expect_packet(self, packet_expected):
+        if self._conn is None:
+            raise ValueError('Connection is not open')
+
+        packet_in = self._conn.recv(len(packet_expected))
+        if not packet_in:
+            raise RuntimeError('Connection was closed')
+        assert packet_in == packet_expected
+
+    def send_packet(self, packet_out):
+        if self._conn is None:
+            raise ValueError('Connection is not open')
+
+        count = self._conn.send(packet_out)
+        if not count:
+            raise RuntimeError('Connection was closed')
+
+
+@pytest.fixture
+def fake_broker():
+    # print('Setup server')
+    server = FakeBroker()
+
+    yield server
+
+    # print('Teardown server')
+    server.finish()