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
|
from __future__ import absolute_import
import time
import unittest
from unittest.mock import MagicMock, patch
from podman.libs.tunnel import Context, Portal, Tunnel
class TestTunnel(unittest.TestCase):
def setUp(self):
self.tunnel_01 = MagicMock(spec=Tunnel)
self.tunnel_02 = MagicMock(spec=Tunnel)
def test_portal_ops(self):
portal = Portal(sweap=500)
portal['unix:/01'] = self.tunnel_01
portal['unix:/02'] = self.tunnel_02
self.assertEqual(portal.get('unix:/01'), self.tunnel_01)
self.assertEqual(portal.get('unix:/02'), self.tunnel_02)
del portal['unix:/02']
with self.assertRaises(KeyError):
portal['unix:/02']
self.assertEqual(len(portal), 1)
def test_portal_reaping(self):
portal = Portal(sweap=0.5)
portal['unix:/01'] = self.tunnel_01
portal['unix:/02'] = self.tunnel_02
self.assertEqual(len(portal), 2)
for entry in portal:
self.assertIn(entry, (self.tunnel_01, self.tunnel_02))
time.sleep(1)
portal.reap()
self.assertEqual(len(portal), 0)
def test_portal_no_reaping(self):
portal = Portal(sweap=500)
portal['unix:/01'] = self.tunnel_01
portal['unix:/02'] = self.tunnel_02
portal.reap()
self.assertEqual(len(portal), 2)
for entry in portal:
self.assertIn(entry, (self.tunnel_01, self.tunnel_02))
@patch('subprocess.Popen')
@patch('os.path.exists', return_value=True)
@patch('weakref.finalize')
def test_tunnel(self, mock_finalize, mock_exists, mock_Popen):
mock_Popen.return_value.returncode = 0
context = Context(
'unix:/01',
'io.podman',
'/tmp/user/socket',
'/run/podman/socket',
'user',
'hostname',
None,
'~/.ssh/id_rsa',
)
tunnel = Tunnel(context).bore()
cmd = [
'ssh',
'-fNT',
'-q',
'-L',
'{}:{}'.format(context.local_socket, context.remote_socket),
'-i',
context.identity_file,
'{}@{}'.format(context.username, context.hostname),
]
mock_finalize.assert_called_once_with(tunnel, tunnel.close)
mock_exists.assert_called_once_with(context.local_socket)
mock_Popen.assert_called_once_with(cmd, close_fds=True)
|
