Added test framework with mocked socket connection

Added test framework with mocked socket connection

Author: Axel Dahlberg

.travis.yml

@@ -23,6 +23,7 @@ install:
 - pip3 install dist/*.whl
 script:
 - make lint
+- make tests
 deploy:
   provider: pypi
   user: adahlberg

Makefile

@@ -17,6 +17,9 @@ python-deps:
 _verified:
 	@echo "CQC-Python is verified!"
 
+tests:
+	@${PYTHON} -m pytest tests
+
 verify: clean python-deps lint _verified
 
 _remove_build:
@@ -35,4 +38,4 @@ _build:
 
 build: _clear_build _build
 
-.PHONY: clean lint python-deps verify build
+.PHONY: clean lint python-deps verify build tests

cqc/pythonLib.py

@@ -674,7 +674,7 @@ def allocate_qubits(self, num_qubits, notify=True, block=True):
 
         return qubits
 
-    def release_qubits(self, qubits, notify=True, block=False, action=False):
+    def release_qubits(self, qubits, notify=True, block=True, action=False):
         """
         Release qubits so backend can free them up for other uses
         :param qubits: a list of qubits to be released
@@ -2339,7 +2339,7 @@ def reset(self, notify=True, block=True):
                 message = self._cqc.readMessage()
                 self._cqc.print_CQC_msg(message)
 
-    def release(self, notify=True, block=False):
+    def release(self, notify=True, block=True):
         """
         Release the current qubit
         :param notify: Do we wish to be notified when done

tests/conftest.py

@@ -0,0 +1,102 @@
+import pytest
+import inspect
+import socket
+from collections import namedtuple
+
+from cqc.pythonLib import CQCConnection
+
+Call = namedtuple("Call", ["name", "args", "kwargs"])
+
+
+def _spy_wrapper(method):
+    """Wraps a method to be able to spy on it"""
+    def new_method(self, *args, **kwargs):
+        if method.__name__ == '__init__':
+            self.calls = []
+        call = Call(method.__name__, args, kwargs)
+        self.calls.append(call)
+        return method(self, *args, **kwargs)
+
+    return new_method
+
+
+def spy_on_class(cls):
+    """Spies on all calls to the methods of a class"""
+    for method_name, method in inspect.getmembers(cls, predicate=inspect.isfunction):
+        setattr(cls, method_name, _spy_wrapper(method))
+    return cls
+
+
+@spy_on_class
+class MockSocket:
+    def __init__(self, *args, **kwargs):
+        pass
+
+    def connect(self, *args, **kwargs):
+        pass
+
+    def send(self, *args, **kwargs):
+        pass
+
+    def recv(self, *args, **kwargs):
+        pass
+
+    def close(self, *args, **kwargs):
+        pass
+
+
+@pytest.fixture
+def mock_socket(monkeypatch):
+    def get_mocked_socket(*args, **kwargs):
+        mock_socket = MockSocket(*args, **kwargs)
+        return mock_socket
+
+    monkeypatch.setattr(socket, "socket", get_mocked_socket)
+
+
+class MockedFirstMessage:
+    """Mocks the second header returned by CQCConnection.readMessage"""
+    class MockedTypeEntry:
+        def __eq__(self, other):
+            """This type will be equal to any integer."""
+            return isinstance(other, int)
+
+    @property
+    def tp(self):
+        return self.MockedTypeEntry()
+
+
+class MockedOtherMessage:
+    """Mocks the second header returned by CQCConnection.readMessage"""
+    next_qubit_id = 0
+
+    @property
+    def qubit_id(self):
+        qid = self.next_qubit_id
+        self.next_qubit_id += 1
+        return qid
+
+    @property
+    def outcome(self):
+        return 0
+
+    @property
+    def datetime(self):
+        return 0
+
+
+@pytest.fixture
+def mock_read_message(monkeypatch):
+    """Mock the readMessage, check_error and print_CQC_msg from CQCConnection when testing."""
+    def mocked_readMessage(self):
+        return [MockedFirstMessage(), MockedOtherMessage()]
+
+    def mocked_print_CQC_msg(self, message):
+        pass
+
+    def mocked_check_error(self, hdr):
+        pass
+
+    monkeypatch.setattr(CQCConnection, "readMessage", mocked_readMessage)
+    monkeypatch.setattr(CQCConnection, "print_CQC_msg", mocked_print_CQC_msg)
+    monkeypatch.setattr(CQCConnection, "check_error", mocked_check_error)

tests/test_cqcconnection.py

@@ -0,0 +1,78 @@
+import pytest
+
+from cqc.pythonLib import CQCConnection, qubit
+from cqc.cqcHeader import CQCHeader, CQCCmdHeader, CQC_TP_COMMAND,\
+    CQC_CMD_HDR_LENGTH, CQC_CMD_H, CQC_CMD_NEW, CQC_CMD_RELEASE
+
+from utilities import get_header
+
+
+def get_expected_headers_simple_h():
+    """What headers we expect"""
+    hdr_tp_cmd = get_header(
+        CQCHeader, 
+        version=2,
+        tp=CQC_TP_COMMAND,
+        app_id=0,
+        length=CQC_CMD_HDR_LENGTH,
+    )
+    hdr_cmd_new = get_header(
+        CQCCmdHeader, 
+        qubit_id=0,
+        instr=CQC_CMD_NEW,
+        notify=True,
+        action=False,
+        block=True,
+    )
+    hdr_cmd_h = get_header(
+        CQCCmdHeader, 
+        qubit_id=0,
+        instr=CQC_CMD_H,
+        notify=True,
+        action=False,
+        block=True,
+    )
+    hdr_cmd_release = get_header(
+        CQCCmdHeader, 
+        qubit_id=0,
+        instr=CQC_CMD_RELEASE,
+        notify=True,
+        action=False,
+        block=True,
+    )
+
+    expected_headers = [
+        hdr_tp_cmd,
+        hdr_cmd_new,
+        hdr_tp_cmd,
+        hdr_cmd_h,
+        hdr_tp_cmd + hdr_cmd_release,
+    ]
+
+    return expected_headers
+
+
+def commands_to_apply_simple_h(cqc):
+    """What to do with the CQCConnection"""
+    q = qubit(cqc)
+    q.H()
+
+
+@pytest.mark.parametrize("commands_to_apply, get_expected_headers", [
+    (commands_to_apply_simple_h, get_expected_headers_simple_h),
+])
+def test_commands(commands_to_apply, get_expected_headers, monkeypatch, mock_socket, mock_read_message):
+
+    with CQCConnection("Test", socket_address=('localhost', 8000), use_classical_communication=False) as cqc:
+        commands_to_apply(cqc)
+
+    expected_headers = get_expected_headers()
+
+    commands_send = list(filter(lambda call: call.name == 'send', cqc._s.calls))
+    assert len(expected_headers) == len(commands_send)
+    for command, expected in zip(commands_send, expected_headers):
+        print(command.args[0])
+        print(expected)
+        print()
+        if expected is not None:
+            assert command.args[0] == expected

tests/utilities.py

@@ -0,0 +1,5 @@
+def get_header(header_class, *args, **kwargs):
+    """Construct and packs a given header"""
+    hdr = header_class()
+    hdr.setVals(*args, **kwargs)
+    return hdr.pack()