summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/leap/base/config.py (renamed from src/leap/base/configuration.py)190
-rw-r--r--src/leap/base/connection.py4
-rw-r--r--src/leap/base/providers.py91
-rw-r--r--src/leap/base/tests/test_config.py (renamed from src/leap/base/tests/test_configuration.py)92
-rw-r--r--src/leap/base/tests/test_providers.py123
-rw-r--r--src/leap/eip/config.py11
-rw-r--r--src/leap/eip/tests/test_config.py20
-rw-r--r--src/leap/testing/basetest.py29
8 files changed, 450 insertions, 110 deletions
diff --git a/src/leap/base/configuration.py b/src/leap/base/config.py
index 155324df..dbd2e2c0 100644
--- a/src/leap/base/configuration.py
+++ b/src/leap/base/config.py
@@ -1,61 +1,99 @@
"""
Configuration Base Class
"""
-
+import configuration # python configuration module, not local!
import grp
import json
import logging
import requests
+import socket
import os
-from leap.util.fileutil import mkdir_p
-
logger = logging.getLogger(name=__name__)
logger.setLevel('DEBUG')
+from leap.util.fileutil import (mkdir_p)
-class Configuration(object):
- """
- All configurations (providers et al) will be managed in this class.
- """
- 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()
+class BaseLeapConfig(object):
+ slug = None
- if not os.path.isdir(default_provider_path):
- mkdir_p(default_provider_path)
+ # XXX we have to enforce that we have a slug (via interface)
+ # get property getter that raises NI..
- definition_file = get_config_file(
- 'definition.json',
- folder=default_provider_path)
+ def save(self):
+ raise NotImplementedError("abstract base class")
- if os.path.isfile(definition_file):
- return
+ def load(self):
+ raise NotImplementedError("abstract base class")
- 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(self, *kwargs):
+ raise NotImplementedError("abstract base class")
+
+ #XXX todo: enable this property after
+ #fixing name clash with "config" in use at
+ #vpnconnection
+
+ #@property
+ #def config(self):
+ #return self.get_config()
+
+ def get_value(self, *kwargs):
+ raise NotImplementedError("abstract base class")
+
+
+class JSONLeapConfig(BaseLeapConfig):
+
+ def __init__(self, *args, **kwargs):
+ # sanity check
+ assert self.slug is not None
+ assert self.spec is not None
+ assert issubclass(self.spec, configuration.Configuration)
+
+ self._config = self.spec()
+ self._config.parse_args(list(args))
+
+ # mandatory baseconfig interface
+
+ def save(self, to=None):
+ if to is None:
+ to = self.filename
+ self._config.serialize(to)
+
+ def load(self, fromfile=None):
+ # load should get a much more generic
+ # argument. it could be, f.i., from_uri,
+ # and call to Fetcher
+
+ if fromfile is None:
+ fromfile = self.filename
+ self._config.deserialize(fromfile)
+
+ def get_config(self):
+ return self._config.config
+
+ # public methods
+
+ def get_filename(self):
+ return self._slug_to_filename()
+
+ @property
+ def filename(self):
+ return self.get_filename()
+
+ def _slug_to_filename(self):
+ # is this going to work in winland if slug is "foo/bar" ?
+ folder, filename = os.path.split(self.slug)
+ # XXX fix import
+ config_file = get_config_file(filename, folder)
+ return config_file
+
+#
+# utility functions
+#
+# (might be moved to some class as we see fit, but
+# let's remain functional for a while)
+#
def get_config_dir():
@@ -69,9 +107,9 @@ def get_config_dir():
# get a more sensible path for win/mac
# kclair: opinion? ^^
return os.path.expanduser(
- os.path.join('~',
- '.config',
- 'leap'))
+ os.path.join('~',
+ '.config',
+ 'leap'))
def get_config_file(filename, folder=None):
@@ -116,6 +154,9 @@ def get_groupname():
return grp.getgrgid(gid).gr_name
+# json stuff
+
+# XXX merge with JSONConfig
def get_config_json(config_file=None):
"""
will replace get_config function be developing them
@@ -141,12 +182,71 @@ def get_config_json(config_file=None):
return json.load(open(config_file))
+def get_definition_file(url=None):
+ """
+ """
+ #TODO: determine good default location of definition file.
+ r = requests.get(url)
+ return r.json
+
+
def is_internet_up():
"""TODO: Build more robust network diagnosis capabilities
"""
try:
- response = requests.get('http://128.30.52.45', timeout=1)
+ requests.get('http://128.30.52.45', timeout=1)
return True
- except requests.Timeout as err:
+ except requests.Timeout: # as err:
pass
return False
+
+#
+# XXX merge conflict
+# tests are still using this deprecated Configuration object.
+# moving it here transiently until I clean merge commit.
+# -- kali 2012-08-24 00:32
+#
+
+
+class Configuration(object):
+ """
+ All configurations (providers et al) will be managed in this class.
+ """
+ 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
diff --git a/src/leap/base/connection.py b/src/leap/base/connection.py
index 8cd78433..9cdc33fa 100644
--- a/src/leap/base/connection.py
+++ b/src/leap/base/connection.py
@@ -5,13 +5,13 @@ from __future__ import (division, unicode_literals, print_function)
import logging
-from leap.base.configuration import Configuration
+from leap.base.config import JSONLeapConfig
from leap.base.authentication import Authentication
logger = logging.getLogger(name=__name__)
-class Connection(Configuration, Authentication):
+class Connection(JSONLeapConfig, Authentication):
def __init__(self, *args, **kwargs):
self.connection_state = None
self.desired_connection_state = None
diff --git a/src/leap/base/providers.py b/src/leap/base/providers.py
new file mode 100644
index 00000000..6fc050a0
--- /dev/null
+++ b/src/leap/base/providers.py
@@ -0,0 +1,91 @@
+import configuration
+
+from leap.base.config import JSONLeapConfig
+
+##########################################################
+# hacking in progress:
+
+# Specs are instances of configuration.Configuration class
+# and have to carry an options attr.
+#
+# Configs have:
+# - slug
+# - definition
+
+# TODO:
+# - have a good type cast repertory
+# - raise validation errors?
+##########################################################
+
+
+class LeapProviderSpec(configuration.Configuration):
+ options = {
+ 'serial': {
+ 'type': int,
+ 'default': 1,
+ 'required': True,
+ },
+ 'version': {
+ 'type': unicode,
+ 'default': '0.1.0'
+ #'required': True
+ },
+ 'domain': {
+ 'type': unicode, # XXX define uri type
+ 'default': 'testprovider.example.org'
+ #'required': True,
+ },
+ 'display_name': {
+ 'type': unicode, # XXX multilingual object?
+ 'default': 'test provider'
+ #'required': True
+ },
+ 'description': {
+ 'default': 'test provider'
+ },
+ 'enrollment_policy': {
+ 'type': unicode, # oneof ??
+ 'default': 'open'
+ },
+ 'services': {
+ 'type': list, # oneof ??
+ 'default': ['eip']
+ },
+ 'api_version': {
+ 'type': unicode,
+ 'default': '0.1.0' # version regexp
+ },
+ 'api_uri': {
+ 'type': unicode # uri
+ },
+ 'public_key': {
+ 'type': unicode # fingerprint
+ },
+ 'ca_cert': {
+ 'type': unicode
+ },
+ 'ca_cert_uri': {
+ 'type': unicode
+ },
+ }
+
+
+class LeapProviderDefinition(JSONLeapConfig):
+ slug = 'definition.json'
+ spec = LeapProviderSpec
+
+
+class LeapProvider(object):
+ # XXX ???
+ # do we need this?
+ # can we hook here the network fetching stuff?
+ # maybe (bstorming a little bit):
+
+ # config = LeapProviderDefinition
+ # fetcher = foo.FetcherClass
+ pass
+
+
+class LeapProviderSet(object):
+ def __init__(self):
+ self.count = 0
diff --git a/src/leap/base/tests/test_configuration.py b/src/leap/base/tests/test_config.py
index 17c8ed1f..c5231de2 100644
--- a/src/leap/base/tests/test_configuration.py
+++ b/src/leap/base/tests/test_config.py
@@ -1,13 +1,14 @@
-import mock
import os
import platform
-import requests
+import socket
import tempfile
+import mock
+import requests
+
+from leap.base import config
from leap.testing.basetest import BaseLeapTest
-from leap.base.configuration import Configuration
-from leap.base import configuration as config
try:
import unittest2 as unittest
@@ -19,6 +20,9 @@ _system = platform.system()
class DefinitionTestCase(BaseLeapTest):
+ # XXX See how to merge with test_providers
+ # -- kali 2012-08-24 00:38
+
__name__ = "provider_config_tests"
def setUp(self):
@@ -49,23 +53,23 @@ class DefinitionTestCase(BaseLeapTest):
u'serial': 1,
u'services': [u'eip'],
u'version': u'0.1.0'}
- cf = Configuration("http://localhost/")
+ cf = config.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()
+ cf = config.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()
+ cf = config.Configuration()
self.assertIsInstance(cf.error, str)
def test_invalid_url(self):
- cf = Configuration("ht")
+ cf = config.Configuration("ht")
self.assertTrue(cf.error)
@@ -73,6 +77,12 @@ class ConfigHelperFunctions(BaseLeapTest):
__name__ = "config_helper_tests"
+ def setUp(self):
+ pass
+
+ def tearDown(self):
+ pass
+
#
# tests
#
@@ -105,6 +115,40 @@ class ConfigHelperFunctions(BaseLeapTest):
"""
self._missing_test_for_plat(do_raise=True)
+ #
+ # 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(),
+ #XXX not correct!!!
+ #hardcoded home
+ '/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)
+
# provider paths
@unittest.skipUnless(_system == "Linux", "linux only")
@@ -136,34 +180,6 @@ class ConfigHelperFunctions(BaseLeapTest):
"""
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)
+if __name__ == "__main__":
+ unittest.main()
diff --git a/src/leap/base/tests/test_providers.py b/src/leap/base/tests/test_providers.py
new file mode 100644
index 00000000..2f029930
--- /dev/null
+++ b/src/leap/base/tests/test_providers.py
@@ -0,0 +1,123 @@
+import json
+try:
+ import unittest2 as unittest
+except ImportError:
+ import unittest
+
+import os
+
+from leap.testing.basetest import BaseLeapTest
+from leap.base import providers
+
+EXPECTED_DEFAULT_CONFIG = {
+ "api_version": "0.1.0",
+ "description": "test provider",
+ "display_name": "test provider",
+ "domain": "testprovider.example.org",
+ "enrollment_policy": "open",
+ "serial": 1,
+ "services": [
+ "eip"
+ ],
+ "version": "0.1.0"
+}
+
+
+class TestLeapProviderDefinition(BaseLeapTest):
+ def setUp(self):
+ self.definition = providers.LeapProviderDefinition()
+ #XXX change to self.definition.config when property is fixed
+ self.config = self.definition.get_config()
+
+ def tearDown(self):
+ if hasattr(self, 'testfile') and os.path.isfile(self.testfile):
+ os.remove(self.testfile)
+
+ # tests
+
+ def test_provider_dump(self):
+ # check a good provider definition is dumped to disk
+ self.testfile = self.get_tempfile('test.json')
+ self.definition.save(to=self.testfile)
+ deserialized = json.load(open(self.testfile, 'rb'))
+ self.assertEqual(deserialized, EXPECTED_DEFAULT_CONFIG)
+
+ def test_provider_dump_to_slug(self):
+ # same as above, but we test the ability to save to a
+ # file generated from the slug.
+ # XXX THIS TEST SHOULD MOVE TO test_baseconfig
+ self.definition.save()
+ filename = self.definition.filename
+ deserialized = json.load(open(filename, 'rb'))
+ self.assertEqual(deserialized, EXPECTED_DEFAULT_CONFIG)
+
+ def test_provider_load(self):
+ # check loading provider from disk file
+ self.testfile = self.get_tempfile('test_load.json')
+ with open(self.testfile, 'w') as wf:
+ wf.write(json.dumps(EXPECTED_DEFAULT_CONFIG))
+ self.definition.load(fromfile=self.testfile)
+ self.assertDictEqual(self.config,
+ EXPECTED_DEFAULT_CONFIG)
+
+ @unittest.skip
+ def test_load_malformed_json_definition(self):
+ raise NotImplementedError
+
+ @unittest.skip
+ def test_type_validation(self):
+ # check various type validation
+ # type cast
+ raise NotImplementedError
+
+
+class TestLeapProvider(BaseLeapTest):
+ def setUp(self):
+ pass
+
+ def tearDown(self):
+ pass
+
+ ###
+
+ # XXX ??
+
+
+class TestLeapProviderSet(BaseLeapTest):
+
+ def setUp(self):
+ self.providers = providers.LeapProviderSet()
+
+ def tearDown(self):
+ pass
+ ###
+
+ def test_get_zero_count(self):
+ self.assertEqual(self.providers.count, 0)
+
+ @unittest.skip
+ def test_count_defined_providers(self):
+ # check the method used for making
+ # the list of providers
+ raise NotImplementedError
+
+ @unittest.skip
+ def test_get_default_provider(self):
+ raise NotImplementedError
+
+ @unittest.skip
+ def test_should_be_at_least_one_provider_after_init(self):
+ # when we init an empty environment,
+ # there should be at least one provider,
+ # that will be a dump of the default provider definition
+ # somehow a high level test
+ raise NotImplementedError
+
+ @unittest.skip
+ def test_get_eip_remote_from_default_provider(self):
+ # from: default provider
+ # expect: remote eip domain
+ raise NotImplementedError
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/src/leap/eip/config.py b/src/leap/eip/config.py
index b461422a..8d5c19da 100644
--- a/src/leap/eip/config.py
+++ b/src/leap/eip/config.py
@@ -1,17 +1,16 @@
import ConfigParser
-import grp
import logging
import os
-import json
import platform
import socket
from leap.util.fileutil import (which, mkdir_p,
check_and_fix_urw_only)
-from leap.base.configuration import (get_default_provider_path,
- get_config_file,
- get_username,
- get_groupname)
+from leap.base.config import (get_default_provider_path,
+ get_config_file,
+ get_username,
+ get_groupname,
+ validate_ip)
from leap.baseapp.permcheck import (is_pkexec_in_system,
is_auth_agent_running)
from leap.eip import exceptions as eip_exceptions
diff --git a/src/leap/eip/tests/test_config.py b/src/leap/eip/tests/test_config.py
index 0e1a3a01..3c5a1cde 100644
--- a/src/leap/eip/tests/test_config.py
+++ b/src/leap/eip/tests/test_config.py
@@ -1,9 +1,7 @@
import ConfigParser
import os
import platform
-import shutil
import socket
-import tempfile
try:
import unittest2 as unittest
@@ -11,7 +9,8 @@ except ImportError:
import unittest
from leap.testing.basetest import BaseLeapTest
-from leap.eip import config
+from leap.base import config as base_config
+from leap.eip import config as eip_config
_system = platform.system()
@@ -30,19 +29,6 @@ class EIPConfigTest(BaseLeapTest):
# helpers
#
- def get_username(self):
- return config.get_username()
-
- def get_groupname(self):
- return config.get_groupname()
-
- def _missing_test_for_plat(self, do_raise=False):
- if do_raise:
- raise NotImplementedError(
- "This test is not implemented "
- "for the running platform: %s" %
- _system)
-
def touch_exec(self):
tfile = os.path.join(
self.tempfile,
@@ -90,7 +76,7 @@ class EIPConfigTest(BaseLeapTest):
def test_build_ovpn_command_empty_config(self):
_config = self.get_empty_config()
- command, args = config.build_ovpn_command(
+ command, args = eip_config.build_ovpn_command(
_config,
do_pkexec_check=False)
self.assertEqual(command, 'openvpn')
diff --git a/src/leap/testing/basetest.py b/src/leap/testing/basetest.py
index ccf1a76f..a55b0525 100644
--- a/src/leap/testing/basetest.py
+++ b/src/leap/testing/basetest.py
@@ -1,4 +1,5 @@
import os
+import platform
import shutil
import tempfile
@@ -7,6 +8,10 @@ try:
except ImportError:
import unittest
+from leap.base.config import get_username, get_groupname
+
+_system = platform.system()
+
class BaseLeapTest(unittest.TestCase):
@@ -26,12 +31,32 @@ class BaseLeapTest(unittest.TestCase):
os.environ["PATH"] = cls.old_path
shutil.rmtree(cls.tempdir)
+ # you have to override these methods
+ # this way we ensure we did not put anything
+ # here that you can forget to call.
+
def setUp(self):
raise NotImplementedError("abstract base class")
def tearDown(self):
raise NotImplementedError("abstract base class")
+ #
+ # helper methods
+ #
+
+ def get_tempfile(self, filename):
+ return os.path.join(self.tempdir, filename)
+
+ def get_username(self):
+ return get_username()
+
+ def get_groupname(self):
+ return get_groupname()
-if __name__ == "__main__":
- unittest.main()
+ def _missing_test_for_plat(self, do_raise=False):
+ if do_raise:
+ raise NotImplementedError(
+ "This test is not implemented "
+ "for the running platform: %s" %
+ _system)
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-10 16:24:24 +0000