<commit_before>from __future__ import absolute_import

import os
import time
import unittest
from nose.tools import eq_, ok_
import binascii

try:
    from Queue import Queue, Empty
except ImportError:
    # Python 3
    from queue import Queue, Empty

from openxc.tools.common import configure_logging
from openxc.interface import UsbVehicleInterface, BluetoothVehicleInterface

SOURCE = None

def setUpModule():
    configure_logging()

    # A bit of a hack to let us pass the product ID in at the command line, so
    # we can have 2 devices attached for testing at a time so it's more
    # automated. Set the VI_FUNC_TESTS_USB_PRODUCT_ID environment variable to a
    # number you want to use for the product ID.
    usb_product_id = os.getenv('VI_FUNC_TESTS_USB_PRODUCT_ID', None)
    use_bluetooth = os.getenv('VI_FUNC_TESTS_USE_BLUETOOTH', False)

    global SOURCE
    if use_bluetooth is not False:
        SOURCE = BluetoothVehicleInterface(payload_format="json")
    else:
        SOURCE = UsbVehicleInterface(payload_format="json", product_id=usb_product_id)
    SOURCE.start()

class ViFunctionalTests(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        super(ViFunctionalTests, cls).setUpClass()
        global SOURCE
        cls.vi = SOURCE
        cls.vi.callback = cls.receive

    def setUp(self):
        self.bus = 1
        self.message_id = 0x42
        self.data = "0x1234"
        ViFunctionalTests.can_message_queue = Queue()
        ViFunctionalTests.simple_vehicle_message_queue = Queue()
        ViFunctionalTests.diagnostic_response_queue = Queue()
        self.vi.set_acceptance_filter_bypass(1, True)
        self.vi.set_acceptance_filter_bypass(2, False)

    @classmethod
    def receive(cls, message, **kwargs):
        if ('message_id' in message and 'bus' in message and 'data' in message and
                getattr(cls, 'can_message_queue', None)):
            cls.can_message_queue.put(message)
        elif ('name' in message and 'value' in message):
            cls.simple_vehicle_message_queue.put(message)
        elif ('message_id' in message and 'bus' in message and 'mode' in message):
            cls.diagnostic_response_queue.put(message)


class ProtobufBaseTests(ViFunctionalTests):
    @classmethod
    def setUpClass(cls):
        super(ProtobufBaseTests, cls).setUpClass()
        cls.vi.set_payload_format("protobuf")

class JsonBaseTests(ViFunctionalTests):
    @classmethod
    def setUpClass(cls):
        super(JsonBaseTests, cls).setUpClass()
        cls.vi.set_payload_format("json")


class ControlCommandTests(object):
    def test_version(self):
        # TODO it'd be nice to read this from src/version.c
        eq_(self.vi.version(), "6.0.4-dev (functional_tests)")

    def test_device_id(self):
        device_id = self.vi.device_id()
        ok_(device_id is not None)
        if device_id != "Unknown":
            eq_(len(device_id), 12)

class ControlCommandTestsJson(ProtobufBaseTests, ControlCommandTests):
    pass

class ControlCommandTestsProtobuf(ProtobufBaseTests, ControlCommandTests):
    pass

class CanMessageTests(object):

    def _check_received_message(self, message):
        eq_(self.message_id, message['message_id'])
        eq_(self.bus, message['bus'])<commit_msg>Update firmware version in functional test suite.<commit_after>from __future__ import absolute_import

import os
import time
import unittest
from nose.tools import eq_, ok_
import binascii

try:
    from Queue import Queue, Empty
except ImportError:
    # Python 3
    from queue import Queue, Empty

from openxc.tools.common import configure_logging
from openxc.interface import UsbVehicleInterface, BluetoothVehicleInterface

SOURCE = None

def setUpModule():
    configure_logging()

    # A bit of a hack to let us pass the product ID in at the command line, so
    # we can have 2 devices attached for testing at a time so it's more
    # automated. Set the VI_FUNC_TESTS_USB_PRODUCT_ID environment variable to a
    # number you want to use for the product ID.
    usb_product_id = os.getenv('VI_FUNC_TESTS_USB_PRODUCT_ID', None)
    use_bluetooth = os.getenv('VI_FUNC_TESTS_USE_BLUETOOTH', False)

    global SOURCE
    if use_bluetooth is not False:
        SOURCE = BluetoothVehicleInterface(payload_format="json")
    else:
        SOURCE = UsbVehicleInterface(payload_format="json", product_id=usb_product_id)
    SOURCE.start()

class ViFunctionalTests(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        super(ViFunctionalTests, cls).setUpClass()
        global SOURCE
        cls.vi = SOURCE
        cls.vi.callback = cls.receive

    def setUp(self):
        self.bus = 1
        self.message_id = 0x42
        self.data = "0x1234"
        ViFunctionalTests.can_message_queue = Queue()
        ViFunctionalTests.simple_vehicle_message_queue = Queue()
        ViFunctionalTests.diagnostic_response_queue = Queue()
        self.vi.set_acceptance_filter_bypass(1, True)
        self.vi.set_acceptance_filter_bypass(2, False)

    @classmethod
    def receive(cls, message, **kwargs):
        if ('message_id' in message and 'bus' in message and 'data' in message and
                getattr(cls, 'can_message_queue', None)):
            cls.can_message_queue.put(message)
        elif ('name' in message and 'value' in message):
            cls.simple_vehicle_message_queue.put(message)
        elif ('message_id' in message and 'bus' in message and 'mode' in message):
            cls.diagnostic_response_queue.put(message)


class ProtobufBaseTests(ViFunctionalTests):
    @classmethod
    def setUpClass(cls):
        super(ProtobufBaseTests, cls).setUpClass()
        cls.vi.set_payload_format("protobuf")

class JsonBaseTests(ViFunctionalTests):
    @classmethod
    def setUpClass(cls):
        super(JsonBaseTests, cls).setUpClass()
        cls.vi.set_payload_format("json")


class ControlCommandTests(object):
    def test_version(self):
        # TODO it'd be nice to read this from src/version.c
        eq_(self.vi.version(), "7.0.0-dev (functional_tests)")

    def test_device_id(self):
        device_id = self.vi.device_id()
        ok_(device_id is not None)
        if device_id != "Unknown":
            eq_(len(device_id), 12)

class ControlCommandTestsJson(ProtobufBaseTests, ControlCommandTests):
    pass

class ControlCommandTestsProtobuf(ProtobufBaseTests, ControlCommandTests):
    pass

class CanMessageTests(object):

    def _check_received_message(self, message):
        eq_(self.message_id, message['message_id'])
        eq_(self.bus, message['bus'])
{'fraction_non_alphanumeric': 0.08449635922330097, 'fraction_numerical': 0.0054611650485436895, 'mean_word_length': 3.4248322147651007, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7216645', 'n_tokens_mistral': 2034, 'n_tokens_neox': 1974, 'n_words': 509}