<commit_before>import os
import unittest
import canopen
import logging

# logging.basicConfig(level=logging.DEBUG)


EDS_PATH = os.path.join(os.path.dirname(__file__), 'sample.eds')


class TestSDO(unittest.TestCase):
    """
    Test SDO client and server against each other.
    """

    @classmethod
    def setUpClass(cls):
        cls.network1 = canopen.Network()
        cls.network1.connect("test", bustype="virtual")
        cls.remote_node = cls.network1.add_node(2, EDS_PATH)

        cls.network2 = canopen.Network()
        cls.network2.connect("test", bustype="virtual")
        cls.local_node = cls.network2.create_node(2, EDS_PATH)

    @classmethod
    def tearDownClass(cls):
        cls.network1.disconnect()
        cls.network2.disconnect()

    def test_expedited_upload(self):
        self.local_node.sdo[0x1400][1].raw = 0x99
        vendor_id = self.remote_node.sdo[0x1400][1].raw
        self.assertEqual(vendor_id, 0x99)

    def test_block_upload_switch_to_expedite_upload(self):
        with self.assertRaises(canopen.SdoCommunicationError) as context:
            self.remote_node.sdo[0x1008].open('r', block_transfer=True)
        # We get this since the sdo client don't support the switch
        # from block upload to expedite upload
        self.assertTrue("Unexpected response 0x41" in context.exception)

    def test_block_download_not_supported(self):
        data = b"TEST DEVICE"
        with self.assertRaises(canopen.SdoAbortedError) as context:
            self.remote_node.sdo[0x1008].open('wb',
                                              size=len(data),
                                              block_transfer=True)
        self.assertEqual(context.exception.code, 0x05040001)

    def test_expedited_upload_default_value_visible_string(self):
        device_name = self.remote_node.sdo["Manufacturer device name"].raw
        self.assertEqual(device_name, b"TEST DEVICE")

    def test_expedited_upload_default_value_real(self):
        sampling_rate = self.remote_node.sdo["Sensor Sampling Rate (Hz)"].raw
        self.assertAlmostEqual(sampling_rate, 5.2, places=2)

    def test_segmented_upload(self):
        self.local_node.sdo["Manufacturer device name"].raw = "Some cool device"
        device_name = self.remote_node.sdo["Manufacturer device name"].data
        self.assertEqual(device_name, b"Some cool device")

    def test_expedited_download(self):
        self.remote_node.sdo["Identity object"]["Vendor-ID"].raw = 0xfeff
        vendor_id = self.local_node.sdo["Identity object"]["Vendor-ID"].raw
        self.assertEqual(vendor_id, 0xfeff)

    def test_segmented_download(self):
        self.remote_node.sdo["Manufacturer device name"].raw = "Another cool device"
        device_name = self.local_node.sdo["Manufacturer device name"].data<commit_msg>Removed 'b since raw is a decoded string<commit_after>import os
import unittest
import canopen
import logging

# logging.basicConfig(level=logging.DEBUG)


EDS_PATH = os.path.join(os.path.dirname(__file__), 'sample.eds')


class TestSDO(unittest.TestCase):
    """
    Test SDO client and server against each other.
    """

    @classmethod
    def setUpClass(cls):
        cls.network1 = canopen.Network()
        cls.network1.connect("test", bustype="virtual")
        cls.remote_node = cls.network1.add_node(2, EDS_PATH)

        cls.network2 = canopen.Network()
        cls.network2.connect("test", bustype="virtual")
        cls.local_node = cls.network2.create_node(2, EDS_PATH)

    @classmethod
    def tearDownClass(cls):
        cls.network1.disconnect()
        cls.network2.disconnect()

    def test_expedited_upload(self):
        self.local_node.sdo[0x1400][1].raw = 0x99
        vendor_id = self.remote_node.sdo[0x1400][1].raw
        self.assertEqual(vendor_id, 0x99)

    def test_block_upload_switch_to_expedite_upload(self):
        with self.assertRaises(canopen.SdoCommunicationError) as context:
            self.remote_node.sdo[0x1008].open('r', block_transfer=True)
        # We get this since the sdo client don't support the switch
        # from block upload to expedite upload
        self.assertTrue("Unexpected response 0x41" in context.exception)

    def test_block_download_not_supported(self):
        data = b"TEST DEVICE"
        with self.assertRaises(canopen.SdoAbortedError) as context:
            self.remote_node.sdo[0x1008].open('wb',
                                              size=len(data),
                                              block_transfer=True)
        self.assertEqual(context.exception.code, 0x05040001)

    def test_expedited_upload_default_value_visible_string(self):
        device_name = self.remote_node.sdo["Manufacturer device name"].raw
        self.assertEqual(device_name, "TEST DEVICE")

    def test_expedited_upload_default_value_real(self):
        sampling_rate = self.remote_node.sdo["Sensor Sampling Rate (Hz)"].raw
        self.assertAlmostEqual(sampling_rate, 5.2, places=2)

    def test_segmented_upload(self):
        self.local_node.sdo["Manufacturer device name"].raw = "Some cool device"
        device_name = self.remote_node.sdo["Manufacturer device name"].data
        self.assertEqual(device_name, b"Some cool device")

    def test_expedited_download(self):
        self.remote_node.sdo["Identity object"]["Vendor-ID"].raw = 0xfeff
        vendor_id = self.local_node.sdo["Identity object"]["Vendor-ID"].raw
        self.assertEqual(vendor_id, 0xfeff)

    def test_segmented_download(self):
        self.remote_node.sdo["Manufacturer device name"].raw = "Another cool device"
        device_name = self.local_node.sdo["Manufacturer device name"].data
{'fraction_non_alphanumeric': 0.10388007054673722, 'fraction_numerical': 0.019400352733686066, 'mean_word_length': 3.1153846153846154, 'pattern_counts': {'":': 0, '<': 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': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2926962', 'n_tokens_mistral': 1802, 'n_tokens_neox': 1726, 'n_words': 312}