From 15b017656e6865b7b85ae389ab3b462c562a1e42 Mon Sep 17 00:00:00 2001 From: antialias Date: Tue, 25 Sep 2012 16:05:02 -0400 Subject: moved LeapNetworkChecker and test in base. --- src/leap/base/checks.py | 80 +++++++++++++++++++++++++++++++++++ src/leap/base/exceptions.py | 19 +++++++++ src/leap/base/tests/test_checks.py | 86 ++++++++++++++++++++++++++++++++++++++ src/leap/eip/checks.py | 76 --------------------------------- src/leap/eip/exceptions.py | 18 -------- src/leap/eip/tests/test_checks.py | 78 ---------------------------------- 6 files changed, 185 insertions(+), 172 deletions(-) create mode 100644 src/leap/base/checks.py create mode 100644 src/leap/base/tests/test_checks.py (limited to 'src/leap') diff --git a/src/leap/base/checks.py b/src/leap/base/checks.py new file mode 100644 index 00000000..c5438b09 --- /dev/null +++ b/src/leap/base/checks.py @@ -0,0 +1,80 @@ +# -*- coding: utf-8 -*- + +import platform + +import ping +import requests + +from leap.base import constants +from leap.base import exceptions + + +class LeapNetworkChecker(object): + """ + all network related checks + """ + # TODO eventually, use a more portable solution + # like psutil + + def run_all(self, checker=None): + if not checker: + checker = self + self.error = None # ? + + # for MVS + checker.check_internet_connection() + checker.is_internet_up() + checker.ping_gateway() + + def check_internet_connection(self): + try: + # XXX remove this hardcoded random ip + requests.get('http://216.172.161.165') + except (requests.HTTPError, requests.RequestException) as e: + raise exceptions.NoInternetConnection(e.message) + except requests.ConnectionError as e: + error = "Unidentified Connection Error" + if e.message == "[Errno 113] No route to host": + if not self.is_internet_up(): + error = "No valid internet connection found." + else: + error = "Provider server appears to be down." + raise exceptions.NoInternetConnection(error) + + def is_internet_up(self): + iface, gateway = self.get_default_interface_gateway() + self.ping_gateway(self) + + def get_default_interface_gateway(self): + """only impletemented for linux so far.""" + if not platform.system() == "Linux": + raise NotImplementedError + + f = open("/proc/net/route") + route_table = f.readlines() + f.close() + #toss out header + route_table.pop(0) + + default_iface = None + gateway = None + while route_table: + line = route_table.pop(0) + iface, destination, gateway = line.split('\t')[0:3] + if destination == '00000000': + default_iface = iface + break + + if not default_iface: + raise exceptions.NoDefaultInterfaceFoundError + + if default_iface not in netifaces.interfaces(): + raise exceptions.InterfaceNotFoundError + + return default_iface, gateway + + def ping_gateway(self, gateway): + #TODO: Discuss how much packet loss (%) is acceptable. + packet_loss = ping.quiet_ping(gateway)[0] + if packet_loss > constants.MAX_ICMP_PACKET_LOSS: + raise exceptions.NoConnectionToGateway diff --git a/src/leap/base/exceptions.py b/src/leap/base/exceptions.py index 9c4aa77b..7771d1f9 100644 --- a/src/leap/base/exceptions.py +++ b/src/leap/base/exceptions.py @@ -4,3 +4,22 @@ class MissingConfigFileError(Exception): class ImproperlyConfigured(Exception): pass + + +class NoDefaultInterfaceFoundError(Exception): + message = "no default interface found" + usermessage = "Looks like your computer is not connected to the internet" + + +class InterfaceNotFoundError(Exception): + # XXX should take iface arg on init maybe? + message = "interface not found" + + +class NoConnectionToGateway(Exception): + message = "no connection to gateway" + usermessage = "Looks like there are problems with your internet connection" + + +class NoInternetConnection(Exception): + message = "No Internet connection found" diff --git a/src/leap/base/tests/test_checks.py b/src/leap/base/tests/test_checks.py new file mode 100644 index 00000000..a3b3ea91 --- /dev/null +++ b/src/leap/base/tests/test_checks.py @@ -0,0 +1,86 @@ +try: + import unittest2 as unittest +except ImportError: + import unittest +import os + +from mock import (patch, Mock) +from StringIO import StringIO + +import ping +import requests + +from leap.base import checks +from leap.base import exceptions +from leap.testing.basetest import BaseLeapTest + +_uid = os.getuid() + + +class LeapNetworkCheckTest(BaseLeapTest): + __name__ = "leap_network_check_tests" + + def setUp(self): + pass + + def tearDown(self): + pass + + def test_checker_should_implement_check_methods(self): + checker = checks.LeapNetworkChecker() + + self.assertTrue(hasattr(checker, "check_internet_connection"), + "missing meth") + self.assertTrue(hasattr(checker, "is_internet_up"), + "missing meth") + self.assertTrue(hasattr(checker, "ping_gateway"), + "missing meth") + + def test_checker_should_actually_call_all_tests(self): + checker = checks.LeapNetworkChecker() + + mc = Mock() + checker.run_all(checker=mc) + self.assertTrue(mc.check_internet_connection.called, "not called") + self.assertTrue(mc.ping_gateway.called, "not called") + self.assertTrue(mc.is_internet_up.called, "not called") + + def test_get_default_interface_no_interface(self): + checker = checks.LeapNetworkChecker() + with patch('leap.base.checks.open', create=True) as mock_open: + with self.assertRaises(exceptions.NoDefaultInterfaceFoundError): + mock_open.return_value = StringIO( + "Iface\tDestination Gateway\t" + "Flags\tRefCntd\tUse\tMetric\t" + "Mask\tMTU\tWindow\tIRTT") + checker.get_default_interface_gateway() + + def test_ping_gateway_fail(self): + checker = checks.LeapNetworkChecker() + with patch.object(ping, "quiet_ping") as mocked_ping: + with self.assertRaises(exceptions.NoConnectionToGateway): + mocked_ping.return_value = [11, "", ""] + checker.ping_gateway("4.2.2.2") + + def test_check_internet_connection_failures(self): + checker = checks.LeapNetworkChecker() + with patch.object(requests, "get") as mocked_get: + mocked_get.side_effect = requests.HTTPError + with self.assertRaises(exceptions.NoInternetConnection): + checker.check_internet_connection() + + with patch.object(requests, "get") as mocked_get: + mocked_get.side_effect = requests.RequestException + with self.assertRaises(exceptions.NoInternetConnection): + checker.check_internet_connection() + + #TODO: Mock possible errors that can be raised by is_internet_up + with patch.object(requests, "get") as mocked_get: + mocked_get.side_effect = requests.ConnectionError + with self.assertRaises(exceptions.NoInternetConnection): + checker.check_internet_connection() + + @unittest.skipUnless(_uid == 0, "root only") + def test_ping_gateway(self): + checker = checks.LeapNetworkChecker() + checker.ping_gateway("4.2.2.2") diff --git a/src/leap/eip/checks.py b/src/leap/eip/checks.py index 20d1296d..9872f8d8 100644 --- a/src/leap/eip/checks.py +++ b/src/leap/eip/checks.py @@ -39,10 +39,6 @@ into base.tests to be invoked by the base leap init routines. However, I'm testing them alltogether for the sake of having the whole unit reachable and testable as a whole. -LeapNetworkChecker ------------------- -Network checks. To be moved to base. -docs TBD """ @@ -52,78 +48,6 @@ def get_ca_cert(): return certs.where(ca_file) -class LeapNetworkChecker(object): - """ - all network related checks - """ - # XXX to be moved to leap.base.checks - # TODO eventually, use a more portable solution - # like psutil - - def run_all(self, checker=None): - if not checker: - checker = self - self.error = None # ? - - # for MVS - checker.check_internet_connection() - checker.is_internet_up() - checker.ping_gateway() - - def check_internet_connection(self): - try: - # XXX remove this hardcoded random ip - requests.get('http://216.172.161.165') - except (requests.HTTPError, requests.RequestException) as e: - raise eipexceptions.NoInternetConnection(e.message) - except requests.ConnectionError as e: - error = "Unidentified Connection Error" - if e.message == "[Errno 113] No route to host": - if not self.is_internet_up(): - error = "No valid internet connection found." - else: - error = "Provider server appears to be down." - raise eipexceptions.NoInternetConnection(error) - - def is_internet_up(self): - iface, gateway = self.get_default_interface_gateway() - self.ping_gateway(self) - - def get_default_interface_gateway(self): - """only impletemented for linux so far.""" - if not platform.system() == "Linux": - raise NotImplementedError - - f = open("/proc/net/route") - route_table = f.readlines() - f.close() - #toss out header - route_table.pop(0) - - default_iface = None - gateway = None - while route_table: - line = route_table.pop(0) - iface, destination, gateway = line.split('\t')[0:3] - if destination == '00000000': - default_iface = iface - break - - if not default_iface: - raise eipexceptions.NoDefaultInterfaceFoundError - - if default_iface not in netifaces.interfaces(): - raise eipexceptions.InterfaceNotFoundError - - return default_iface, gateway - - def ping_gateway(self, gateway): - #TODO: Discuss how much packet loss (%) is acceptable. - packet_loss = ping.quiet_ping(gateway)[0] - if packet_loss > baseconstants.MAX_ICMP_PACKET_LOSS: - raise eipexceptions.NoConnectionToGateway - - class ProviderCertChecker(object): """ Several checks needed for getting diff --git a/src/leap/eip/exceptions.py b/src/leap/eip/exceptions.py index f883a173..6b4ee6aa 100644 --- a/src/leap/eip/exceptions.py +++ b/src/leap/eip/exceptions.py @@ -121,24 +121,6 @@ class EIPInitBadProviderError(EIPClientError): class EIPConfigurationError(EIPClientError): pass - -class NoDefaultInterfaceFoundError(EIPClientError): - message = "no default interface found" - usermessage = "Looks like your computer is not connected to the internet" - - -class InterfaceNotFoundError(EIPClientError): - # XXX should take iface arg on init maybe? - message = "interface not found" - - -class NoConnectionToGateway(EIPClientError): - message = "no connection to gateway" - usermessage = "Looks like there are problems with your internet connection" - -class NoInternetConnection(EIPClientError): - message = "No Internet connection found" - # # Errors that probably we don't need anymore # chase down for them and check. diff --git a/src/leap/eip/tests/test_checks.py b/src/leap/eip/tests/test_checks.py index f412dbec..06133825 100644 --- a/src/leap/eip/tests/test_checks.py +++ b/src/leap/eip/tests/test_checks.py @@ -9,10 +9,8 @@ import os import time import urlparse -from StringIO import StringIO from mock import (patch, Mock) -import ping import requests from leap.base import config as baseconfig @@ -26,8 +24,6 @@ from leap.testing.basetest import BaseLeapTest from leap.testing.https_server import BaseHTTPSServerTestCase from leap.testing.https_server import where as where_cert -_uid = os.getuid() - class NoLogRequestHandler: def log_message(self, *args): @@ -38,78 +34,6 @@ class NoLogRequestHandler: return '' -class LeapNetworkCheckTest(BaseLeapTest): - # XXX to be moved to base.checks - - __name__ = "leap_network_check_tests" - - def setUp(self): - pass - - def tearDown(self): - pass - - def test_checker_should_implement_check_methods(self): - checker = eipchecks.LeapNetworkChecker() - - self.assertTrue(hasattr(checker, "check_internet_connection"), - "missing meth") - self.assertTrue(hasattr(checker, "is_internet_up"), - "missing meth") - self.assertTrue(hasattr(checker, "ping_gateway"), - "missing meth") - - def test_checker_should_actually_call_all_tests(self): - checker = eipchecks.LeapNetworkChecker() - - mc = Mock() - checker.run_all(checker=mc) - self.assertTrue(mc.check_internet_connection.called, "not called") - self.assertTrue(mc.ping_gateway.called, "not called") - self.assertTrue(mc.is_internet_up.called, - "not called") - - def test_get_default_interface_no_interface(self): - checker = eipchecks.LeapNetworkChecker() - with patch('leap.eip.checks.open', create=True) as mock_open: - with self.assertRaises(eipexceptions.NoDefaultInterfaceFoundError): - mock_open.return_value = StringIO( - "Iface\tDestination Gateway\t" - "Flags\tRefCntd\tUse\tMetric\t" - "Mask\tMTU\tWindow\tIRTT") - checker.get_default_interface_gateway() - - def test_ping_gateway_fail(self): - checker = eipchecks.LeapNetworkChecker() - with patch.object(ping, "quiet_ping") as mocked_ping: - with self.assertRaises(eipexceptions.NoConnectionToGateway): - mocked_ping.return_value = [11, "", ""] - checker.ping_gateway("4.2.2.2") - - def test_check_internet_connection_failures(self): - checker = eipchecks.LeapNetworkChecker() - with patch.object(requests, "get") as mocked_get: - mocked_get.side_effect = requests.HTTPError - with self.assertRaises(eipexceptions.NoInternetConnection): - checker.check_internet_connection() - - with patch.object(requests, "get") as mocked_get: - mocked_get.side_effect = requests.RequestException - with self.assertRaises(eipexceptions.NoInternetConnection): - checker.check_internet_connection() - - #TODO: Mock possible errors that can be raised by is_internet_up - with patch.object(requests, "get") as mocked_get: - mocked_get.side_effect = requests.ConnectionError - with self.assertRaises(eipexceptions.NoInternetConnection): - checker.check_internet_connection() - - @unittest.skipUnless(_uid == 0, "root only") - def test_ping_gateway(self): - checker = eipchecks.LeapNetworkChecker() - checker.ping_gateway("4.2.2.2") - - class EIPCheckTest(BaseLeapTest): __name__ = "eip_check_tests" @@ -149,8 +73,6 @@ class EIPCheckTest(BaseLeapTest): "not called") self.assertTrue(mc.check_complete_eip_config.called, "not called") - #self.assertTrue(mc.ping_gateway.called, - #"not called") # test individual check methods -- cgit v1.2.3