Merge branch 'develop' of ssh://leap.se:4422/leap-client into develop

3 files changed, 534 insertions, 0 deletions

diff --git a/src/leap/eip/tests/test_config.py b/src/leap/eip/tests/test_config.py
new file mode 100644
index 00000000..051faa29
--- /dev/null
+++ b/src/leap/eip/tests/test_config.py
@@ -0,0 +1,218 @@
+import ConfigParser
+import os
+import platform
+import shutil
+import socket
+import tempfile
+
+try:
+    import unittest2 as unittest
+except ImportError:
+    import unittest
+
+from leap.eip import config
+
+_system = platform.system()
+
+
+class NotImplementedError(Exception):
+    pass
+
+# XXX use mock_open here?
+
+
+class EIPConfigTest(unittest.TestCase):
+
+    __name__ = "eip_config_tests"
+
+    def setUp(self):
+        self.old_path = os.environ['PATH']
+
+        self.tdir = tempfile.mkdtemp()
+
+        bin_tdir = os.path.join(
+            self.tdir,
+            'bin')
+        os.mkdir(bin_tdir)
+        os.environ['PATH'] = bin_tdir
+
+    def tearDown(self):
+        os.environ['PATH'] = self.old_path
+        shutil.rmtree(self.tdir)
+    #
+    # 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.tdir,
+            'bin',
+            'openvpn')
+        open(tfile, 'bw').close()
+
+    def get_empty_config(self):
+        _config = ConfigParser.ConfigParser()
+        return _config
+
+    def get_minimal_config(self):
+        _config = ConfigParser.ConfigParser()
+        return _config
+
+    def get_expected_openvpn_args(self):
+        args = []
+        username = self.get_username()
+        groupname = self.get_groupname()
+
+        args.append('--user')
+        args.append(username)
+        args.append('--group')
+        args.append(groupname)
+        args.append('--management-client-user')
+        args.append(username)
+        args.append('--management-signal')
+        args.append('--management')
+
+        #XXX hey!
+        #get platform switches here!
+        args.append('/tmp/.eip.sock')
+        args.append('unix')
+        args.append('--config')
+        #XXX bad assumption. FIXME: expand $HOME
+        args.append('/home/%s/.config/leap/providers/default/openvpn.conf' %
+                    username)
+        return args
+
+    #
+    # 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' %
+            self.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)
+
+    #
+    # 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)
+
+    # 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/' %
+            self.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
+
+    # build command string
+    # these tests are going to have to check
+    # many combinations. we should inject some
+    # params in the function call, to disable
+    # some checks.
+
+    def test_build_ovpn_command_empty_config(self):
+        _config = self.get_empty_config()
+        command, args = config.build_ovpn_command(
+            _config,
+            do_pkexec_check=False)
+        self.assertEqual(command, 'openvpn')
+        self.assertEqual(args, self.get_expected_openvpn_args())
+
+    # json config
+
+    def test_get_config_json(self):
+        config_js = config.get_config_json()
+        self.assertTrue(isinstance(config_js, dict))
+        self.assertTrue('transport' in config_js)
+        self.assertTrue('provider' in config_js)
+        self.assertEqual(config_js['provider'], "testprovider.org")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/src/leap/eip/tests/test_eipconnection.py b/src/leap/eip/tests/test_eipconnection.py
new file mode 100644
index 00000000..51772b7c
--- /dev/null
+++ b/src/leap/eip/tests/test_eipconnection.py
@@ -0,0 +1,180 @@
+import ConfigParser
+import logging
+import platform
+
+logging.basicConfig()
+logger = logging.getLogger(name=__name__)
+
+try:
+    import unittest2 as unittest
+except ImportError:
+    import unittest
+
+from mock import Mock, patch  # MagicMock
+
+from leap.eip.eipconnection import EIPConnection
+from leap.eip.exceptions import ConnectionRefusedError
+
+_system = platform.system()
+
+
+class NotImplementedError(Exception):
+    pass
+
+
+@patch('OpenVPNConnection._get_or_create_config')
+@patch('OpenVPNConnection._set_ovpn_command')
+class MockedEIPConnection(EIPConnection):
+    def _get_or_create_config(self):
+        self.config = ConfigParser.ConfigParser()
+        self._set_ovpn_command()
+
+    def _set_ovpn_command(self):
+        self.command = "mock_command"
+        self.args = [1, 2, 3]
+
+
+class EIPConductorTest(unittest.TestCase):
+
+    __name__ = "eip_conductor_tests"
+
+    def setUp(self):
+        self.manager = Mock(
+            name="openvpnmanager_mock")
+
+        self.con = MockedEIPConnection()
+            #manager=self.manager)
+
+    def tearDown(self):
+        del self.con
+
+    #
+    # helpers
+    #
+
+    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)
+
+    #
+    # tests
+    #
+
+    @unittest.skip
+    #ain't manager anymore!
+    def test_manager_was_initialized(self):
+        """
+        manager init ok during conductor init?
+        """
+        self.manager.assert_called_once_with()
+
+    def test_vpnconnection_defaults(self):
+        """
+        default attrs as expected
+        """
+        con = self.con
+        self.assertEqual(con.autostart, True)
+        self.assertEqual(con.missing_pkexec, False)
+        self.assertEqual(con.missing_vpn_keyfile, False)
+        self.assertEqual(con.missing_provider, False)
+        self.assertEqual(con.bad_provider, False)
+
+    def test_config_was_init(self):
+        """
+        is there a config object?
+        """
+        self.assertTrue(isinstance(self.con.config,
+                        ConfigParser.ConfigParser))
+
+    def test_ovpn_command(self):
+        """
+        set_ovpn_command called
+        """
+        self.assertEqual(self.con.command,
+                         "mock_command")
+        self.assertEqual(self.con.args,
+                         [1, 2, 3])
+
+    # connect/disconnect calls
+
+    def test_disconnect(self):
+        """
+        disconnect method calls private and changes status
+        """
+        self.con._disconnect = Mock(
+            name="_disconnect")
+
+        # first we set status to connected
+        self.con.status.set_current(self.con.status.CONNECTED)
+        self.assertEqual(self.con.status.current,
+                         self.con.status.CONNECTED)
+
+        # disconnect
+        self.con.disconnect()
+        self.con._disconnect.assert_called_once_with()
+
+        # new status should be disconnected
+        # XXX this should evolve and check no errors
+        # during disconnection
+        self.assertEqual(self.con.status.current,
+                         self.con.status.DISCONNECTED)
+
+    def test_connect(self):
+        """
+        connect calls _launch_openvpn private
+        """
+        self.con._launch_openvpn = Mock()
+        self.con.connect()
+        self.con._launch_openvpn.assert_called_once_with()
+
+    # XXX tests breaking here ...
+
+    def test_good_poll_connection_state(self):
+        """
+        """
+        #@patch --
+        # self.manager.get_connection_state
+
+        #XXX review this set of poll_state tests
+        #they SHOULD NOT NEED TO MOCK ANYTHING IN THE
+        #lower layers!! -- status, vpn_manager..
+        #right now we're testing implementation, not
+        #behavior!!!
+        good_state = ["1345466946", "unknown_state", "ok",
+                      "192.168.1.1", "192.168.1.100"]
+        self.con.get_connection_state = Mock(return_value=good_state)
+        self.con.status.set_vpn_state = Mock()
+
+        state = self.con.poll_connection_state()
+        good_state[1] = "disconnected"
+        final_state = tuple(good_state)
+        self.con.status.set_vpn_state.assert_called_with("unknown_state")
+        self.assertEqual(state, final_state)
+
+    # TODO between "good" and "bad" (exception raised) cases,
+    # we can still test for malformed states and see that only good
+    # states do have a change (and from only the expected transition
+    # states).
+
+    def test_bad_poll_connection_state(self):
+        """
+        get connection state raises ConnectionRefusedError
+        state is None
+        """
+        self.con.get_connection_state = Mock(
+            side_effect=ConnectionRefusedError('foo!'))
+        state = self.con.poll_connection_state()
+        self.assertEqual(state, None)
+
+
+    # XXX more things to test:
+    # - called config routines during initz.
+    # - raising proper exceptions with no config
+    # - called proper checks on config / permissions
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/src/leap/eip/tests/test_openvpnconnection.py b/src/leap/eip/tests/test_openvpnconnection.py
new file mode 100644
index 00000000..dea75b55
--- /dev/null
+++ b/src/leap/eip/tests/test_openvpnconnection.py
@@ -0,0 +1,136 @@
+import logging
+import platform
+#import socket
+
+logging.basicConfig()
+logger = logging.getLogger(name=__name__)
+
+try:
+    import unittest2 as unittest
+except ImportError:
+    import unittest
+
+from mock import Mock, patch  # MagicMock
+
+from leap.eip import openvpnconnection
+from leap.eip import exceptions as eip_exceptions
+from leap.eip.udstelnet import UDSTelnet
+
+_system = platform.system()
+
+
+class NotImplementedError(Exception):
+    pass
+
+
+mock_UDSTelnet = Mock(spec=UDSTelnet)
+# XXX cautious!!!
+# this might be fragile right now (counting a global
+# reference of calls I think.
+# investigate this other form instead:
+# http://www.voidspace.org.uk/python/mock/patch.html#start-and-stop
+
+# XXX redo after merge-refactor
+
+
+@patch('openvpnconnection.OpenVPNConnection.connect_to_management')
+class MockedOpenVPNConnection(openvpnconnection.OpenVPNConnection):
+    def __init__(self, *args, **kwargs):
+        self.mock_UDSTelnet = Mock()
+        super(MockedOpenVPNConnection, self).__init__(
+            *args, **kwargs)
+        self.tn = self.mock_UDSTelnet(self.host, self.port)
+
+    def connect_to_management(self):
+        #print 'patched connect'
+        self.tn = mock_UDSTelnet(self.host, port=self.port)
+
+
+class OpenVPNConnectionTest(unittest.TestCase):
+
+    __name__ = "vpnconnection_tests"
+
+    def setUp(self):
+        self.manager = MockedOpenVPNConnection()
+
+    def tearDown(self):
+        del self.manager
+
+    #
+    # helpers
+    #
+
+    # XXX hey, refactor this to basetestclass
+
+    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)
+
+    #
+    # tests
+    #
+
+    @unittest.skipIf(_system == "Windows", "lin/mac only")
+    def test_lin_mac_default_init(self):
+        """
+        check default host for management iface
+        """
+        self.assertEqual(self.manager.host, '/tmp/.eip.sock')
+        self.assertEqual(self.manager.port, 'unix')
+
+    @unittest.skipUnless(_system == "Windows", "win only")
+    def test_win_default_init(self):
+        """
+        check default host for management iface
+        """
+        # XXX should we make the platform specific switch
+        # here or in the vpn command string building?
+        self.assertEqual(self.manager.host, 'localhost')
+        self.assertEqual(self.manager.port, 7777)
+
+    def test_port_types_init(self):
+        self.manager = MockedOpenVPNConnection(port="42")
+        self.assertEqual(self.manager.port, 42)
+        self.manager = MockedOpenVPNConnection()
+        self.assertEqual(self.manager.port, "unix")
+        self.manager = MockedOpenVPNConnection(port="bad")
+        self.assertEqual(self.manager.port, None)
+
+    def test_connect_raises_missing_socket(self):
+        self.manager = openvpnconnection.OpenVPNConnection()
+        with self.assertRaises(eip_exceptions.MissingSocketError):
+            self.manager.connect_to_management()
+
+    def test_uds_telnet_called_on_connect(self):
+        self.manager.connect_to_management()
+        mock_UDSTelnet.assert_called_with(
+            self.manager.host,
+            port=self.manager.port)
+
+    @unittest.skip
+    def test_connect(self):
+        raise NotImplementedError
+        # XXX calls close
+        # calls UDSTelnet mock.
+
+    # XXX
+    # tests to write:
+    # UDSTelnetTest (for real?)
+    # HAVE A LOOK AT CORE TESTS FOR TELNETLIB.
+    # very illustrative instead...
+
+    # - raise MissingSocket
+    # - raise ConnectionRefusedError
+    # - test send command
+    #   - tries connect
+    #   - ... tries?
+    #   - ... calls _seek_to_eof
+    #   - ... read_until --> return value
+    #   - ...
+
+
+if __name__ == "__main__":
+    unittest.main()