Test load_config() (hardbyte#1396)

* listener_test: Do not modify environment for other tests

* network_test: Do not modify environment for other tests

Also do not skip the test since there will always be an interface (previously
set globally and now prepared in setUp function.

* test_load_file_config: Test with different values

Testing with the same interface values would not show possible issues in case
there were any.

* tests: Add test for load_config

Add tests for load_config with contexts and environment variables.

Closes hardbyte#345

Signed-off-by: Martin Kletzander <[email protected]>

Signed-off-by: Martin Kletzander <[email protected]>

Author: nertpinx

test/listener_test.py

@@ -15,9 +15,6 @@
 
 from .data.example_data import generate_message
 
-channel = "virtual_channel_0"
-can.rc["interface"] = "virtual"
-
 logging.basicConfig(level=logging.DEBUG)
 
 # makes the random number generator deterministic
@@ -55,10 +52,15 @@ def testClassesImportable(self):
 
 class BusTest(unittest.TestCase):
     def setUp(self):
+        # Save all can.rc defaults
+        self._can_rc = can.rc
+        can.rc = {"interface": "virtual"}
         self.bus = can.interface.Bus()
 
     def tearDown(self):
         self.bus.shutdown()
+        # Restore the defaults
+        can.rc = self._can_rc
 
 
 class ListenerTest(BusTest):

test/network_test.py

@@ -14,10 +14,8 @@
 import can
 
 channel = "vcan0"
-can.rc["interface"] = "virtual"
 
 
-@unittest.skipIf("interface" not in can.rc, "Need a CAN interface")
 class ControllerAreaNetworkTestCase(unittest.TestCase):
     """
     This test ensures that what messages go in to the bus is what comes out.
@@ -42,6 +40,15 @@ class ControllerAreaNetworkTestCase(unittest.TestCase):
         for b in range(num_messages)
     )
 
+    def setUp(self):
+        # Save all can.rc defaults
+        self._can_rc = can.rc
+        can.rc = {"interface": "virtual"}
+
+    def tearDown(self):
+        # Restore the defaults
+        can.rc = self._can_rc
+
     def producer(self, ready_event, msg_read):
         self.client_bus = can.interface.Bus(channel=channel)
         ready_event.wait()

test/test_load_config.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python
+
+import os
+import shutil
+import tempfile
+import unittest
+from tempfile import NamedTemporaryFile
+
+import can
+
+
+class LoadConfigTest(unittest.TestCase):
+    configuration = {
+        "default": {"interface": "serial", "channel": "0"},
+        "one": {"interface": "kvaser", "channel": "1", "bitrate": 100000},
+        "two": {"channel": "2"},
+    }
+
+    def setUp(self):
+        # Create a temporary directory
+        self.test_dir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        # Remove the directory after the test
+        shutil.rmtree(self.test_dir)
+
+    def _gen_configration_file(self, sections):
+        with NamedTemporaryFile(
+            mode="w", dir=self.test_dir, delete=False
+        ) as tmp_config_file:
+            content = []
+            for section in sections:
+                content.append("[{}]".format(section))
+                for k, v in self.configuration[section].items():
+                    content.append("{} = {}".format(k, v))
+            tmp_config_file.write("\n".join(content))
+            return tmp_config_file.name
+
+    def _dict_to_env(self, d):
+        return {f"CAN_{k.upper()}": str(v) for k, v in d.items()}
+
+    def test_config_default(self):
+        tmp_config = self._gen_configration_file(["default"])
+        config = can.util.load_config(path=tmp_config)
+        self.assertEqual(config, self.configuration["default"])
+
+    def test_config_whole_default(self):
+        tmp_config = self._gen_configration_file(self.configuration)
+        config = can.util.load_config(path=tmp_config)
+        self.assertEqual(config, self.configuration["default"])
+
+    def test_config_whole_context(self):
+        tmp_config = self._gen_configration_file(self.configuration)
+        config = can.util.load_config(path=tmp_config, context="one")
+        self.assertEqual(config, self.configuration["one"])
+
+    def test_config_merge_context(self):
+        tmp_config = self._gen_configration_file(self.configuration)
+        config = can.util.load_config(path=tmp_config, context="two")
+        expected = self.configuration["default"]
+        expected.update(self.configuration["two"])
+        self.assertEqual(config, expected)
+
+    def test_config_merge_environment_to_context(self):
+        tmp_config = self._gen_configration_file(self.configuration)
+        env_data = {"interface": "serial", "bitrate": 125000}
+        env_dict = self._dict_to_env(env_data)
+        with unittest.mock.patch.dict("os.environ", env_dict):
+            config = can.util.load_config(path=tmp_config, context="one")
+        expected = self.configuration["one"]
+        expected.update(env_data)
+        self.assertEqual(config, expected)
+
+    def test_config_whole_environment(self):
+        tmp_config = self._gen_configration_file(self.configuration)
+        env_data = {"interface": "socketcan", "channel": "3", "bitrate": 250000}
+        env_dict = self._dict_to_env(env_data)
+        with unittest.mock.patch.dict("os.environ", env_dict):
+            config = can.util.load_config(path=tmp_config, context="one")
+        expected = self.configuration["one"]
+        expected.update(env_data)
+        self.assertEqual(config, expected)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_load_file_config.py

@@ -11,7 +11,7 @@
 class LoadFileConfigTest(unittest.TestCase):
     configuration = {
         "default": {"interface": "virtual", "channel": "0"},
-        "one": {"interface": "virtual", "channel": "1"},
+        "one": {"interface": "kvaser", "channel": "1"},
         "two": {"channel": "2"},
         "three": {"extra": "extra value"},
     }