Merge branch 'master' of https://github.com/ReproNim/niceman

* 'master' of https://github.com/ReproNim/niceman:
  MNT: Update collections.Mapping for new location
  ENH: docker: Pull :latest if no tag or digest is specified
  MNT: Unpin docker-py
  NF: run: Extend --follow with optional stopping/deleting actions
  TST: run: Convert resource manager from module to function scope
  RF: orchestrators: Add `failed` keyword argument to log_failed()
  RF: orchestrators: Convert failed_subjobs property to method
  ENH: orchestrators: Do more fine-grained logging around fetch()
  RF: orchestrators: Move log_failed() calls earlier

reproman/distributions/tests/test_docker.py

@@ -22,7 +22,7 @@
 @mark.skipif_no_network
 @mark.skipif_no_docker_engine
 def test_docker_trace_tag():
-    client = docker.Client()
+    client = docker.APIClient()
     client.pull('alpine:3.6')
 
     tracer = DockerTracer()
@@ -40,7 +40,7 @@ def test_docker_trace_tag():
 @mark.skipif_no_network
 @mark.skipif_no_docker_engine
 def test_docker_trace_id():
-    client = docker.Client()
+    client = docker.APIClient()
     repo_id = 'sha256:f625bd3ff910ad2c68a405ccc5e294d2714fc8cfe7b5d80a8331c72ad5cc7630'
     name = 'alpine@' + repo_id
     client.pull(name)
@@ -62,7 +62,7 @@ def test_docker_trace_id():
 @mark.skipif_no_network
 @mark.skipif_no_docker_engine
 def test_docker_trace_local_image():
-    client = docker.Client()
+    client = docker.APIClient()
     client.pull('alpine:3.6')
     tracer = DockerTracer()
     # Test tracing a local image not saved in a repository
@@ -85,7 +85,7 @@ def test_docker_trace_local_image():
 @mark.skipif_no_docker_engine
 def test_docker_distribution():
 
-    client = docker.Client()
+    client = docker.APIClient()
     session = get_local_session()
 
     # Verify alpine:3.5 image is not stored locally

reproman/interface/run.py

@@ -11,6 +11,7 @@
 
 from argparse import REMAINDER
 import collections
+from collections.abc import Mapping
 import glob
 import logging
 import itertools
@@ -22,6 +23,7 @@
 from reproman.interface.base import Interface
 from reproman.interface.common_opts import resref_opt
 from reproman.interface.common_opts import resref_type_opt
+from reproman.support.constraints import EnsureChoice
 from reproman.support.jobs.local_registry import LocalRegistry
 from reproman.support.jobs.orchestrators import Orchestrator
 from reproman.support.jobs.orchestrators import ORCHESTRATORS
@@ -55,7 +57,7 @@ def update(d, u):
         Taken from https://stackoverflow.com/a/3233356
         """
         for k, v in u.items():
-            if isinstance(v, collections.Mapping):
+            if isinstance(v, Mapping):
                 d[k] = update(d.get(k, {}), v)
             else:
                 d[k] = v
@@ -286,7 +288,12 @@ class Run(Interface):
             depends on the orchestrator."""),
         follow=Parameter(
             args=("--follow",),
-            action="store_true",
+            metavar="ACTION",
+            const=True,
+            nargs="?",
+            constraints=EnsureChoice(False, True,
+                                     "stop", "stop-if-success",
+                                     "delete", "delete-if-success"),
             doc="""Continue to follow the submitted command instead of
             submitting it and detaching."""),
         command=Parameter(
@@ -387,7 +394,8 @@ def fmt(d):
                 resref_type = "name"
             else:
                 raise ValueError("No resource specified")
-        resource = get_manager().get_resource(resref, resref_type)
+        manager = get_manager()
+        resource = manager.get_resource(resref, resref_type)
 
         if "orchestrator" not in spec:
             # TODO: We could just set this as the default for the Parameter,
@@ -407,5 +415,25 @@ def fmt(d):
 
         if follow:
             orc.follow()
-            orc.fetch()
+            if follow is True:
+                remote_fn = None
+            else:
+                only_on_success = follow.endswith("-if-success")
+                do_delete = follow.split("-")[0] == "delete"
+
+                def remote_fn(res, failed):
+                    if failed and only_on_success:
+                        lgr.info("Not stopping%s resource '%s' "
+                                 "because there were failed jobs",
+                                 " or deleting" if do_delete else "",
+                                 res.name)
+                    else:
+                        lgr.info("Stopping%s resource '%s' after %s run",
+                                 " and deleting" if do_delete else "",
+                                 res.name,
+                                 "failed" if failed else "successful")
+                        manager.stop(res)
+                        if do_delete:
+                            manager.delete(res)
+            orc.fetch(on_remote_finish=remote_fn)
             lreg.unregister(orc.jobid)

reproman/interface/tests/test_create.py

@@ -26,7 +26,7 @@ def test_create_interface():
     Test creating an environment
     """
 
-    with patch('docker.Client') as client, \
+    with patch('docker.APIClient') as client, \
         patch('reproman.resource.ResourceManager.save_inventory'), \
         patch('reproman.resource.ResourceManager._get_inventory'), \
         swallow_logs(new_level=logging.DEBUG) as log:

reproman/interface/tests/test_delete.py

@@ -62,7 +62,7 @@ def test_delete_interface():
     Test deleting a resource.
     """
 
-    with patch('docker.Client', new_callable=mock_docker_client) as client, \
+    with patch('docker.APIClient', new_callable=mock_docker_client) as client, \
             patch('reproman.interface.delete.get_manager',
                 new=mock_get_manager), \
             swallow_logs(new_level=logging.DEBUG) as log:

reproman/interface/tests/test_install.py

@@ -20,7 +20,7 @@
 @mark.skipif_no_docker_dependencies
 def test_install_interface(demo1_spec):
 
-    with patch('docker.Client') as client, \
+    with patch('docker.APIClient') as client, \
         patch('reproman.distributions.debian.DebianDistribution.install_packages'), \
         patch('reproman.resource.ResourceManager._get_inventory') as get_inventory, \
         patch('requests.get') as requests, \

reproman/interface/tests/test_login.py

@@ -23,7 +23,7 @@ def test_login_interface():
     Test logging into an environment
     """
 
-    with patch('docker.Client') as client, \
+    with patch('docker.APIClient') as client, \
         patch('reproman.resource.ResourceManager._get_inventory') as get_inventory, \
         patch('dockerpty.start'), \
         swallow_logs(new_level=logging.DEBUG) as log:

reproman/interface/tests/test_ls.py

@@ -57,7 +57,7 @@ def ls_fn(resource_manager):
     def fn(*args, **kwargs):
         skipif.no_docker_dependencies()
         with contextlib.ExitStack() as stack:
-            stack.enter_context(patch("docker.Client"))
+            stack.enter_context(patch("docker.APIClient"))
             stack.enter_context(patch("reproman.interface.ls.get_manager",
                                       return_value=resource_manager))
             return ls(*args, **kwargs)

reproman/interface/tests/test_run.py

@@ -163,7 +163,7 @@ def test_resolve_batch_params_eq(tmpdir, params, spec):
 
 
 job_registry = fixtures.job_registry_fixture()
-resource_manager = fixtures.resource_manager_fixture(scope="module")
+resource_manager = fixtures.resource_manager_fixture(scope="function")
 
 
 @pytest.fixture(scope="function")
@@ -303,6 +303,32 @@ def test_run_and_follow(context):
     assert op.exists(op.join(path, "ok"))
 
 
+@pytest.mark.parametrize("action",
+                         ["stop", "stop-if-success",
+                          "delete", "delete-if-success"])
+def test_run_and_follow_action(context, action):
+    run = context["run_fn"]
+    expect = "does not support the 'stop' feature"
+    with swallow_logs(new_level=logging.INFO) as log:
+        run(command=["false"], resref="myshell",
+            follow=action)
+        if action.endswith("-if-success"):
+            assert expect not in log.out
+        else:
+            assert expect in log.out
+
+    if action != "delete":
+        with swallow_logs(new_level=logging.INFO) as log:
+            run(command=["true"], resref="myshell",
+                follow=action)
+            assert expect in log.out
+
+    if action.startswith("delete"):
+        resman = context["resource_manager"]
+        with pytest.raises(ResourceNotFoundError):
+            resman.get_resource("myshell", resref_type="name")
+
+
 def test_jobs_auto_fetch_with_query(context):
     path = context["directory"]
     run = context["run_fn"]

reproman/resource/docker_container.py

@@ -30,6 +30,16 @@
 lgr = logging.getLogger('reproman.resource.docker_container')
 
 
+def _image_latest_default(image):
+    # Given the restricted character set for names, the presence of ":" or "@"
+    # should be a reliable indication of a tag or digest, respectively. See
+    # - https://docs.docker.com/engine/reference/commandline/tag/#extended-description
+    # - vendor/github.com/docker/distribution/reference/regexp.go
+    if ":" not in image and "@" not in image:
+        image += ":latest"
+    return image
+
+
 @attr.s
 class DockerContainer(Resource):
     """
@@ -43,8 +53,10 @@ class DockerContainer(Resource):
     id = attrib()
     type = attrib(default='docker-container')
 
-    image = attrib(default='ubuntu:latest',
-        doc="Docker base image ID from which to create the running instance")
+    image = attrib(
+        default='ubuntu:latest',
+        doc="Docker base image ID from which to create the running instance",
+        converter=_image_latest_default)
     engine_url = attrib(default='unix:///var/run/docker.sock',
         doc="Docker server URL where engine is listening for connections")
     seccomp_unconfined = attrib(default=False,
@@ -71,7 +83,7 @@ def is_engine_running(base_url=None):
         """
         from requests.exceptions import ConnectionError
         try:
-            session = docker.Client(base_url=base_url)
+            session = docker.APIClient(base_url=base_url)
             session.info()
         except docker.errors.InvalidConfigFile as exc:
             lgr.error(
@@ -108,7 +120,7 @@ def connect(self):
         Open a connection to the environment.
         """
         # Open a client connection to the Docker engine.
-        self._client = docker.Client(base_url=self.engine_url)
+        self._client = docker.APIClient(base_url=self.engine_url)
 
         containers = []
         for container in self._client.containers(all=True):
@@ -280,7 +292,10 @@ def get(self, src_path, dest_path=None, uid=-1, gid=-1):
         dest_path = self._prepare_dest_path(src_path, dest_path)
         dest_dir = os.path.dirname(dest_path)
         stream, stat = self.client.get_archive(self.container, src_path)
-        tarball = tarfile.open(fileobj=io.BytesIO(stream.read()))
+        # get_archive() returns a generator with the content (in 2 MB chunks by
+        # default). Consider exposing those chunks as a stream rather than
+        # joining them.
+        tarball = tarfile.open(fileobj=io.BytesIO(b"".join(stream)))
         tarball.extractall(path=dest_dir)
         os.rename(os.path.join(dest_dir, src_basename), dest_path)
 

reproman/resource/tests/test_docker_container.py

@@ -31,7 +31,7 @@
 @mark.skipif_no_docker_dependencies
 def test_dockercontainer_class(resman):
 
-    with patch('docker.Client') as client, \
+    with patch('docker.APIClient') as client, \
         patch('dockerpty.start') as dockerpty, \
             swallow_logs(new_level=logging.DEBUG) as log:
 
@@ -175,3 +175,12 @@ def test_container_exists(setup_ubuntu):
     from ..docker_container import DockerContainer
     assert DockerContainer.is_container_running(setup_ubuntu['name'])
     assert not DockerContainer.is_container_running('foo')
+
+
+@mark.skipif_no_docker_dependencies
+def test_image_name_latest_default():
+    from ..docker_container import DockerContainer
+    for img, expected in [("debian:buster", "debian:buster"),
+                          ("busybox@ddeeaa", "busybox@ddeeaa"),
+                          ("busybox", "busybox:latest")]:
+        assert DockerContainer(name="cname", image=img).image == expected

reproman/resource/tests/test_session.py

@@ -418,7 +418,7 @@ def test_session_container(location, testing_container, check_methods):
     """
     cls = import_resource(*location)
     import docker
-    client = docker.Client()
+    client = docker.APIClient()
     container = next(c for c in client.containers()
                      if '/testing-container' in c['Names'])
     assert container