diff options
Diffstat (limited to 'src/leap/base')
| -rw-r--r-- | src/leap/base/configuration.py | 67 | ||||
| -rw-r--r-- | src/leap/base/tests/test_configuration.py | 169 |
2 files changed, 224 insertions, 12 deletions
diff --git a/src/leap/base/configuration.py b/src/leap/base/configuration.py index 31092f8a..155324df 100644 --- a/src/leap/base/configuration.py +++ b/src/leap/base/configuration.py @@ -3,20 +3,59 @@ Configuration Base Class """ import grp +import json import logging import requests import os +from leap.util.fileutil import mkdir_p + logger = logging.getLogger(name=__name__) logger.setLevel('DEBUG') class Configuration(object): """ - I have no idea how configuration - (txt vs. sqlite) will be done, but let's stub it now. + All configurations (providers et al) will be managed in this class. """ - pass + def __init__(self, provider_url=None): + try: + self.providers = {} + self.error = False + provider_file = self.check_and_get_definition_file(provider_url) + self.providers['default'] = get_config_json(provider_file) + except (requests.HTTPError, requests.RequestException) as e: + self.error = e.message + except requests.ConnectionError as e: + if e.message == "[Errno 113] No route to host": + if not is_internet_up: + self.error = "No valid internet connection found" + else: + self.error = "Provider server appears currently down." + + def check_and_get_definition_file(self, provider_url): + """ + checks if provider definition.json file is present. + if not downloads one from the web. + """ + default_provider_path = get_default_provider_path() + + if not os.path.isdir(default_provider_path): + mkdir_p(default_provider_path) + + definition_file = get_config_file( + 'definition.json', + folder=default_provider_path) + + if os.path.isfile(definition_file): + return + + else: + r = requests.get(provider_url) + r.raise_for_status() + with open(definition_file, 'wb') as f: + f.write(json.dumps(r.json, indent=4)) + return definition_file def get_config_dir(): @@ -87,6 +126,7 @@ def get_config_json(config_file=None): @rtype: dictionary """ if not config_file: + #TODO: NOT SURE WHAT this default should be, if anything fpath = get_config_file('eip.json') if not os.path.isfile(fpath): dpath, cfile = os.path.split(fpath) @@ -94,16 +134,19 @@ def get_config_json(config_file=None): mkdir_p(dpath) with open(fpath, 'wb') as configfile: configfile.flush() - config_file = open(fpath) - - config = json.load(config_file) + return json.load(open(fpath)) - return config + else: + #TODO: add validity checks of file + return json.load(open(config_file)) -def get_definition_file(url=None): - """ +def is_internet_up(): + """TODO: Build more robust network diagnosis capabilities """ - #TODO: determine good default location of definition file. - r = requests.get(url) - return r.json + try: + response = requests.get('http://128.30.52.45', timeout=1) + return True + except requests.Timeout as err: + pass + return False diff --git a/src/leap/base/tests/test_configuration.py b/src/leap/base/tests/test_configuration.py new file mode 100644 index 00000000..17c8ed1f --- /dev/null +++ b/src/leap/base/tests/test_configuration.py @@ -0,0 +1,169 @@ +import mock +import os +import platform +import requests +import tempfile + +from leap.testing.basetest import BaseLeapTest +from leap.base.configuration import Configuration + +from leap.base import configuration as config + +try: + import unittest2 as unittest +except ImportError: + import unittest + +_system = platform.system() + + +class DefinitionTestCase(BaseLeapTest): + + __name__ = "provider_config_tests" + + def setUp(self): + self.old_home = os.environ['HOME'] + self.home = tempfile.mkdtemp() + os.environ['HOME'] = self.home + pass + + #Not correct removing the test directories but will be refactor out + #with kali's new test classes + def tearDown(self): + os.environ['HOME'] = self.old_home + pass + + def test_complete_file(self): + with mock.patch.object(requests, "get") as mock_method: + mock_method.return_value.status_code = 200 + mock_method.return_value.json = { + u'api_uri': u'https://api.testprovider.org/', + u'api_version': u'0.1.0', + u'ca_cert': u'8aab80ae4326fd30721689db813733783fe0bd7e', + u'ca_cert_uri': u'https://testprovider.org/cacert.pem', + u'description': {u'en': u'This is a test provider'}, + u'display_name': {u'en': u'Test Provider'}, + u'domain': u'testprovider.org', + u'enrollment_policy': u'open', + u'public_key': u'cb7dbd679f911e85bc2e51bd44afd7308ee19c21', + u'serial': 1, + u'services': [u'eip'], + u'version': u'0.1.0'} + cf = Configuration("http://localhost/") + self.assertIn('default', cf.providers) + + def test_connection_error(self): + with mock.patch.object(requests, "get") as mock_method: + mock_method.side_effect = requests.ConnectionError + cf = Configuration() + self.assertIsInstance(cf.error, str) + + def test_file_not_found(self): + with mock.patch.object(requests, "get") as mock_method: + mock_method.side_effect = requests.HTTPError + cf = Configuration() + self.assertIsInstance(cf.error, str) + + def test_invalid_url(self): + cf = Configuration("ht") + self.assertTrue(cf.error) + + +class ConfigHelperFunctions(BaseLeapTest): + + __name__ = "config_helper_tests" + + # + # tests + # + + # XXX fixme! /home/user should + # be replaced for proper home lookup. + + @unittest.skipUnless(_system == "Linux", "linux only") + def test_lin_get_config_file(self): + """ + config file path where expected? (linux) + """ + self.assertEqual( + config.get_config_file( + 'test', folder="foo/bar"), + '/home/%s/.config/leap/foo/bar/test' % + config.get_username()) + + @unittest.skipUnless(_system == "Darwin", "mac only") + def test_mac_get_config_file(self): + """ + config file path where expected? (mac) + """ + self._missing_test_for_plat(do_raise=True) + + @unittest.skipUnless(_system == "Windows", "win only") + def test_win_get_config_file(self): + """ + config file path where expected? + """ + self._missing_test_for_plat(do_raise=True) + + # provider paths + + @unittest.skipUnless(_system == "Linux", "linux only") + def test_get_default_provider_path(self): + """ + is default provider path ok? + """ + self.assertEqual( + config.get_default_provider_path(), + '/home/%s/.config/leap/providers/default/' % + config.get_username()) + + # validate ip + + def test_validate_ip(self): + """ + check our ip validation + """ + config.validate_ip('3.3.3.3') + with self.assertRaises(socket.error): + config.validate_ip('255.255.255.256') + with self.assertRaises(socket.error): + config.validate_ip('foobar') + + @unittest.skip + def test_validate_domain(self): + """ + code to be written yet + """ + pass + + # + # XXX hey, I'm raising exceptions here + # on purpose. just wanted to make sure + # that the skip stuff is doing it right. + # If you're working on win/macos tests, + # feel free to remove tests that you see + # are too redundant. + + @unittest.skipUnless(_system == "Linux", "linux only") + def test_lin_get_config_dir(self): + """ + nice config dir? (linux) + """ + self.assertEqual( + config.get_config_dir(), + '/home/%s/.config/leap' % + self.get_username()) + + @unittest.skipUnless(_system == "Darwin", "mac only") + def test_mac_get_config_dir(self): + """ + nice config dir? (mac) + """ + self._missing_test_for_plat(do_raise=True) + + @unittest.skipUnless(_system == "Windows", "win only") + def test_win_get_config_dir(self): + """ + nice config dir? (win) + """ + self._missing_test_for_plat(do_raise=True) |