GNOME.org
 

[Notes] [Git][BuildStream/buildstream][valentindavid/netrc] Add tests for .netrc support in tar/zip/remote plugins

Title: GitLab

Valentin David pushed to branch valentindavid/netrc at BuildStream / buildstream

Commits:

7 changed files:

Changes:

  • dev-requirements.txt
    ... ... @@ -9,3 +9,4 @@ pytest-pep8
    9 9 
     pytest-pylint
    10 10 
     pytest-xdist
    11 11 
     pytest-timeout
    12 
    +pyftpdlib

  • tests/sources/remote.py
    ... ... @@ -5,6 +5,7 @@ import pytest
    5 5 
     from buildstream._exceptions import ErrorDomain
    6 6 
     from buildstream import _yaml
    7 7 
     from tests.testutils import cli
    8 
    +from tests.testutils.file_server import create_file_server
    8 9
    9 10 
     DATA_DIR = os.path.join(
    10 11 
         os.path.dirname(os.path.realpath(__file__)),
    ... ... @@ -22,6 +23,16 @@ def generate_project(project_dir, tmpdir):
    22 23 
         }, project_file)
    23 24
    24 25
    26 
    +def generate_project_file_server(server, project_dir):
    27 
    +    project_file = os.path.join(project_dir, "project.conf")
    28 
    +    _yaml.dump({
    29 
    +        'name': 'foo',
    30 
    +        'aliases': {
    31 
    +            'tmpdir': server.base_url()
    32 
    +        }
    33 
    +    }, project_file)
    34 
    +
    35 
    +
    25 36 
     # Test that without ref, consistency is set appropriately.
    26 37 
     @pytest.mark.datafiles(os.path.join(DATA_DIR, 'no-ref'))
    27 38 
     def test_no_ref(cli, tmpdir, datafiles):
    ... ... @@ -164,3 +175,35 @@ def test_executable(cli, tmpdir, datafiles):
    164 175 
         assert (mode & stat.S_IEXEC)
    165 176 
         # Assert executable by anyone
    166 177 
         assert(mode & (stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH))
    178 
    +
    179 
    +
    180 
    +@pytest.mark.parametrize('server_type', ('FTP', 'HTTP'))
    181 
    +@pytest.mark.datafiles(os.path.join(DATA_DIR, 'single-file'))
    182 
    +def test_use_netrc(cli, datafiles, server_type, tmpdir):
    183 
    +    fake_home = os.path.join(str(tmpdir), 'fake_home')
    184 
    +    os.makedirs(fake_home, exist_ok=True)
    185 
    +    project = str(datafiles)
    186 
    +    checkoutdir = os.path.join(str(tmpdir), 'checkout')
    187 
    +
    188 
    +    os.environ['HOME'] = fake_home
    189 
    +    with open(os.path.join(fake_home, '.netrc'), 'wb') as f:
    190 
    +        os.fchmod(f.fileno(), 0o700)
    191 
    +        f.write(b'machine 127.0.0.1\n')
    192 
    +        f.write(b'login testuser\n')
    193 
    +        f.write(b'password 12345\n')
    194 
    +
    195 
    +    with create_file_server(server_type) as server:
    196 
    +        server.add_user('testuser', '12345', project)
    197 
    +        generate_project_file_server(server, project)
    198 
    +
    199 
    +        server.start()
    200 
    +
    201 
    +        result = cli.run(project=project, args=['fetch', 'target.bst'])
    202 
    +        result.assert_success()
    203 
    +        result = cli.run(project=project, args=['build', 'target.bst'])
    204 
    +        result.assert_success()
    205 
    +        result = cli.run(project=project, args=['checkout', 'target.bst', checkoutdir])
    206 
    +        result.assert_success()
    207 
    +
    208 
    +        checkout_file = os.path.join(checkoutdir, 'file')
    209 
    +        assert(os.path.exists(checkout_file))

  • tests/sources/tar.py
    ... ... @@ -8,6 +8,7 @@ from shutil import copyfile, rmtree
    8 8 
     from buildstream._exceptions import ErrorDomain
    9 9 
     from buildstream import _yaml
    10 10 
     from tests.testutils import cli
    11 
    +from tests.testutils.file_server import create_file_server
    11 12 
     from tests.testutils.site import HAVE_LZIP
    12 13 
     from . import list_dir_contents
    13 14
    ... ... @@ -49,6 +50,16 @@ def generate_project(project_dir, tmpdir):
    49 50 
         }, project_file)
    50 51
    51 52
    53 
    +def generate_project_file_server(server, project_dir):
    54 
    +    project_file = os.path.join(project_dir, "project.conf")
    55 
    +    _yaml.dump({
    56 
    +        'name': 'foo',
    57 
    +        'aliases': {
    58 
    +            'tmpdir': server.base_url()
    59 
    +        }
    60 
    +    }, project_file)
    61 
    +
    62 
    +
    52 63 
     # Test that without ref, consistency is set appropriately.
    53 64 
     @pytest.mark.datafiles(os.path.join(DATA_DIR, 'no-ref'))
    54 65 
     def test_no_ref(cli, tmpdir, datafiles):
    ... ... @@ -302,3 +313,43 @@ def test_read_only_dir(cli, tmpdir, datafiles):
    302 313 
                 else:
    303 314 
                     os.remove(path)
    304 315 
             rmtree(str(tmpdir), onerror=make_dir_writable)
    316 
    +
    317 
    +@pytest.mark.parametrize('server_type', ('FTP', 'HTTP'))
    318 
    +@pytest.mark.datafiles(os.path.join(DATA_DIR, 'fetch'))
    319 
    +def test_use_netrc(cli, datafiles, server_type, tmpdir):
    320 
    +    file_server_files = os.path.join(str(tmpdir), 'file_server')
    321 
    +    fake_home = os.path.join(str(tmpdir), 'fake_home')
    322 
    +    os.makedirs(file_server_files, exist_ok=True)
    323 
    +    os.makedirs(fake_home, exist_ok=True)
    324 
    +    project = str(datafiles)
    325 
    +    checkoutdir = os.path.join(str(tmpdir), 'checkout')
    326 
    +
    327 
    +    os.environ['HOME'] = fake_home
    328 
    +    with open(os.path.join(fake_home, '.netrc'), 'wb') as f:
    329 
    +        os.fchmod(f.fileno(), 0o700)
    330 
    +        f.write(b'machine 127.0.0.1\n')
    331 
    +        f.write(b'login testuser\n')
    332 
    +        f.write(b'password 12345\n')
    333 
    +
    334 
    +    with create_file_server(server_type) as server:
    335 
    +        server.add_user('testuser', '12345', file_server_files)
    336 
    +        generate_project_file_server(server, project)
    337 
    +
    338 
    +        src_tar = os.path.join(file_server_files, 'a.tar.gz')
    339 
    +        _assemble_tar(os.path.join(str(datafiles), 'content'), 'a', src_tar)
    340 
    +
    341 
    +        server.start()
    342 
    +
    343 
    +        result = cli.run(project=project, args=['track', 'target.bst'])
    344 
    +        result.assert_success()
    345 
    +        result = cli.run(project=project, args=['fetch', 'target.bst'])
    346 
    +        result.assert_success()
    347 
    +        result = cli.run(project=project, args=['build', 'target.bst'])
    348 
    +        result.assert_success()
    349 
    +        result = cli.run(project=project, args=['checkout', 'target.bst', checkoutdir])
    350 
    +        result.assert_success()
    351 
    +
    352 
    +        original_dir = os.path.join(str(datafiles), 'content', 'a')
    353 
    +        original_contents = list_dir_contents(original_dir)
    354 
    +        checkout_contents = list_dir_contents(checkoutdir)
    355 
    +        assert(checkout_contents == original_contents)

  • tests/sources/zip.py
    ... ... @@ -5,6 +5,7 @@ import zipfile
    5 5 
     from buildstream._exceptions import ErrorDomain
    6 6 
     from buildstream import _yaml
    7 7 
     from tests.testutils import cli
    8 
    +from tests.testutils.file_server import create_file_server
    8 9 
     from . import list_dir_contents
    9 10
    10 11 
     DATA_DIR = os.path.join(
    ... ... @@ -35,6 +36,16 @@ def generate_project(project_dir, tmpdir):
    35 36 
         }, project_file)
    36 37
    37 38
    39 
    +def generate_project_file_server(server, project_dir):
    40 
    +    project_file = os.path.join(project_dir, "project.conf")
    41 
    +    _yaml.dump({
    42 
    +        'name': 'foo',
    43 
    +        'aliases': {
    44 
    +            'tmpdir': server.base_url()
    45 
    +        }
    46 
    +    }, project_file)
    47 
    +
    48 
    +
    38 49 
     # Test that without ref, consistency is set appropriately.
    39 50 
     @pytest.mark.datafiles(os.path.join(DATA_DIR, 'no-ref'))
    40 51 
     def test_no_ref(cli, tmpdir, datafiles):
    ... ... @@ -176,3 +187,44 @@ def test_stage_explicit_basedir(cli, tmpdir, datafiles):
    176 187 
         original_contents = list_dir_contents(original_dir)
    177 188 
         checkout_contents = list_dir_contents(checkoutdir)
    178 189 
         assert(checkout_contents == original_contents)
    190 
    +
    191 
    +
    192 
    +@pytest.mark.parametrize('server_type', ('FTP', 'HTTP'))
    193 
    +@pytest.mark.datafiles(os.path.join(DATA_DIR, 'fetch'))
    194 
    +def test_use_netrc(cli, datafiles, server_type, tmpdir):
    195 
    +    file_server_files = os.path.join(str(tmpdir), 'file_server')
    196 
    +    fake_home = os.path.join(str(tmpdir), 'fake_home')
    197 
    +    os.makedirs(file_server_files, exist_ok=True)
    198 
    +    os.makedirs(fake_home, exist_ok=True)
    199 
    +    project = str(datafiles)
    200 
    +    checkoutdir = os.path.join(str(tmpdir), 'checkout')
    201 
    +
    202 
    +    os.environ['HOME'] = fake_home
    203 
    +    with open(os.path.join(fake_home, '.netrc'), 'wb') as f:
    204 
    +        os.fchmod(f.fileno(), 0o700)
    205 
    +        f.write(b'machine 127.0.0.1\n')
    206 
    +        f.write(b'login testuser\n')
    207 
    +        f.write(b'password 12345\n')
    208 
    +
    209 
    +    with create_file_server(server_type) as server:
    210 
    +        server.add_user('testuser', '12345', file_server_files)
    211 
    +        generate_project_file_server(server, project)
    212 
    +
    213 
    +        src_zip = os.path.join(file_server_files, 'a.zip')
    214 
    +        _assemble_zip(os.path.join(str(datafiles), 'content'), src_zip)
    215 
    +
    216 
    +        server.start()
    217 
    +
    218 
    +        result = cli.run(project=project, args=['track', 'target.bst'])
    219 
    +        result.assert_success()
    220 
    +        result = cli.run(project=project, args=['fetch', 'target.bst'])
    221 
    +        result.assert_success()
    222 
    +        result = cli.run(project=project, args=['build', 'target.bst'])
    223 
    +        result.assert_success()
    224 
    +        result = cli.run(project=project, args=['checkout', 'target.bst', checkoutdir])
    225 
    +        result.assert_success()
    226 
    +
    227 
    +        original_dir = os.path.join(str(datafiles), 'content', 'a')
    228 
    +        original_contents = list_dir_contents(original_dir)
    229 
    +        checkout_contents = list_dir_contents(checkoutdir)
    230 
    +        assert(checkout_contents == original_contents)

  • tests/testutils/file_server.py
    1 
    +from contextlib import contextmanager
    2 
    +
    3 
    +from .ftp_server import SimpleFtpServer
    4 
    +from .http_server import SimpleHttpServer
    5 
    +
    6 
    +
    7 
    +@contextmanager
    8 
    +def create_file_server(file_server_type):
    9 
    +    if file_server_type == 'FTP':
    10 
    +        server = SimpleFtpServer()
    11 
    +    elif file_server_type == 'HTTP':
    12 
    +        server = SimpleHttpServer()
    13 
    +    else:
    14 
    +        assert False
    15 
    +
    16 
    +    try:
    17 
    +        yield server
    18 
    +    finally:
    19 
    +        server.stop()

  • tests/testutils/ftp_server.py
    1 
    +import multiprocessing
    2 
    +
    3 
    +from pyftpdlib.authorizers import DummyAuthorizer
    4 
    +from pyftpdlib.handlers import FTPHandler
    5 
    +from pyftpdlib.servers import FTPServer
    6 
    +
    7 
    +
    8 
    +class SimpleFtpServer(multiprocessing.Process):
    9 
    +    def __init__(self):
    10 
    +        super().__init__()
    11 
    +        self.authorizer = DummyAuthorizer()
    12 
    +        handler = FTPHandler
    13 
    +        handler.authorizer = self.authorizer
    14 
    +        self.server = FTPServer(('127.0.0.1', 0), handler)
    15 
    +
    16 
    +    def run(self):
    17 
    +        self.server.serve_forever()
    18 
    +
    19 
    +    def stop(self):
    20 
    +        self.server.close_all()
    21 
    +        self.server.close()
    22 
    +        self.terminate()
    23 
    +        self.join()
    24 
    +
    25 
    +    def allow_anonymous(self, cwd):
    26 
    +        self.authorizer.add_anonymous(cwd)
    27 
    +
    28 
    +    def add_user(self, user, password, cwd):
    29 
    +        self.authorizer.add_user(user, password, cwd, perm='elradfmwMT')
    30 
    +
    31 
    +    def base_url(self):
    32 
    +        return 'ftp://127.0.0.1:{}'.format(self.server.address[1])

  • tests/testutils/http_server.py
    1 
    +import multiprocessing
    2 
    +import os
    3 
    +import posixpath
    4 
    +import html
    5 
    +import threading
    6 
    +import base64
    7 
    +from http.server import SimpleHTTPRequestHandler, HTTPServer, HTTPStatus
    8 
    +
    9 
    +
    10 
    +class Unauthorized(Exception):
    11 
    +    pass
    12 
    +
    13 
    +
    14 
    +class RequestHandler(SimpleHTTPRequestHandler):
    15 
    +
    16 
    +    def get_root_dir(self):
    17 
    +        authorization = self.headers.get('authorization')
    18 
    +        if not authorization:
    19 
    +            if not self.server.anonymous_dir:
    20 
    +                raise Unauthorized('unauthorized')
    21 
    +            return self.server.anonymous_dir
    22 
    +        else:
    23 
    +            authorization = authorization.split()
    24 
    +            if len(authorization) != 2 or authorization[0].lower() != 'basic':
    25 
    +                raise Unauthorized('unauthorized')
    26 
    +            try:
    27 
    +                decoded = base64.decodebytes(authorization[1].encode('ascii'))
    28 
    +                user, password = decoded.decode('ascii').split(':')
    29 
    +                expected_password, directory = self.server.users[user]
    30 
    +                if password == expected_password:
    31 
    +                    return directory
    32 
    +            except:
    33 
    +                raise Unauthorized('unauthorized')
    34 
    +            return None
    35 
    +
    36 
    +    def unauthorized(self):
    37 
    +        shortmsg, longmsg = self.responses[HTTPStatus.UNAUTHORIZED]
    38 
    +        self.send_response(HTTPStatus.UNAUTHORIZED, shortmsg)
    39 
    +        self.send_header('Connection', 'close')
    40 
    +
    41 
    +        content = (self.error_message_format % {
    42 
    +            'code': HTTPStatus.UNAUTHORIZED,
    43 
    +            'message': html.escape(longmsg, quote=False),
    44 
    +            'explain': html.escape(longmsg, quote=False)
    45 
    +        })
    46 
    +        body = content.encode('UTF-8', 'replace')
    47 
    +        self.send_header('Content-Type', self.error_content_type)
    48 
    +        self.send_header('Content-Length', str(len(body)))
    49 
    +        self.send_header('WWW-Authenticate', 'Basic realm="{}"'.format(self.server.realm))
    50 
    +        self.end_headers()
    51 
    +        self.end_headers()
    52 
    +
    53 
    +        if self.command != 'HEAD' and body:
    54 
    +            self.wfile.write(body)
    55 
    +
    56 
    +    def do_GET(self):
    57 
    +        try:
    58 
    +            super().do_GET()
    59 
    +        except Unauthorized:
    60 
    +            self.unauthorized()
    61 
    +
    62 
    +    def do_HEAD(self):
    63 
    +        try:
    64 
    +            super().do_HEAD()
    65 
    +        except Unauthorized:
    66 
    +            self.unauthorized()
    67 
    +
    68 
    +    def translate_path(self, path):
    69 
    +        path = path.split('?', 1)[0]
    70 
    +        path = path.split('#', 1)[0]
    71 
    +        path = posixpath.normpath(path)
    72 
    +        assert(posixpath.isabs(path))
    73 
    +        path = posixpath.relpath(path, '/')
    74 
    +        return os.path.join(self.get_root_dir(), path)
    75 
    +
    76 
    +
    77 
    +class AuthHTTPServer(HTTPServer):
    78 
    +    def __init__(self, *args, **kwargs):
    79 
    +        self.users = {}
    80 
    +        self.anonymous_dir = None
    81 
    +        self.realm = 'Realm'
    82 
    +        super().__init__(*args, **kwargs)
    83 
    +
    84 
    +
    85 
    +class SimpleHttpServer(multiprocessing.Process):
    86 
    +    def __init__(self):
    87 
    +        self.__stop = multiprocessing.Queue()
    88 
    +        super().__init__()
    89 
    +        self.server = AuthHTTPServer(('127.0.0.1', 0), RequestHandler)
    90 
    +        self.started = False
    91 
    +
    92 
    +    def start(self):
    93 
    +        self.started = True
    94 
    +        super().start()
    95 
    +
    96 
    +    def run(self):
    97 
    +        t = threading.Thread(target=self.server.serve_forever)
    98 
    +        t.start()
    99 
    +        self.__stop.get()
    100 
    +        self.server.shutdown()
    101 
    +        t.join()
    102 
    +
    103 
    +    def stop(self):
    104 
    +        if not self.started:
    105 
    +            return
    106 
    +        self.__stop.put(None)
    107 
    +        self.terminate()
    108 
    +        self.join()
    109 
    +
    110 
    +    def allow_anonymous(self, cwd):
    111 
    +        self.server.anonymous_dir = cwd
    112 
    +
    113 
    +    def add_user(self, user, password, cwd):
    114 
    +        self.server.users[user] = (password, cwd)
    115 
    +
    116 
    +    def base_url(self):
    117 
    +        return 'http://127.0.0.1:{}'.format(self.server.server_port)

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.com. If you'd like to receive fewer emails, you can adjust your notification settings.

    [Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]