passerelle/tests/test_utils_sftp.py

# 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