mirror of
https://github.com/containers/podman.git
synced 2025-08-06 19:44:14 +08:00
Add unittests and fix bugs
* Improved error messages * Improved checking of user input Signed-off-by: Jhon Honce <jhonce@redhat.com> Closes: #978 Approved by: mheon
This commit is contained in:
79
contrib/python/test/test_tunnel.py
Normal file
79
contrib/python/test/test_tunnel.py
Normal file
@ -0,0 +1,79 @@
|
||||
from __future__ import absolute_import
|
||||
|
||||
import time
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import podman
|
||||
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):
|
||||
context = Context(
|
||||
'unix:/01',
|
||||
'io.projectatomic.podman',
|
||||
'/tmp/user/socket',
|
||||
'/run/podman/socket',
|
||||
'user',
|
||||
'hostname',
|
||||
'~/.ssh/id_rsa',
|
||||
)
|
||||
tunnel = Tunnel(context).bore('unix:/01')
|
||||
|
||||
cmd = [
|
||||
'ssh',
|
||||
'-nNT',
|
||||
'-L',
|
||||
'{}:{}'.format(context.local_socket, context.remote_socket),
|
||||
'-i',
|
||||
context.identity_file,
|
||||
'ssh://{}@{}'.format(context.username, context.hostname),
|
||||
]
|
||||
|
||||
mock_finalize.assert_called_once_with(tunnel, tunnel.close, 'unix:/01')
|
||||
mock_exists.assert_called_once_with(context.local_socket)
|
||||
mock_Popen.assert_called_once_with(cmd, close_fds=True)
|
Reference in New Issue
Block a user