<commit_before>import time

import sys

from mpf.tests.MpfTestCase import MpfTestCase
from mpf.tests.loop import MockSerial, MockSocket


class MockLisySocket(MockSerial, MockSocket):

    def read(self, length):
        del length
        if not self.queue:
            return b""
        msg = self.queue.pop()
        return msg

    def read_ready(self):
        return bool(self.queue)

    def write_ready(self):
        return True

    def write(self, msg):
        if msg in self.permanent_commands and msg not in self.expected_commands:
            if self.permanent_commands[msg] is not None:
                self.queue.append(self.permanent_commands[msg])
            return len(msg)

        # print("Serial received: " + "".join("\\x%02x" % b for b in msg) + " len: " + str(len(msg)))
        if msg not in self.expected_commands:
            self.crashed = True
            print("Unexpected command: " + "".join("\\x%02x" % b for b in msg) + " len: " + str(len(msg)))
            raise AssertionError("Unexpected command: " + "".join("\\x%02x" % b for b in msg) +
                                 " len: " + str(len(msg)))

        if self.expected_commands[msg] is not None:
            self.queue.append(self.expected_commands[msg])

        del self.expected_commands[msg]
        return len(msg)

    def send(self, data):
        return self.write(data)

    def recv(self, size):
        return self.read(size)

    def __init__(self):
        super().__init__()
        self.name = "SerialMock"
        self.expected_commands = {}
        self.queue = []
        self.permanent_commands = {}
        self.crashed = False


class TestLisy(MpfTestCase):

    def getConfigFile(self):
        return 'config.yaml'

    def getMachinePath(self):
        return 'tests/machine_files/lisy/'

    def _mock_loop(self):
        self.clock.mock_serial("com1", self.serialMock)
        self.clock.mock_socket("localhost", 1234, self.serialMock)

    def tearDown(self):
        self.assertFalse(self.serialMock.crashed)
        super().tearDown()

    def get_platform(self):
        return False

    def _wait_for_processing(self):
        start = time.time()
        while self.serialMock.expected_commands and not self.serialMock.crashed and time.time() < start + 10:
            self.advance_time_and_run(.01)

    def setUp(self):
        self.expected_duration = 1.5
        self.serialMock = MockLisySocket()

        self.serialMock.permanent_commands = {
            b'\x29': b'\x7F',           # changed switches? -> no
            b'\x65': b'\x00'            # watchdog
        }

        self.serialMock.expected_commands = {
            b'\x00': b'LISY1\00',       # hw LISY1
            b'\x01': b'4.01\00',        # lisy version
            b'\x02': b'0.05\00',        # api version
            b'\x64': b'\x00',           # reset -> ok
            b'\x03': b'\x28',           # get number of lamps -> 40
            b'\x04': b'\x09',           # get number of solenoids -> 9
            b'\x06': b'\x05',           # get number of displays -> 5
            b'\x1e\x00': None,          # clear display
            b'\x1f\x00': None,          # clear display
            b'\x20\x00': None,          # clear display
        }

        for row in range(8):
            for col in range(8):
                self.serialMock.expected_commands[bytes([40, row * 10 + col])] = b'\x00'\
                    if row * 10 + col != 37 else b'\x01'

        super().setUp()

        self._wait_for_processing()
        self.assertFalse(self.serialMock.expected_commands)

    def test_platform(self):
        if sys.version_info[0] == 3 and sys.version_info[1] == 4:
            # this fails on python 3.4 because of some asyncio bugs
            return

        # wait for watchdog
        self.serialMock.expected_commands = {
            b'\x65': b'\x00'            # watchdog
        }
        self._wait_for_processing()

        # test initial switch state
        self.assertSwitchState("s_test00", False)
        self.assertSwitchState("s_test37", True)
        self.assertSwitchState("s_test77_nc", True)

        self.serialMock.expected_commands = {<commit_msg>skip lisy test early<commit_after>import time

import sys

from mpf.tests.MpfTestCase import MpfTestCase
from mpf.tests.loop import MockSerial, MockSocket


class MockLisySocket(MockSerial, MockSocket):

    def read(self, length):
        del length
        if not self.queue:
            return b""
        msg = self.queue.pop()
        return msg

    def read_ready(self):
        return bool(self.queue)

    def write_ready(self):
        return True

    def write(self, msg):
        if msg in self.permanent_commands and msg not in self.expected_commands:
            if self.permanent_commands[msg] is not None:
                self.queue.append(self.permanent_commands[msg])
            return len(msg)

        # print("Serial received: " + "".join("\\x%02x" % b for b in msg) + " len: " + str(len(msg)))
        if msg not in self.expected_commands:
            self.crashed = True
            print("Unexpected command: " + "".join("\\x%02x" % b for b in msg) + " len: " + str(len(msg)))
            raise AssertionError("Unexpected command: " + "".join("\\x%02x" % b for b in msg) +
                                 " len: " + str(len(msg)))

        if self.expected_commands[msg] is not None:
            self.queue.append(self.expected_commands[msg])

        del self.expected_commands[msg]
        return len(msg)

    def send(self, data):
        return self.write(data)

    def recv(self, size):
        return self.read(size)

    def __init__(self):
        super().__init__()
        self.name = "SerialMock"
        self.expected_commands = {}
        self.queue = []
        self.permanent_commands = {}
        self.crashed = False


class TestLisy(MpfTestCase):

    def getConfigFile(self):
        return 'config.yaml'

    def getMachinePath(self):
        return 'tests/machine_files/lisy/'

    def _mock_loop(self):
        self.clock.mock_serial("com1", self.serialMock)
        self.clock.mock_socket("localhost", 1234, self.serialMock)

    def tearDown(self):
        self.assertFalse(self.serialMock.crashed)
        super().tearDown()

    def get_platform(self):
        return False

    def _wait_for_processing(self):
        start = time.time()
        while self.serialMock.expected_commands and not self.serialMock.crashed and time.time() < start + 10:
            self.advance_time_and_run(.01)

    def setUp(self):
        if sys.version_info[0] == 3 and sys.version_info[1] == 4:
            # this fails on python 3.4 because of some asyncio bugs
            self.skipTest("Test is unstable in Python 3.4")
            return
        self.expected_duration = 1.5
        self.serialMock = MockLisySocket()

        self.serialMock.permanent_commands = {
            b'\x29': b'\x7F',           # changed switches? -> no
            b'\x65': b'\x00'            # watchdog
        }

        self.serialMock.expected_commands = {
            b'\x00': b'LISY1\00',       # hw LISY1
            b'\x01': b'4.01\00',        # lisy version
            b'\x02': b'0.05\00',        # api version
            b'\x64': b'\x00',           # reset -> ok
            b'\x03': b'\x28',           # get number of lamps -> 40
            b'\x04': b'\x09',           # get number of solenoids -> 9
            b'\x06': b'\x05',           # get number of displays -> 5
            b'\x1e\x00': None,          # clear display
            b'\x1f\x00': None,          # clear display
            b'\x20\x00': None,          # clear display
        }

        for row in range(8):
            for col in range(8):
                self.serialMock.expected_commands[bytes([40, row * 10 + col])] = b'\x00'\
                    if row * 10 + col != 37 else b'\x01'

        super().setUp()

        self._wait_for_processing()
        self.assertFalse(self.serialMock.expected_commands)

    def test_platform(self):
        # wait for watchdog
        self.serialMock.expected_commands = {
            b'\x65': b'\x00'            # watchdog
        }
        self._wait_for_processing()

        # test initial switch state
        self.assertSwitchState("s_test00", False)
        self.assertSwitchState("s_test37", True)
        self.assertSwitchState("s_test77_nc", True)

        self.serialMock.expected_commands = {
{'fraction_non_alphanumeric': 0.11898554159753497, 'fraction_numerical': 0.024887414079165678, 'mean_word_length': 1.859708573364961, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '79395', 'n_tokens_mistral': 2694, 'n_tokens_neox': 2524, 'n_words': 591}