186 lines
6.0 KiB
Python
186 lines
6.0 KiB
Python
# passerelle - uniform access to multiple data sources and services
|
|
# Copyright (C) 2019 Entr'ouvert
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it
|
|
# under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import os
|
|
|
|
import pytest
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.db import models
|
|
from django.utils.encoding import force_str
|
|
|
|
from passerelle.utils.sftp import SFTP, SFTPField, SFTPFormField
|
|
|
|
|
|
@pytest.fixture
|
|
def ssh_key():
|
|
with open(os.path.join(os.path.dirname(__file__), 'ssh_key'), 'rb') as fd:
|
|
yield fd.read()
|
|
|
|
|
|
@pytest.fixture
|
|
def ssh_key_with_password():
|
|
with open(os.path.join(os.path.dirname(__file__), 'ssh_key_with_password'), 'rb') as fd:
|
|
yield fd.read()
|
|
|
|
|
|
def test_http_url(sftpserver):
|
|
with pytest.raises(ValueError):
|
|
SFTP('https://coin.org/')
|
|
|
|
|
|
def test_missing_hostname(sftpserver):
|
|
with pytest.raises(ValueError):
|
|
SFTP('sftp://a:x@/hop/')
|
|
|
|
|
|
def test_sftp_ok(sftpserver):
|
|
with sftpserver.serve_content({'DILA': {'a.zip': 'a'}}):
|
|
with SFTP(
|
|
'sftp://john:doe@{server.host}:{server.port}/DILA/'.format(server=sftpserver)
|
|
).client() as sftp:
|
|
assert sftp.listdir() == ['a.zip']
|
|
|
|
|
|
def test_sftp_bad_paths(sftpserver):
|
|
with sftpserver.serve_content({'DILA': {'a.zip': 'a'}}):
|
|
with SFTP(
|
|
'sftp://john:doe@{server.host}:{server.port}/DILA/'.format(server=sftpserver)
|
|
).client() as sftp:
|
|
with pytest.raises(ValueError):
|
|
sftp.chdir('..')
|
|
with pytest.raises(ValueError):
|
|
sftp.chdir('/')
|
|
with pytest.raises(ValueError):
|
|
sftp.chdir('/coin')
|
|
|
|
|
|
def test_form_field(sftpserver, ssh_key, ssh_key_with_password):
|
|
from django import forms
|
|
|
|
class Form(forms.Form):
|
|
sftp = SFTPFormField(label='sftp')
|
|
|
|
with sftpserver.serve_content({'DILA': {'a.zip': 'a'}}):
|
|
url = 'sftp://john:doe@{server.host}:{server.port}/DILA/'.format(server=sftpserver)
|
|
|
|
form = Form(data={'sftp_0': 'http://coin.org'})
|
|
assert not form.is_valid()
|
|
assert 'Enter a valid URL.' in str(form.errors)
|
|
|
|
form = Form(data={'sftp_0': url})
|
|
assert form.is_valid()
|
|
sftp = form.cleaned_data['sftp']
|
|
assert isinstance(sftp, SFTP)
|
|
assert sftp.url == url
|
|
assert sftp.username == 'john'
|
|
assert sftp.password == 'doe'
|
|
assert sftp.hostname == sftpserver.host
|
|
assert sftp.port == sftpserver.port
|
|
assert sftp.path == 'DILA'
|
|
assert not sftp.private_key
|
|
with form.cleaned_data['sftp'].client() as sftp:
|
|
assert sftp.listdir() == ['a.zip']
|
|
|
|
form = Form(data={'sftp_0': url, 'sftp_2': force_str(ssh_key, 'ascii')})
|
|
assert form.is_valid()
|
|
sftp = form.cleaned_data['sftp']
|
|
assert isinstance(sftp, SFTP)
|
|
assert sftp.url == url
|
|
assert sftp.username == 'john'
|
|
assert sftp.password == 'doe'
|
|
assert sftp.hostname == sftpserver.host
|
|
assert sftp.port == sftpserver.port
|
|
assert sftp.path == 'DILA'
|
|
assert sftp.private_key
|
|
with form.cleaned_data['sftp'].client() as sftp:
|
|
assert sftp.listdir() == ['a.zip']
|
|
|
|
form = Form(
|
|
data={'sftp_0': url},
|
|
files={'sftp_1': SimpleUploadedFile('ssh_key', ssh_key, 'application/octet-stream')},
|
|
)
|
|
assert form.is_valid()
|
|
sftp = form.cleaned_data['sftp']
|
|
assert isinstance(sftp, SFTP)
|
|
assert sftp.url == url
|
|
assert sftp.username == 'john'
|
|
assert sftp.password == 'doe'
|
|
assert sftp.hostname == sftpserver.host
|
|
assert sftp.port == sftpserver.port
|
|
assert sftp.path == 'DILA'
|
|
assert sftp.private_key
|
|
with form.cleaned_data['sftp'].client() as sftp:
|
|
assert sftp.listdir() == ['a.zip']
|
|
|
|
form = Form(data={'sftp_0': url, 'sftp_2': force_str(ssh_key_with_password, 'ascii')})
|
|
assert not form.is_valid()
|
|
assert 'key invalid' in str(form.errors)
|
|
|
|
form = Form(
|
|
data={
|
|
'sftp_0': url,
|
|
'sftp_2': force_str(ssh_key_with_password, 'ascii'),
|
|
'sftp_3': 'coucou',
|
|
}
|
|
)
|
|
assert form.is_valid()
|
|
with form.cleaned_data['sftp'].client() as sftp:
|
|
assert sftp.listdir() == ['a.zip']
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_model(db):
|
|
from django.db import connection
|
|
|
|
class Model(models.Model):
|
|
sftp = SFTPField(blank=True)
|
|
|
|
class Meta:
|
|
app_label = 'test'
|
|
|
|
with connection.schema_editor() as editor:
|
|
editor.create_model(Model)
|
|
yield Model
|
|
editor.delete_model(Model)
|
|
|
|
|
|
def test_model_field(temp_model, ssh_key_with_password):
|
|
instance = temp_model.objects.create()
|
|
assert instance.sftp is None
|
|
|
|
url = 'sftp://john:doe@example.com:45/a/b'
|
|
sftp = SFTP(url=url, private_key_content=ssh_key_with_password, private_key_password='coucou')
|
|
instance.sftp = sftp
|
|
instance.save()
|
|
|
|
instance = temp_model.objects.get()
|
|
assert instance.sftp is not None
|
|
assert instance.sftp.url == url
|
|
assert instance.sftp.private_key is not None
|
|
instance.sftp = None
|
|
instance.save()
|
|
|
|
instance = temp_model.objects.get()
|
|
assert instance.sftp is None
|
|
instance.delete()
|
|
|
|
temp_model.objects.create(sftp=sftp)
|
|
|
|
instance = temp_model.objects.get()
|
|
assert instance.sftp is not None
|
|
assert instance.sftp.url == url
|
|
assert instance.sftp.private_key is not None
|