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 .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())