DoorAdmin/tests/test_door_handle.py

import datetime
import time

import pytest
from pathlib import Path
from imaginaerraum_door_admin.door_handle import DoorHandle
import paho.mqtt.client as mqtt


@pytest.fixture
def nfc_socket(tmp_path):
    # mock up a Unix socket server for testing
    import threading
    import socketserver

    class ThreadedRequestHandler(socketserver.BaseRequestHandler):
        def handle(self):
            data = str(self.request.recv(1024), 'ascii')
            cur_thread = threading.current_thread()
            response = bytes("{}: {}".format(cur_thread.name, data), 'ascii')
            try:
                self.request.sendall(response)
            except BrokenPipeError:
                pass

    class TreadedUnixServer(socketserver.ThreadingMixIn,
                            socketserver.UnixStreamServer):
        pass

    nfc_socket_file = tmp_path / 'nfc.sock'
    server = TreadedUnixServer(str(nfc_socket_file), ThreadedRequestHandler)
    with server:
        nfc_socket = server.server_address

        # Start a thread with the server -- that thread will then start one
        # more thread for each request
        server_thread = threading.Thread(target=server.serve_forever)
        # Exit the server thread when the main thread terminates
        server_thread.daemon = True
        server_thread.start()
        print("Server loop running in thread:", server_thread.name)

        yield nfc_socket

        server.shutdown()


def test_create_door_handle(nfc_socket, tmp_path):
    # test with invalid token file
    with pytest.raises(FileNotFoundError):
        door_handle = DoorHandle(token_file='no_such_file',
                                 mqtt_host='test.mosquitto.org')

    token_file = tmp_path / 'door_tokens'
    token_file.write_text('')
    door_handle = DoorHandle(
        token_file=token_file, mqtt_host='test.mosquitto.org',
        nfc_socket=nfc_socket
    )
    door_handle.nfc_sock.close()


@pytest.fixture
def door_handle(nfc_socket, tmp_path):
    token_file = tmp_path / 'door_tokens'
    token_file.write_text(
        """# token | name | organization | email | valid_thru
        04387cfa186280|Gandalf|Wizards|gandalf@shire.me|
        043a81fa186280|Bilbo|Hobbits|bilbo@shire.me|
        #04538cfa186280|Gimli|Dwarves|gimli@shire.me|
        """
    )
    door_handle = DoorHandle(
        token_file=token_file, mqtt_host='test.mosquitto.org',
        nfc_socket=nfc_socket
    )
    yield door_handle
    door_handle.nfc_sock.close()


@pytest.fixture
def door_handle_no_nfc(tmp_path):
    token_file = tmp_path / 'door_tokens'
    token_file.write_text('')
    door_handle = DoorHandle(
        token_file=token_file, mqtt_host='test.mosquitto.org',
    )
    yield door_handle


def test_store_tokens(door_handle):
    new_tokens = {
       '042979fa186280': {
           'name': 'Pippin',
            'organization': 'Hobbits',
            'email': 'pippin@shire.me',
            'valid_thru': None,
            'inactive': False
       }
    }

    door_handle.store_tokens(new_tokens)

    token_file = Path(door_handle.token_file)
    assert token_file.exists()
    assert '042979fa186280' in token_file.read_text()


def test_mqtt_messages(door_handle):
    # test sending messages to the door_handle via MQTT and see if the state
    # gets updated accordingly
    client = mqtt.Client()

    con = client.connect('test.mosquitto.org', 1883)
    client.publish('door/position/value', 10).wait_for_publish(1)
    time.sleep(1)
    assert door_handle.encoder_position == 10

    client.publish('door/state/value', 'open').wait_for_publish(1)
    time.sleep(1)
    assert door_handle.state == 'open'

    timestamp = datetime.datetime.now().replace(microsecond=0)
    token = '042979fa186280'
    client.publish(
        'door/token/last_invalid',
        f"{timestamp.strftime('%Y-%m-%d %H:%M:%S')};{token}"
    ).wait_for_publish(1)
    time.sleep(1)
    most_recent_token = door_handle.get_most_recent_token()
    assert most_recent_token['timestamp'] == timestamp
    assert most_recent_token['token'] == token

    client.disconnect()


def test_open_door(door_handle):
    door_handle.open_door()


def test_open_door_broken_socket(door_handle_no_nfc):
    # test broken nfc_socket connection
    with pytest.raises(RuntimeError):
        door_handle_no_nfc.open_door()


def test_close_door(door_handle):
    door_handle.close_door()


def test_close_door_broken_socket(door_handle_no_nfc):
    # test broken nfc_socket connection
    with pytest.raises(RuntimeError):
        door_handle_no_nfc.close_door()