Allow remote .conda artifact listing

.github/workflows/tests.yml

@@ -46,7 +46,7 @@ jobs:
       # conda setup requires this special shell
       shell: bash -l {0}
       run: |
-        conda build -c conda conda.recipe
+        conda build -c conda -c conda-forge conda.recipe
 
 # TODO: Re-enable codecov when we figure out how to grab coverage from the conda build
 #       environment

conda.recipe/meta.yaml

@@ -27,15 +27,18 @@ requirements:
     - python
     - zstandard >=0.15
     - conda-package-streaming >=0.9.0
+    - requests
 
 test:
   source_files:
     - tests
   requires:
     - mock
+    - bottle
     - pytest
     - pytest-cov
     - pytest-mock
+    - pytest-xprocess
   imports:
     - conda_package_handling
     - conda_package_handling.api

news/254-list-remote

@@ -0,0 +1,19 @@
+### Enhancements
+
+* Allow `cph list` on remote `.conda` artifact URLs. (#252 via #254)
+
+### Bug fixes
+
+* <news item>
+
+### Deprecations
+
+* <news item>
+
+### Docs
+
+* <news item>
+
+### Other
+
+* <news item>

setup.py

@@ -34,6 +34,6 @@
             "myst-parser",
             "mdit-py-plugins>=0.3.0",
         ],
-        "test": ["mock", "pytest", "pytest-cov", "pytest-mock"],
+        "test": ["mock", "pytest", "pytest-cov", "pytest-mock", "pytest-xprocess", "bottle"],
     },
 )

src/conda_package_handling/cli.py

@@ -109,7 +109,7 @@ def build_parser():
         aliases=["l"],
         help="List package contents like `python -m tarfile --list ...` would do.",
     )
-    list_parser.add_argument("archive_path", help="path to archive to inspect")
+    list_parser.add_argument("archive_path", help="path or URL to archive to inspect")
     list_parser.add_argument(
         "-v",
         "--verbose",

src/conda_package_handling/conda_fmt.py

@@ -8,11 +8,16 @@
 
 import json
 import os
+import stat
 import tarfile
+import time
+from contextlib import closing
 from typing import Callable
 from zipfile import ZIP_STORED, ZipFile
 
 import zstandard
+from conda_package_streaming.package_streaming import stream_conda_component
+from conda_package_streaming.url import conda_reader_for_url
 
 from . import utils
 from .interface import AbstractBaseFormat
@@ -145,9 +150,46 @@ def get_pkg_details(in_file):
         md5, sha256 = utils.checksums(in_file, ("md5", "sha256"))
         return {"size": size, "md5": md5, "sha256": sha256}
 
-    @staticmethod
-    def list_contents(fn, verbose=False, **kw):
+    @classmethod
+    def list_contents(cls, fn, verbose=False, **kw):
         components = utils.ensure_list(kw.get("components")) or ("info", "pkg")
+        if "://" in fn:
+            return cls._list_remote_contents(fn, components=components, verbose=verbose)
+        # local resource
         if not os.path.isabs(fn):
             fn = os.path.abspath(fn)
         _list(fn, components=components, verbose=verbose)
+
+    @staticmethod
+    def _list_remote_contents(url, verbose=False, components=("info", "pkg")):
+        """
+        List contents of a remote .conda artifact (by URL). It only fetches the 'info' component
+        and uses the metadata to infer details of the 'pkg' component. Some fields like
+        modification time or permissions will be missing in verbose mode.
+        """
+        components = utils.ensure_list(components or ("info", "pkg"))
+        lines = {}
+        filename, conda = conda_reader_for_url(url)
+        with closing(conda):
+            for tar, member in stream_conda_component(filename, conda, component="info"):
+                path = member.name + ("/" if member.isdir() else "")
+                if "info" in components:
+                    line = ""
+                    if verbose:
+                        line = (
+                            f"{stat.filemode(member.mode)} "
+                            f"{member.uname or member.uid}/{member.gname or member.gid} "
+                            f"{member.size:10d} "
+                        )
+                        line += "%d-%02d-%02d %02d:%02d:%02d " % time.localtime(member.mtime)[:6]
+                    lines[path] = line + path
+                if "pkg" in components and member.name == "info/paths.json":
+                    data = json.loads(tar.extractfile(member).read().decode())
+                    assert data.get("paths_version", 1) == 1, data
+                    for path in data.get("paths", ()):
+                        line = ""
+                        if verbose:
+                            size = path["size_in_bytes"]
+                            line = f"?????????? ?/? {size:10d} ????-??-?? ??:??:?? "
+                        lines[path["_path"]] = line + path["_path"]
+        print(*[line for _, line in sorted(lines.items())], sep="\n")

src/conda_package_handling/tarball.py

@@ -94,6 +94,8 @@ def get_pkg_details(in_file):
 
     @staticmethod
     def list_contents(fn, verbose=False, **kw):
+        if "://" in fn:
+            raise ValueError("Remote .tar.bz2 artifact listing is not supported.")
         if not os.path.isabs(fn):
             fn = os.path.abspath(fn)
         streaming._list(str(fn), components=["pkg"], verbose=verbose)

tests/conftest.py

@@ -1,7 +1,11 @@
 import os
 import shutil
+import sys
+from pathlib import Path
+from textwrap import dedent
 
 import pytest
+from xprocess import ProcessStarter
 
 
 @pytest.fixture(scope="function")
@@ -30,3 +34,42 @@ def return_to_saved_path():
     request.addfinalizer(return_to_saved_path)
 
     return str(tmpdir)
+
+
+@pytest.fixture(scope="session")
+def localserver(xprocess):
+    port = 8000
+    datadir = Path(__file__).parent / "data"
+
+    class Starter(ProcessStarter):
+        pattern = "Hit Ctrl-C to quit."
+        terminate_on_interrupt = True
+        timeout = 10
+        args = [
+            sys.executable,
+            "-u",  # unbuffered
+            "-c",
+            # Adapted from conda-package-streaming/tests/server.py
+            dedent(
+                f"""
+                from bottle import route, run, static_file
+
+                @route("/<filename>", "GET")
+                def serve_file(filename):
+                    mimetype = "auto"
+                    # from https://repo.anaconda.com/ behavior:
+                    if filename.endswith(".tar.bz2"):
+                        mimetype = "application/x-tar"
+                    elif filename.endswith(".conda"):
+                        mimetype = "binary/octet-stream"
+                    return static_file(filename, root="{datadir.as_posix()}", mimetype=mimetype)
+
+                run(port={port})
+                """
+            ),
+        ]
+
+    pid, logfile = xprocess.ensure("bottle.server", Starter)
+    print("Logfile at", str(logfile))
+    yield f"http://localhost:{port}"
+    xprocess.getinfo("bottle.server").terminate()

tests/test_cli.py

@@ -90,3 +90,35 @@ def test_list(artifact, n_files, capsys):
         assert listed_files < n_files  # info folder filtered out
     else:
         assert listed_files == n_files  # no info filtering in tar.bz2
+
+
+@pytest.mark.parametrize(
+    "fn,n_files",
+    [
+        ("mock-2.0.0-py37_1000.conda", 43),
+        ("mock-2.0.0-py37_1000.tar.bz2", -1),
+    ],
+)
+def test_list_remote(capsys, localserver, fn, n_files):
+    "Integration test to ensure `cph list <URL>` works correctly."
+    url = "/".join([localserver, fn])
+    if url.endswith(".tar.bz2"):
+        # This is not supported in streaming mode
+        with pytest.raises(ValueError):
+            cli.main(["list", url])
+        return
+
+    cli.main(["list", url])
+    stdout, _ = capsys.readouterr()
+    assert n_files == sum(bool(line.strip()) for line in stdout.splitlines())
+
+    # Local list should be the same as a 'remote' list
+    cli.main(["list", str(Path(__file__).parent / "data" / fn)])
+    stdout_local, _ = capsys.readouterr()
+    assert list(map(str.strip, stdout_local.splitlines())) == list(
+        map(str.strip, stdout.splitlines())
+    )
+
+    cli.main(["list", url, "-v"])
+    stdout_verbose, _ = capsys.readouterr()
+    assert n_files == sum(bool(line.strip()) for line in stdout_verbose.splitlines())