summaryrefslogtreecommitdiff
path: root/src/leap/eip/tests/test_openvpnconnection.py
blob: 61769f04e35901be4cdf79de8afc7ab732e02b31 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
import logging
import os
import platform
import psutil
import shutil
#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 config as eipconfig
from leap.eip import openvpnconnection
from leap.eip import exceptions as eipexceptions
from leap.eip.udstelnet import UDSTelnet
from leap.testing.basetest import BaseLeapTest

_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(BaseLeapTest):

    __name__ = "vpnconnection_tests"

    def setUp(self):
        # XXX this will have to change for win, host=localhost
        host = eipconfig.get_socket_path()
        self.manager = MockedOpenVPNConnection(host=host)

    def tearDown(self):
        # remove the socket folder.
        # XXX only if posix. in win, host is localhost, so nothing
        # has to be done.
        if self.manager.host:
            folder, fpath = os.path.split(self.manager.host)
            assert folder.startswith('/tmp/leap-tmp')  # safety check
            shutil.rmtree(folder)

        del self.manager

    #
    # tests
    #

    def test_detect_vpn(self):
        openvpn_connection = openvpnconnection.OpenVPNConnection()
        with patch.object(psutil, "get_process_list") as mocked_psutil:
            with self.assertRaises(eipexceptions.OpenVPNAlreadyRunning):
                mocked_process = Mock()
                mocked_process.name = "openvpn"
                mocked_psutil.return_value = [mocked_process]
                openvpn_connection._check_if_running_instance()
        openvpn_connection._check_if_running_instance()

    @unittest.skipIf(_system == "Windows", "lin/mac only")
    def test_lin_mac_default_init(self):
        """
        check default host for management iface
        """
        self.assertTrue(self.manager.host.startswith('/tmp/leap-tmp'))
        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_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()