Merge pull request #2962 from MycroftAI/feature/configurable-network-tests

Make network tests configurable

mycroft/configuration/mycroft.conf

@@ -146,6 +146,15 @@
         "ssl": false
   },
 
+  // URIs to use for testing network connection.
+  "network_tests": {
+    "dns_primary": "8.8.8.8",
+    "dns_secondary": "8.8.4.4",
+    "web_url": "https://www.google.com",
+    "ncsi_endpoint": "http://www.msftncsi.com/ncsi.txt",
+    "ncsi_expected_text": "Microsoft NCSI"
+  },
+
   // Settings used by the wake-up-word listener
   // Override: REMOTE
   "listener": {

mycroft/util/network_utils.py

@@ -6,6 +6,14 @@ from urllib.error import URLError
 from .log import LOG
 
 
+def _get_network_tests_config():
+    """Get network_tests object from mycroft.configuration."""
+    # Wrapped to avoid circular import errors.
+    from mycroft.configuration import Configuration
+    config = Configuration.get()
+    return config.get('network_tests', {})
+
+
 def connected():
     """Check connection by connecting to 8.8.8.8 and if google.com is
     reachable if this fails, Check Microsoft NCSI is used as a backup.
@@ -27,16 +35,19 @@ def _connected_ncsi():
     Returns:
         True if internet connection can be detected
     """
+    config = _get_network_tests_config()
+    ncsi_endpoint = config.get('ncsi_endpoint')
+    expected_text = config.get('ncsi_expected_text')
     try:
-        r = requests.get('http://www.msftncsi.com/ncsi.txt')
-        if r.text == 'Microsoft NCSI':
+        r = requests.get(ncsi_endpoint)
+        if r.text == expected_text:
             return True
     except Exception:
-        pass
+        LOG.error("Unable to verify connection via NCSI endpoint.")
     return False
 
 
-def _connected_dns(host="8.8.8.8", port=53, timeout=3):
+def _connected_dns(host=None, port=53, timeout=3):
     """Check internet connection by connecting to DNS servers
 
     Returns:
@@ -46,18 +57,25 @@ def _connected_dns(host="8.8.8.8", port=53, timeout=3):
     # Host: 8.8.8.8 (google-public-dns-a.google.com)
     # OpenPort: 53/tcp
     # Service: domain (DNS/TCP)
+    config = _get_network_tests_config()
+    if host is None:
+        host = config.get('dns_primary')
     try:
         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         s.settimeout(timeout)
         s.connect((host, port))
         return True
     except IOError:
+        LOG.error("Unable to connect to primary DNS server, "
+                  "trying secondary...")
         try:
             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             s.settimeout(timeout)
-            s.connect(("8.8.4.4", port))
+            dns_secondary = config.get('dns_secondary')
+            s.connect((dns_secondary, port))
             return True
         except IOError:
+            LOG.error("Unable to connect to secondary DNS server.")
             return False
 
 
@@ -67,10 +85,12 @@ def _connected_google():
         True if connection attempt succeeded
     """
     connect_success = False
+    config = _get_network_tests_config()
+    url = config.get('web_url')
     try:
-        urlopen('https://www.google.com', timeout=3)
+        urlopen(url, timeout=3)
     except URLError as ue:
-        LOG.debug('Attempt to connect to internet failed: ' + str(ue.reason))
+        LOG.error('Attempt to connect to internet failed: ' + str(ue.reason))
     else:
         connect_success = True
 

test/unittests/util/test_network_utils.py

@@ -0,0 +1,52 @@
+from unittest import TestCase, mock
+
+from mycroft.util.network_utils import connected
+
+
+class TestNetworkConnected(TestCase):
+    def test_default_config_succeeds(self):
+        """Check that happy path succeeds"""
+        self.assertTrue(connected())
+
+
+@mock.patch('mycroft.configuration.Configuration')
+class TestNetworkFailure(TestCase):
+
+    def test_dns_and_ncsi_fail(self, mock_conf):
+        """Check that DNS and NCSI failure results in False response"""
+        mock_conf.get.return_value = {
+            "network_tests": {
+                "dns_primary": "127.0.0.1",
+                "dns_secondary": "127.0.0.1",
+                "web_url": "https://www.google.com",
+                "ncsi_endpoint": "http://www.msftncsi.com/ncsi.txt",
+                "ncsi_expected_text": "Unexpected text"
+            }
+        }
+        self.assertFalse(connected())
+
+    def test_secondary_dns_succeeds(self, mock_conf):
+        """Check that only primary DNS failing still succeeds"""
+        mock_conf.get.return_value = {
+            "network_tests": {
+                "dns_primary": "127.0.0.1",
+                "dns_secondary": "8.8.4.4",
+                "web_url": "https://www.google.com",
+                "ncsi_endpoint": "http://www.msftncsi.com/ncsi.txt",
+                "ncsi_expected_text": "Microsoft NCSI"
+            }
+        }
+        self.assertTrue(connected())
+
+    def test_dns_success_url_fail(self, mock_conf):
+        """Check that URL connection failure results in False response"""
+        mock_conf.get.return_value = {
+            "network_tests": {
+                "dns_primary": "8.8.8.8",
+                "dns_secondary": "8.8.4.4",
+                "web_url": "https://test.invalid",
+                "ncsi_endpoint": "http://www.msftncsi.com/ncsi.txt",
+                "ncsi_expected_text": "Microsoft NCSI"
+            }
+        }
+        self.assertFalse(connected())