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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
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.host = host
self.manager = MockedOpenVPNConnection(host=host)
def tearDown(self):
pass
def doCleanups(self):
super(BaseLeapTest, self).doCleanups()
self.cleanupSocketDir()
def cleanupSocketDir(self):
# remove the socket folder.
# XXX only if posix. in win, host is localhost, so nothing
# has to be done.
if self.host:
folder, fpath = os.path.split(self.host)
try:
assert folder.startswith('/tmp/leap-tmp') # safety check
shutil.rmtree(folder)
except:
self.fail("could not remove temp file")
del self.manager
#
# tests
#
def test_detect_vpn(self):
# XXX review, not sure if captured all the logic
# while fixing. kali.
openvpn_connection = openvpnconnection.OpenVPNConnection()
with patch.object(psutil, "process_iter") as mocked_psutil:
mocked_process = Mock()
mocked_process.name = "openvpn"
mocked_process.cmdline = ["openvpn", "-foo", "-bar", "-gaaz"]
mocked_psutil.return_value = [mocked_process]
with self.assertRaises(eipexceptions.OpenVPNAlreadyRunning):
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):
oldmanager = self.manager
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)
self.manager = oldmanager
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()
|
