some conductor / manager tests

4 files changed, 260 insertions, 13 deletions

diff --git a/src/leap/eip/conductor.py b/src/leap/eip/conductor.py
index 8f9d605..9bb3bd4 100644
--- a/src/leap/eip/conductor.py
+++ b/src/leap/eip/conductor.py
@@ -73,7 +73,7 @@ class OpenVPNConnection(object):
     """
     # Connection Methods
 
-    def __init__(self, config_file=None,
+    def __init__(self, manager=None, config_file=None,
                  watcher_cb=None, debug=False):
         #XXX FIXME
         #change watcher_cb to line_observer
@@ -88,10 +88,12 @@ to be triggered for each one of them.
         :type signal_map: dict
         """
         # XXX get host/port from config
-        self.manager = OpenVPNManager()
         self.debug = debug
-        #print('conductor:%s' % debug)
+        if manager is None:
+            manager = OpenVPNManager
+        self.manager = manager()
 
+        self.config = None
         self.config_file = config_file
         self.watcher_cb = watcher_cb
         #self.signal_maps = signal_maps
@@ -183,6 +185,7 @@ to be triggered for each one of them.
         home file, or config file passed in command line.
         populates command and args to be passed to subprocess.
         """
+        # XXX inject config object in a previous step
         config = get_config(config_file=self.config_file)
         self.config = config
 
@@ -225,9 +228,6 @@ to be triggered for each one of them.
         self.subp = subp
         self.watcher = watcher
 
-        #conn_result = self.status.CONNECTED
-        #return conn_result
-
     def _try_connection(self):
         """
         attempts to connect
@@ -317,17 +317,10 @@ class EIPConductor(OpenVPNConnection):
             self.subp = None
         # XXX signal state changes! :)
 
-    def _is_alive(self):
-        """
-        don't know yet
-        """
-        pass
-
     def _connect(self):
         """
         entry point for connection cascade methods.
         """
-        #conn_result = ConState.DISCONNECTED
         try:
             conn_result = self._try_connection()
         except UnrecoverableError as except_msg:
diff --git a/src/leap/eip/test_conductor.py b/src/leap/eip/test_conductor.py
new file mode 100644
index 0000000..051bac5
--- /dev/null
+++ b/src/leap/eip/test_conductor.py
@@ -0,0 +1,140 @@
+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.conductor import EIPConductor
+
+_system = platform.system()
+
+
+class NotImplementedError(Exception):
+    pass
+
+
+@patch('OpenVPNConnection._get_or_create_config')
+@patch('OpenVPNConnection._set_ovpn_command')
+class MockedEIPConductor(EIPConductor):
+    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 = MockedEIPConductor(
+            manager=self.manager)
+        #self.con._connect = Mock(
+            #name="_connect")
+
+    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
+    #
+
+    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 more things to test:
+    # - called config routinges 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/test_vpnmanager.py b/src/leap/eip/test_vpnmanager.py
new file mode 100644
index 0000000..ebca652
--- /dev/null
+++ b/src/leap/eip/test_vpnmanager.py
@@ -0,0 +1,110 @@
+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 import vpnmanager  # import OpenVPNManager
+
+_system = platform.system()
+
+
+class NotImplementedError(Exception):
+    pass
+
+
+@patch('vpnmanager.OpenVPNManager.connect')
+class MockedOpenVPNManager(vpnmanager.OpenVPNManager):
+    def __init__(self, *args, **kwargs):
+        self.mock_UDSTelnet = Mock()
+        super(MockedOpenVPNManager, self).__init__(
+            *args, **kwargs)
+        self.tn = self.mock_UDSTelnet(self.host, self.port)
+
+
+class OpenVPNManagerTest(unittest.TestCase):
+
+    __name__ = "openvpn_manager_tests"
+
+    def setUp(self):
+        self.manager = MockedOpenVPNManager()
+
+    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 = MockedOpenVPNManager(port="42")
+        self.assertEqual(self.manager.port, 42)
+        self.manager = MockedOpenVPNManager()
+        self.assertEqual(self.manager.port, "unix")
+        self.manager = MockedOpenVPNManager(port="bad")
+        self.assertEqual(self.manager.port, None)
+
+    def test_connect_raises_missing_socket(self):
+        self.manager = vpnmanager.OpenVPNManager()
+        with self.assertRaises(vpnmanager.MissingSocketError):
+            self.manager.connect()
+
+    def test_connect(self):
+        pass
+        # XXX calls close
+        # calls UDSTelnet
+
+    # XXX
+    # tests to write:
+    # UDSTelnet (for real?)
+    # - raise MissingSocket
+    # - raise ConnectionRefusedError
+    # - test send command
+    #   - tries connect
+    #   - ... tries?
+    #   - ... calls _seek_to_eof
+    #   - ... read_until --> return value
+    #   - ...
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/src/leap/eip/vpnmanager.py b/src/leap/eip/vpnmanager.py
index caf7ab7..a4d872f 100644
--- a/src/leap/eip/vpnmanager.py
+++ b/src/leap/eip/vpnmanager.py
@@ -65,6 +65,10 @@ class OpenVPNManager(object):
         self.host = host
         if isinstance(port, str) and port.isdigit():
             port = int(port)
+        elif port == "unix":
+            port = "unix"
+        else:
+            port = None
         self.port = port
         self.password = password
         self.tn = None