Infra
/
lemur
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Implement delete file and delete token tests

This commit is contained in:
Mathias Petermann 2020-11-11 11:23:24 +01:00
parent ae7a044b9c
commit e12ee1d89c
1 changed files with 54 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
lemur/plugins/lemur_sftp/tests/test_sftp.py
View File

@ -20,9 +20,9 @@ class TestSftp(unittest.TestCase):
@patch("lemur.plugins.lemur_sftp.plugin.paramiko") @patch("lemur.plugins.lemur_sftp.plugin.paramiko")
def test_upload_file_single_with_password(self, mock_paramiko): def test_upload_file_single_with_password(self, mock_paramiko):
dst_path = '/tmp/non-existant' dst_path = '/var/non-existent'
files = {'first-file': 'data'} files = {'first-file': 'data'}
options = [{'name': 'host', 'value': 'non-existant'}, {'name': 'port', 'value': '22'}, options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
{'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'}] {'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'}]
mock_sftp = Mock() mock_sftp = Mock()
@ -34,19 +34,19 @@ class TestSftp(unittest.TestCase):
self.sftp_destination.upload_file(dst_path, files, options) self.sftp_destination.upload_file(dst_path, files, options)
mock_sftp.open.assert_called_once() mock_sftp.open.assert_called_once_with('/var/non-existent/first-file', 'w')
handle = mock_sftp.open() handle = mock_sftp.open()
handle.write.assert_called_once_with('data') handle.write.assert_called_once_with('data')
mock_ssh.close.assert_called_once() mock_ssh.close.assert_called_once()
mock_ssh.connect.assert_called_with('non-existant', username='test_acme', port='22', mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
password='test_password') password='test_password')
@patch("lemur.plugins.lemur_sftp.plugin.paramiko") @patch("lemur.plugins.lemur_sftp.plugin.paramiko")
def test_upload_file_multiple_with_key(self, mock_paramiko): def test_upload_file_multiple_with_key(self, mock_paramiko):
dst_path = '/tmp/non-existant' dst_path = '/var/non-existent'
files = {'first-file': 'data', 'second-file': 'data2'} files = {'first-file': 'data', 'second-file': 'data2'}
options = [{'name': 'host', 'value': 'non-existant'}, {'name': 'port', 'value': '22'}, options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
{'name': 'user', 'value': 'test_acme'}, {'name': 'privateKeyPath', 'value': '/tmp/id_rsa'}, {'name': 'user', 'value': 'test_acme'}, {'name': 'privateKeyPath', 'value': '/var/id_rsa'},
{'name': 'privateKeyPass', 'value': 'ssh-key-password'}] {'name': 'privateKeyPass', 'value': 'ssh-key-password'}]
mock_sftp = Mock() mock_sftp = Mock()
@ -60,22 +60,22 @@ class TestSftp(unittest.TestCase):
self.sftp_destination.upload_file(dst_path, files, options) self.sftp_destination.upload_file(dst_path, files, options)
mock_sftp.open.assert_called() mock_sftp.open.assert_called_with('/var/non-existent/second-file', 'w')
handle = mock_sftp.open() handle = mock_sftp.open()
handle.write.assert_called_with('data2') handle.write.assert_called_with('data2')
mock_ssh.close.assert_called_once() mock_ssh.close.assert_called_once()
mock_paramiko.RSAKey.from_private_key_file.assert_called_with('/tmp/id_rsa', 'ssh-key-password') mock_paramiko.RSAKey.from_private_key_file.assert_called_with('/var/id_rsa', 'ssh-key-password')
mock_ssh.connect.assert_called_with('non-existant', username='test_acme', port='22', mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
pkey='ssh-rsa test-key') pkey='ssh-rsa test-key')
@patch("lemur.plugins.lemur_sftp.plugin.paramiko") @patch("lemur.plugins.lemur_sftp.plugin.paramiko")
def test_upload_acme_token(self, mock_paramiko): def test_upload_acme_token(self, mock_paramiko):
token_path = './well-known/acme-challenge/some-token-path' token_path = './well-known/acme-challenge/some-token-path'
token = 'token-data' token = 'token-data'
options = [{'name': 'host', 'value': 'non-existant'}, {'name': 'port', 'value': '22'}, options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
{'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'}, {'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'},
{'name': 'destinationPath', 'value': '/tmp/destination-path'}] {'name': 'destinationPath', 'value': '/var/destination-path'}]
mock_sftp = Mock() mock_sftp = Mock()
mock_sftp.open = mock_open() mock_sftp.open = mock_open()
@ -86,9 +86,49 @@ class TestSftp(unittest.TestCase):
self.sftp_destination.upload_acme_token(token_path, token, options) self.sftp_destination.upload_acme_token(token_path, token, options)
mock_sftp.open.assert_called_once() mock_sftp.open.assert_called_once_with('/var/destination-path/some-token-path', 'w')
handle = mock_sftp.open() handle = mock_sftp.open()
handle.write.assert_called_once_with('token-data') handle.write.assert_called_once_with('token-data')
mock_ssh.close.assert_called_once() mock_ssh.close.assert_called_once()
mock_ssh.connect.assert_called_with('non-existant', username='test_acme', port='22', mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
password='test_password')
@patch("lemur.plugins.lemur_sftp.plugin.paramiko")
def test_delete_file_with_password(self, mock_paramiko):
dst_path = '/var/non-existent'
files = {'first-file': None}
options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
{'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'}]
mock_sftp = Mock()
mock_ssh = mock_paramiko.SSHClient.return_value
mock_ssh.connect = MagicMock()
mock_ssh.open_sftp.return_value = mock_sftp
self.sftp_destination.delete_file(dst_path, files, options)
mock_sftp.remove.assert_called_once_with('/var/non-existent/first-file')
mock_ssh.close.assert_called_once()
mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
password='test_password')
@patch("lemur.plugins.lemur_sftp.plugin.paramiko")
def test_delete_acme_token(self, mock_paramiko):
token_path = './well-known/acme-challenge/some-token-path'
options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
{'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'},
{'name': 'destinationPath', 'value': '/var/destination-path'}]
mock_sftp = Mock()
mock_ssh = mock_paramiko.SSHClient.return_value
mock_ssh.connect = MagicMock()
mock_ssh.open_sftp.return_value = mock_sftp
self.sftp_destination.delete_acme_token(token_path, options)
mock_sftp.remove.assert_called_once_with('/var/destination-path/some-token-path')
mock_ssh.close.assert_called_once()
mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
password='test_password') password='test_password')