summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorkali <kali@leap.se>2012-08-22 04:11:44 +0900
committerkali <kali@leap.se>2012-08-22 04:11:44 +0900
commit04cf64af3702ab85a670efe6850c60f20bbf7eb0 (patch)
tree17fcd14f86351a88abfee056c001f2266d2a3bc5
parent78c83ab3bb7f95564bdc537d29f6d278b1710b17 (diff)
conductor tests
-rw-r--r--src/leap/eip/test_conductor.py180
-rw-r--r--src/leap/util/test_fileutil.py10
2 files changed, 184 insertions, 6 deletions
diff --git a/src/leap/eip/test_conductor.py b/src/leap/eip/test_conductor.py
new file mode 100644
index 0000000..51772b7
--- /dev/null
+++ b/src/leap/eip/test_conductor.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/util/test_fileutil.py b/src/leap/util/test_fileutil.py
index 849deca..f5dbe10 100644
--- a/src/leap/util/test_fileutil.py
+++ b/src/leap/util/test_fileutil.py
@@ -52,8 +52,7 @@ class FileUtilTest(unittest.TestCase):
def test_is_user_executable(self):
"""
- test that a 700 file
- is an 700 file. kindda oximoronic, but...
+ touch_exec_file creates in mode 700?
"""
# XXX could check access X_OK
@@ -63,10 +62,10 @@ class FileUtilTest(unittest.TestCase):
def test_which(self):
"""
+ which implementation ok?
not a very reliable test,
but I cannot think of anything smarter now
I guess it's highly improbable that copy
- command is somewhere else..?
"""
# XXX yep, we can change the syspath
# for the test... !
@@ -78,7 +77,7 @@ class FileUtilTest(unittest.TestCase):
def test_mkdir_p(self):
"""
- test our mkdir -p implementation
+ our own mkdir -p implementation ok?
"""
testdir = self.get_file_path(
os.path.join('test', 'foo', 'bar'))
@@ -88,8 +87,7 @@ class FileUtilTest(unittest.TestCase):
def test_check_and_fix_urw_only(self):
"""
- test function that fixes perms on
- files that should be rw only for owner
+ ensure check_and_fix_urx_only ok?
"""
fp = self.touch_exec_file()
mode = self.get_mode(fp)