Bug fix for empty .shed.yml file and more testing.

planemo/shed.py

@@ -133,11 +133,14 @@ def shed_init(ctx, path, **kwds):

def shed_repo_config(path):
    shed_yaml_path = os.path.join(path, SHED_CONFIG_NAME)
    config = {}
    if os.path.exists(shed_yaml_path):
        with open(shed_yaml_path, "r") as f:
            return yaml.load(f)
    else:
        return {}
            config = yaml.load(f)

    if config is None:  # yaml may yield None
        config = {}
    return config


def tool_shed_client(ctx=None, **kwds):

tests/data/repos/multi_repos_nested/cat1/.shed.yml

@@ -0,0 +1,3 @@
name: cat1
owner: iuc
description: cat1 tool

tests/data/repos/multi_repos_nested/cat2/.shed.yml

@@ -0,0 +1,3 @@
name: cat2
owner: iuc
description: cat2 tool

tests/test_shed_upload.py

@@ -4,6 +4,7 @@
from os.path import exists, join
import os
import tarfile
import shutil

from .test_utils import CliShedTestCase

@@ -47,6 +48,16 @@ def test_cannont_recreate(self):
            self._check_exit_code(create_command)
            self._check_exit_code(create_command, exit_code=1)

    def test_upload_recusrive(self):
        with self._isolate_repo("multi_repos_nested") as f:
            upload_command = [
                "shed_upload", "-r", "--force_repository_creation"
            ]
            upload_command.extend(self._shed_args())
            self._check_exit_code(upload_command)
            self._verify_upload(f, ["cat1.xml", "macros.xml"], ["cat1"])
            self._verify_upload(f, ["cat2.xml", "macros.xml"], ["cat2"])

    def test_upload_filters_invalid_suite(self):
        with self._isolate_repo("suite_1") as f:
            # No .shed.yml, make sure to test it can infer type
@@ -85,11 +96,16 @@ def test_upload_not_filters_unrestricted(self):
            assert exists(join(target, "README.rst"))

    def _verify_single_uploaded(self, f):
        target = self._download_repo(f)
        assert exists(join(target, "cat.xml"))
        self._verify_upload(f, ["cat.xml"])

    def _verify_upload(self, f, download_files=[], download_args=[]):
        target = self._download_repo(f, download_args)
        for download_file in download_files:
            assert exists(join(target, download_file))

    def _download_repo(self, f):
    def _download_repo(self, f, download_args=[]):
        download_command = ["shed_download"]
        download_command.extend(download_args)
        download_command.extend(self._shed_args(read_only=True))
        self._check_exit_code(download_command)
        download = join(f, "shed_download.tar.gz")
@@ -98,6 +114,8 @@ def _download_repo(self, f):

    def _untar(self, f, path):
        target = join(f, "download")
        if exists(target):
            shutil.rmtree(target)
        os.makedirs(target)
        tar = tarfile.open(path, "r:gz")
        tar.extractall(path=target)

tests/test_utils.py

@@ -1,6 +1,7 @@
""" Provide abstractions over click testing of the
app and unittest.
"""
from __future__ import print_function
import contextlib
import os
from tempfile import mkdtemp
@@ -91,6 +92,14 @@ def _shed_args(self, read_only=False):
            args.extend(["--shed_key", "ignored"])
        return args

    def _print_shed_info(self):
        tsi = shed.tool_shed_client(
            None,
            shed_target=self.mock_shed.url,
            key="ignored",
        )
        print(tsi.repositories.get_repositories())


@contextlib.contextmanager
def mock_shed_client():