-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_metro_client.py
149 lines (116 loc) · 5.93 KB
/
test_metro_client.py
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
"""
Unit tests for SSH metro client.
Run with:
$ python setup.py test
"""
import unittest
import socket
import time
from sshmetroclient import main as metroclient
from fake_ssh_server import start_ssh_server
from fake_metro_server import start_metro_server
from multiprocessing import Process
try:
from unittest import mock
except ImportError:
# for Python 2.7 unittest does not have mock, so a mock library is used instead.
import mock
def get_free_port():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(('', 0))
port = s.getsockname()[1]
s.close()
return port
class CommandArgumentsTests(unittest.TestCase):
def setUp(self):
self.ssh_server_port = get_free_port()
self.metro_server_port = get_free_port()
self.ssh_server_thread = Process(target=start_ssh_server, args=(self.ssh_server_port,))
self.ssh_server_thread.start()
# start_ssh_server(self.ssh_server_port, {'test': 'abc123'})
# start_metro_server(self.metro_server_port, 'localhost', self.ssh_server_port)
self.metro_server_thread = Process(target=start_metro_server, args=(self.metro_server_port, 'localhost',
self.ssh_server_port))
self.metro_server_thread.start()
# Metro server needs a little time to be fully started
time.sleep(1)
def tearDown(self):
self.ssh_server_thread.terminate()
self.metro_server_thread.terminate()
@mock.patch('getpass.getpass')
def test_no_arguments_passed(self, getpass):
getpass.return_value = 'abc123'
with self.assertRaises(SystemExit):
metroclient.main(['main.py'])
@mock.patch('getpass.getpass')
def test_no_metro_server_argument(self, getpass):
getpass.return_value = 'abc123'
with self.assertRaises(SystemExit):
metroclient.main(['main.py', 'testuser@localmachine'])
@mock.patch('getpass.getpass')
def test_no_target_machine_argument(self, getpass):
getpass.return_value = 'abc123'
with self.assertRaises(SystemExit):
metroclient.main(['main.py', 'metroserver.com'])
@mock.patch('getpass.getpass')
def test_incorrect_format_target_machine_string_no_user(self, getpass):
getpass.return_value = 'abc123'
with self.assertRaises(SystemExit):
metroclient.main(['main.py', '@localmachine', 'localhost:%d' % self.metro_server_port])
@mock.patch('getpass.getpass')
def test_incorrect_format_target_machine_string_missing_port(self, getpass):
getpass.return_value = 'abc123'
with self.assertRaises(SystemExit):
metroclient.main(['main.py', 'user@localmachine:', 'localhost:%d' % self.metro_server_port])
@mock.patch('getpass.getpass')
def test_incorrect_format_target_machine_string_two_colons(self, getpass):
getpass.return_value = 'abc123'
with self.assertRaises(SystemExit):
metroclient.main(['main.py', 'user@localmachine:55:33', 'localhost:%d' % self.metro_server_port])
@mock.patch('getpass.getpass')
def test_incorrect_format_target_machine_string_two_slashes(self, getpass):
getpass.return_value = 'abc123'
with self.assertRaises(SystemExit):
metroclient.main(['main.py', 'user/password1/password2@localmachine', 'localhost:%d' %
self.metro_server_port])
@mock.patch('getpass.getpass')
def test_incorrect_format_target_machine_slash_misplaced(self, getpass):
getpass.return_value = 'abc123'
with self.assertRaises(SystemExit):
metroclient.main(['main.py', 'username@localmachine/misplaced', 'localhost:%d' % self.metro_server_port])
@mock.patch('getpass.getpass')
def test_incorrect_format_target_machine_string_two_at(self, getpass):
getpass.return_value = 'abc123'
with self.assertRaises(SystemExit):
metroclient.main(['main.py', 'user@machine1@machine2', 'localhost:%d' % self.metro_server_port])
def test_password_included_in_target_machine_string(self):
metroclient.main(['main.py', 'user/test123@localhost:2222', 'localhost:%d' % self.metro_server_port])
class ConnectingToMetroServer(unittest.TestCase):
def setUp(self):
self.ssh_server_port = get_free_port()
self.metro_server_port = get_free_port()
self.ssh_server_thread = Process(target=start_ssh_server, args=(self.ssh_server_port,))
self.ssh_server_thread.start()
# start_ssh_server(self.ssh_server_port, {'test': 'abc123'})
# start_metro_server(self.metro_server_port, 'localhost', self.ssh_server_port)
self.metro_server_thread = Process(target=start_metro_server, args=(self.metro_server_port, 'localhost',
self.ssh_server_port))
self.metro_server_thread.start()
# Metro server needs a little time to be fully started
time.sleep(1)
def tearDown(self):
self.ssh_server_thread.terminate()
self.metro_server_thread.terminate()
def test_successful_connection_password_included(self):
metroclient.main(['main.py', 'user/test123@localhost:2222', 'localhost:%d' % self.metro_server_port])
@mock.patch('getpass.getpass')
def test_successfu_connection_password_given_later(self, getpass):
getpass.return_value = 'abc123'
metroclient.main(['main.py', 'user@localhost:2222', 'localhost:%d' % self.metro_server_port])
def test_connection_failure_to_ssh_tunnel(self):
with self.assertRaises(SystemExit):
metroclient.main(['main.py', 'user/[email protected]:2222', 'localhost:%d' %
self.metro_server_port])
def test_connection_to_not_refused_port(self):
with self.assertRaises(SystemExit):
metroclient.main(['main.py', 'user/test123@localhost:15', 'localhost:%d' % self.metro_server_port])