Minor config update and add tests.

test/config_tests/config_v1_test.py

@@ -1,5 +1,6 @@
 """Unit test for Configuration logic."""
 import asyncio
+import os
 import unittest
 from unittest.mock import patch
 
@@ -608,3 +609,55 @@ def test_config_file_error(self):
             mock_parse.side_effect = XKNXException()
             XKNX(config="xknx.yaml")
             self.assertEqual(mock_err.call_count, 1)
+
+    XKNX_GENERAL_OWN_ADDRESS = "1.2.3"
+    XKNX_GENERAL_RATE_LIMIT = "20"
+    XKNX_GENERAL_MULTICAST_GROUP = "1.2.3.4"
+    XKNX_GENERAL_MULTICAST_PORT = "1111"
+
+    def test_config_general_from_env(self):
+        os.environ["XKNX_GENERAL_OWN_ADDRESS"] = self.XKNX_GENERAL_OWN_ADDRESS
+        os.environ["XKNX_GENERAL_RATE_LIMIT"] = self.XKNX_GENERAL_RATE_LIMIT
+        os.environ["XKNX_GENERAL_MULTICAST_GROUP"] = self.XKNX_GENERAL_MULTICAST_GROUP
+        os.environ["XKNX_GENERAL_MULTICAST_PORT"] = self.XKNX_GENERAL_MULTICAST_PORT
+        self.xknx = XKNX(config="xknx.yaml")
+        os.unsetenv("XKNX_GENERAL_OWN_ADDRESS")
+        os.unsetenv("XKNX_GENERAL_RATE_LIMIT")
+        os.unsetenv("XKNX_GENERAL_MULTICAST_GROUP")
+        os.unsetenv("XKNX_GENERAL_MULTICAST_PORT")
+        self.assertEqual(str(self.xknx.own_address), self.XKNX_GENERAL_OWN_ADDRESS)
+        self.assertEqual(self.xknx.rate_limit, int(self.XKNX_GENERAL_RATE_LIMIT))
+        self.assertEqual(self.xknx.multicast_group, self.XKNX_GENERAL_MULTICAST_GROUP)
+        self.assertEqual(
+            self.xknx.multicast_port, int(self.XKNX_GENERAL_MULTICAST_PORT)
+        )
+
+    XKNX_CONNECTION_GATEWAY_IP = "192.168.12.34"
+    XKNX_CONNECTION_GATEWAY_PORT = "1234"
+    XKNX_CONNECTION_LOCAL_IP = "192.168.11.11"
+    XKNX_CONNECTION_ROUTE_BACK = "true"
+
+    def test_config_cnx_from_env(self):
+        os.environ["XKNX_CONNECTION_GATEWAY_IP"] = self.XKNX_CONNECTION_GATEWAY_IP
+        os.environ["XKNX_CONNECTION_GATEWAY_PORT"] = self.XKNX_CONNECTION_GATEWAY_PORT
+        os.environ["XKNX_CONNECTION_LOCAL_IP"] = self.XKNX_CONNECTION_LOCAL_IP
+        os.environ["XKNX_CONNECTION_ROUTE_BACK"] = self.XKNX_CONNECTION_ROUTE_BACK
+        self.xknx = XKNX(config="xknx.yaml")
+        os.unsetenv("XKNX_CONNECTION_GATEWAY_IP")
+        os.unsetenv("XKNX_CONNECTION_GATEWAY_PORT")
+        os.unsetenv("XKNX_CONNECTION_LOCAL_IP")
+        os.unsetenv("XKNX_CONNECTION_ROUTE_BACK")
+        self.assertEqual(
+            self.xknx.connection_config.gateway_ip, self.XKNX_CONNECTION_GATEWAY_IP
+        )
+        self.assertEqual(
+            self.xknx.connection_config.gateway_port,
+            int(self.XKNX_CONNECTION_GATEWAY_PORT),
+        )
+        self.assertEqual(
+            self.xknx.connection_config.local_ip, self.XKNX_CONNECTION_LOCAL_IP
+        )
+        self.assertEqual(
+            self.xknx.connection_config.route_back,
+            bool(self.XKNX_CONNECTION_ROUTE_BACK),
+        )

test/io_tests/connect_test.py

@@ -28,19 +28,31 @@ def tearDown(self):
         self.loop.close()
 
     def test_connect(self):
+        self.__test_connect()
+
+    def test_connect_route_back_true(self):
+        self.__test_connect(route_back=True)
+
+    def __test_connect(self, route_back: bool = False):
         """Test connecting from KNX bus."""
         xknx = XKNX()
         udp_client = UDPClient(xknx, ("192.168.1.1", 0), ("192.168.1.2", 1234))
-        connect = Connect(xknx, udp_client, route_back=False)
+        connect = Connect(xknx, udp_client, route_back=route_back)
         connect.timeout_in_seconds = 0
 
         self.assertEqual(connect.awaited_response_class, ConnectResponse)
 
         # Expected KNX/IP-Frame:
         exp_knxipframe = KNXIPFrame(xknx)
         exp_knxipframe.init(KNXIPServiceType.CONNECT_REQUEST)
-        exp_knxipframe.body.control_endpoint = HPAI(ip_addr="192.168.1.3", port=4321)
-        exp_knxipframe.body.data_endpoint = HPAI(ip_addr="192.168.1.3", port=4321)
+        if route_back:
+            exp_knxipframe.body.control_endpoint = HPAI()
+            exp_knxipframe.body.data_endpoint = HPAI()
+        else:
+            exp_knxipframe.body.control_endpoint = HPAI(
+                ip_addr="192.168.1.3", port=4321
+            )
+            exp_knxipframe.body.data_endpoint = HPAI(ip_addr="192.168.1.3", port=4321)
         exp_knxipframe.body.request_type = ConnectRequestType.TUNNEL_CONNECTION
         exp_knxipframe.normalize()
         with patch("xknx.io.UDPClient.send") as mock_udp_send, patch(

test/io_tests/connectionstate_test.py

@@ -27,12 +27,18 @@ def tearDown(self):
         self.loop.close()
 
     def test_connectionstate(self):
+        self.__test_connectionstate()
+
+    def test_connectionstate_route_back_true(self):
+        self.__test_connectionstate(route_back=True)
+
+    def __test_connectionstate(self, route_back: bool = False):
         """Test connectionstateing from KNX bus."""
         xknx = XKNX()
         communication_channel_id = 23
         udp_client = UDPClient(xknx, ("192.168.1.1", 0), ("192.168.1.2", 1234))
         connectionstate = ConnectionState(
-            xknx, udp_client, communication_channel_id, route_back=False
+            xknx, udp_client, communication_channel_id, route_back=route_back
         )
         connectionstate.timeout_in_seconds = 0
 
@@ -47,7 +53,12 @@ def test_connectionstate(self):
         exp_knxipframe = KNXIPFrame(xknx)
         exp_knxipframe.init(KNXIPServiceType.CONNECTIONSTATE_REQUEST)
         exp_knxipframe.body.communication_channel_id = communication_channel_id
-        exp_knxipframe.body.control_endpoint = HPAI(ip_addr="192.168.1.3", port=4321)
+        if route_back:
+            exp_knxipframe.body.control_endpoint = HPAI()
+        else:
+            exp_knxipframe.body.control_endpoint = HPAI(
+                ip_addr="192.168.1.3", port=4321
+            )
         exp_knxipframe.normalize()
         with patch("xknx.io.UDPClient.send") as mock_udp_send, patch(
             "xknx.io.UDPClient.getsockname"

test/io_tests/disconnect_test.py

@@ -21,12 +21,18 @@ def tearDown(self):
         self.loop.close()
 
     def test_disconnect(self):
+        self.__test_disconnect()
+
+    def test_disconnect_route_back_true(self):
+        self.__test_disconnect(route_back=True)
+
+    def __test_disconnect(self, route_back: bool = False):
         """Test disconnecting from KNX bus."""
         xknx = XKNX()
         communication_channel_id = 23
         udp_client = UDPClient(xknx, ("192.168.1.1", 0), ("192.168.1.2", 1234))
         disconnect = Disconnect(
-            xknx, udp_client, communication_channel_id, route_back=False
+            xknx, udp_client, communication_channel_id, route_back=route_back
         )
         disconnect.timeout_in_seconds = 0
 
@@ -37,7 +43,12 @@ def test_disconnect(self):
         exp_knxipframe = KNXIPFrame(xknx)
         exp_knxipframe.init(KNXIPServiceType.DISCONNECT_REQUEST)
         exp_knxipframe.body.communication_channel_id = communication_channel_id
-        exp_knxipframe.body.control_endpoint = HPAI(ip_addr="192.168.1.3", port=4321)
+        if route_back:
+            exp_knxipframe.body.control_endpoint = HPAI()
+        else:
+            exp_knxipframe.body.control_endpoint = HPAI(
+                ip_addr="192.168.1.3", port=4321
+            )
         exp_knxipframe.normalize()
         with patch("xknx.io.UDPClient.send") as mock_udp_send, patch(
             "xknx.io.UDPClient.getsockname"

xknx/config/config_v1.py

@@ -65,18 +65,22 @@ def parse_general(self, doc):
 
     def env_general(self):
         """Get Env Vars for the general section."""
-        if os.getenv("XKNX_GENERAL_OWN_ADDRESS", default=None):
+        own_address = os.getenv("XKNX_GENERAL_OWN_ADDRESS", default=None)
+        if own_address:
             logger.debug("XKNX_GENERAL_OWN_ADDRESS overwrite from env")
-            self.xknx.own_address = os.getenv("XKNX_GENERAL_OWN_ADDRESS")
-        if os.getenv("XKNX_GENERAL_RATE_LIMIT", default=None):
+            self.xknx.own_address = IndividualAddress(own_address)
+        rate_limit = os.getenv("XKNX_GENERAL_RATE_LIMIT", default=None)
+        if rate_limit:
             logger.debug("XKNX_GENERAL_RATE_LIMIT overwrite from env")
-            self.xknx.rate_limit = os.getenv("XKNX_GENERAL_RATE_LIMIT")
-        if os.getenv("XKNX_GENERAL_MULTICAST_GROUP", default=None):
+            self.xknx.rate_limit = int(rate_limit)
+        multicast_group = os.getenv("XKNX_GENERAL_MULTICAST_GROUP", default=None)
+        if multicast_group:
             logger.debug("XKNX_GENERAL_MULTICAST_GROUP overwrite from env")
-            self.xknx.multicast_group = os.getenv("XKNX_GENERAL_MULTICAST_GROUP")
-        if os.getenv("XKNX_GENERAL_MULTICAST_PORT", default=None):
+            self.xknx.multicast_group = multicast_group
+        multicast_port = os.getenv("XKNX_GENERAL_MULTICAST_PORT", default=None)
+        if multicast_port:
             logger.debug("XKNX_GENERAL_MULTICAST_PORT overwrite from env")
-            self.xknx.multicast_port = os.getenv("XKNX_GENERAL_MULTICAST_PORT")
+            self.xknx.multicast_port = int(multicast_port)
 
     def parse_connection(self, doc):
         """Parse the connection section of xknx.yaml."""