Add conan ext:check-prevs command

extensions/commands/ext/cmd_check_prevs.py

@@ -0,0 +1,83 @@
+from conan.api.conan_api import ConanAPI
+from conan.api.model import MultiPackagesList, ListPattern, PackagesList, PkgReference
+from conan.api.output import ConanOutput
+
+from conan.cli import make_abs_path
+from conan.cli.command import conan_command, OnceArgument
+from conan.cli.commands.list import print_list_text, print_list_json, print_list_compact
+from conan.cli.formatters.list import list_packages_html
+
+from conan.errors import ConanException
+
+
+@conan_command(group="Extension", formatters={"text": print_list_text,
+                                              "json": print_list_json,
+                                              "html": list_packages_html,
+                                              "compact": print_list_compact})
+def check_prevs(conan_api: ConanAPI, parser, *args):
+    """
+    Ensure that the selected references only contains 1 package revision in the given remotes
+    """
+    parser.add_argument('pattern', nargs="?",
+                        help="A pattern in the form 'pkg/version#revision:package_id#revision', "
+                             "e.g: \"zlib/1.2.13:*\" means all binaries for zlib/1.2.13. "
+                             "If revision is not specified, it is assumed latest one.")
+    parser.add_argument("-l", "--list", help="Package list file")
+    parser.add_argument('-p', '--package-query', default=None, action=OnceArgument,
+                        help="Only upload packages matching a specific query. e.g: os=Windows AND "
+                             "(arch=x86 OR compiler=gcc)")
+    parser.add_argument("-r", "--remote", action="append", default=None,
+                        help='Look in the specified remote or remotes server')
+
+    args = parser.parse_args(*args)
+    remotes = conan_api.remotes.list(args.remote)
+
+    if args.pattern is None and args.list is None:
+        raise ConanException("Missing pattern or package list file")
+    if args.pattern and args.list:
+        raise ConanException("Cannot define both the pattern and the package list file")
+    if args.package_query and args.list:
+        raise ConanException("Cannot define package-query and the package list file")
+
+    result = MultiPackagesList()
+
+    for remote in remotes:
+        if args.list:
+            listfile = make_abs_path(args.list)
+            multi_package_list = MultiPackagesList.load(listfile)
+            if remote.name not in multi_package_list.lists:
+                ConanOutput().warning(f"No packages for remote '{remote.name}' were found "
+                                      "in the package list, skipping it.")
+                continue
+            pkglist = multi_package_list[remote.name]
+        else:
+            ref_pattern = ListPattern(args.pattern, rrev=None, prev="*")
+            if not ref_pattern.package_id:
+                raise ConanException("The pattern must include a package_id, e.g: "
+                                     "\"zlib/1.2.13:*\" means all binaries for zlib/1.2.13")
+            pkglist = conan_api.list.select(ref_pattern, args.package_query, remote)
+        remote_pkglist = PackagesList()
+        # Can't iterate packages because we might have been not given a package revision
+        for ref, _ in pkglist.items():
+            recipe_dict = pkglist.recipe_dict(ref)
+            for package_id, pkg_info in recipe_dict.get("packages", {}).items():
+                prevs = pkg_info.get("revisions", {})
+                revisions = []
+                if len(prevs) > 1:
+                    # No need to ask the server (again if coming from the select endpoint)
+                    # we already know there are multiple revisions for this package_id
+                    for prev in prevs:
+                        revisions.append(PkgReference(ref, package_id, prev))
+                else:
+                    revisions = conan_api.list.package_revisions(PkgReference(ref, package_id),
+                                                                 remote)
+                if len(revisions) > 1:
+                    remote_pkglist.add_ref(ref)
+                    for pref in revisions:
+                        remote_pkglist.add_pref(pref)
+                    result.add(remote.name, remote_pkglist)
+
+    return {
+        "conan_error": "Multiple package revisions found" if result.lists else None,
+        "results": result.serialize(),
+    }

tests/test_ext_commands.py

@@ -0,0 +1,83 @@
+import os
+import textwrap
+
+import pytest
+from conan.test.utils.tools import TestClient
+
+
+@pytest.fixture(scope="package")
+def client():
+    tc = TestClient(light=True,
+                    default_server_user=True,
+                    custom_commands_folder=os.path.join(os.path.dirname(os.path.realpath(__file__)),
+                                                        os.path.pardir, "extensions", "commands"))
+    conanfile = textwrap.dedent("""
+        import os
+        import time
+        from conan import ConanFile
+        from conan.tools.files import save
+
+        class Pkg(ConanFile):
+            version = "1.0"
+            def package(self):
+                save(self, os.path.join(self.package_folder, "file.txt"), str(time.time()))
+        """)
+    tc.save({"conanfile.py": conanfile})
+
+    tc.run("create --name=hello")
+    tc.run("create --name=hello")
+    tc.run("create --name=greetings")
+    tc.run("create --name=greetings")
+
+    tc.run("create --name=bye")
+    tc.run("upload '*:*#*' -c -r default")
+    return tc
+
+
+@pytest.mark.parametrize("remote", [True, False])
+@pytest.mark.parametrize("pattern", ["*:*", "*:*#*"])
+def test_check_prev_patterns_find(client, pattern, remote):
+    remote = "-r default" if remote else ""
+    client.run(f"ext:check-prevs {pattern} {remote}", assert_error=True)
+    assert "ERROR: Multiple package revisions found" in client.out
+    assert "bye/1.0" not in client.out
+    assert "hello/1.0" in client.out
+    assert "greetings/1.0" in client.out
+
+
+@pytest.mark.parametrize("pattern", ["bye/1.0:*", "bye/1.0:*#*"])
+def test_check_prev_patterns_no_find(client, pattern):
+    client.run(f"ext:check-prevs {pattern} -r default")
+    assert "ERROR: Multiple package revisions found" not in client.out
+    assert "bye/1.0" not in client.out
+    assert "hello/1.0" not in client.out
+    assert "greetings/1.0" not in client.out
+
+
+@pytest.mark.parametrize("pattern", ["*:*", "*:*#*"])
+def test_check_prev_pkglist_find(client, pattern):
+    client.run(f"list {pattern} -r default -f=json", redirect_stdout="list.json")
+    client.run(f"ext:check-prevs -l list.json", assert_error=True)
+    assert "ERROR: Multiple package revisions found" in client.out
+    assert "bye/1.0" not in client.out
+    assert "hello/1.0" in client.out
+    assert "greetings/1.0" in client.out
+
+
+@pytest.mark.parametrize("pattern", ["bye/1.0:*", "bye/1.0:*#*"])
+def test_check_prev_pkglist_no_find(client, pattern):
+    client.run(f"list {pattern} -r default -f=json", redirect_stdout="list.json")
+    client.run(f"ext:check-prevs -l list.json")
+    assert "ERROR: Multiple package revisions found" not in client.out
+    assert "bye/1.0" not in client.out
+    assert "hello/1.0" not in client.out
+    assert "greetings/1.0" not in client.out
+
+
+def test_check_prev_errors(client):
+    client.run("ext:check-prevs * -l list.json", assert_error=True)
+    assert 'ERROR: Cannot define both the pattern and the package list file' in client.out
+
+    client.run(f"ext:check-prevs hello/1.0 -r default", assert_error=True)
+    assert "ERROR: Multiple package revisions found" not in client.out
+    assert "ERROR: The pattern must include a package_id" in client.out