Add a PHP Composer (Packagist) finder

Adds the phpcompser finder type, which finds PHP packages at
packagist.org.

soufi/cli.py

@@ -201,6 +201,17 @@ def phppecl(cls, name, version, timeout=None):
         )
         return cls.find(pecl_finder)
 
+    @classmethod
+    def phpcomposer(cls, name, version, timeout=None):
+        composer_finder = finder.factory(
+            "phpcomposer",
+            name=name,
+            version=version,
+            s_type=finder.SourceType.phpcomposer,
+            timeout=timeout,
+        )
+        return cls.find(composer_finder)
+
 
 def make_archive_from_discovery_source(disc_src, fname):
     try:

soufi/finder.py

@@ -32,6 +32,7 @@ class SourceType(enum.Enum):
     nuget = "nuget"
     crate = "crate"
     phppecl = "phppecl"
+    phpcomposer = "phpcomposer"
 
 
 @enum.unique

soufi/finders/php_composer.py

@@ -0,0 +1,83 @@
+# Copyright (c) 2024 Cisco Systems, Inc. and its affiliates
+# All rights reserved.
+
+
+import requests
+
+from soufi import exceptions, finder
+
+DEFAULT_INDEX = "https://repo.packagist.org/"
+
+
+class PHPComposer(finder.SourceFinder):
+    """Find PHP Composer packages at the Packagist repo.
+
+    Looks up the package in the Packagist repo and returns the URL for the
+    package.
+    """
+
+    distro = finder.SourceType.phpcomposer.value
+
+    def _find(self):
+        source_url, extension = self.get_source_url()
+        return PHPComposerDiscoveredSource(
+            [source_url],
+            timeout=self.timeout,
+            archive_extension=f".{extension}",
+        )
+
+    def get_source_url(self):
+        """Examine the repo to find the source URL for the package.
+
+        Packagist has two APIs (see https://packagist.org/apidoc). The main one
+        doesn't seem to allow queries for specific packages (nor at specific
+        versions), so we can use repo.packagist.org optimistically by querying
+        a known URL pattern at repo.packagist.org/p2/{package}.json. If the
+        package exists, the metadata that is returned contains a list of all
+        versions of the package that will need to be iterated to get the
+        download URL.
+
+        Note that the name of Composer packages usually takes the form of
+        vendor/package, so the name of the package is the concatenation of
+        these two parts.
+
+        :return: A tuple of (URL, type) where the type is the archive_extension
+            to use.
+        """
+        url = f"{DEFAULT_INDEX}p2/{self.name}.json"
+        resp = requests.get(url, timeout=self.timeout)
+        if resp.status_code != requests.codes.ok:
+            raise exceptions.SourceNotFound
+
+        try:
+            versions = resp.json()["packages"][self.name]
+            for version in versions:
+                if self.version in (
+                    version["version"],
+                    version['version_normalized'],
+                ):
+                    return version["dist"]["url"], version["dist"]["type"]
+        except Exception:
+            raise exceptions.DownloadError(
+                "Malformed JSON response from {url}"
+            )
+
+        raise exceptions.SourceNotFound
+
+
+class PHPComposerDiscoveredSource(finder.DiscoveredSource):
+    """A discovered PHP Composer package."""
+
+    make_archive = finder.DiscoveredSource.remote_url_is_archive
+
+    def __init__(self, *args, archive_extension, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.archive_extension = archive_extension
+
+    def populate_archive(self, *args, **kwargs):  # pragma: no cover
+        # Required by the base class but PECL archives are already tarballs so
+        # nothing to do.
+        pass
+
+    def __repr__(self):
+        return f"{self.urls[0]}: {self.archive_extension}"

soufi/finders/php_pecl.py

@@ -40,7 +40,7 @@ def get_source_url(self):
 
 
 class PHPPECLDiscoveredSource(finder.DiscoveredSource):
-    """A discovered Rust Crate package."""
+    """A discovered PHP PECL package."""
 
     make_archive = finder.DiscoveredSource.remote_url_is_archive
     archive_extension = ".tgz"

soufi/tests/finders/test_phpcomposer_finder.py

@@ -0,0 +1,105 @@
+# Copyright (c) 2024 Cisco Systems, Inc. and its affiliates
+# All rights reserved.
+
+import json
+from pathlib import Path
+from unittest import mock
+
+import requests
+
+from soufi import exceptions, testing
+from soufi.finder import SourceType
+from soufi.finders import php_composer
+from soufi.testing import base
+
+
+class TestPHPPECLFinder(base.TestCase):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.testing = Path(testing.__path__[0]) / 'data' / 'phpcomposer'
+
+    def make_finder(self, name=None, version=None):
+        if name is None:
+            name = self.factory.make_string("name")
+        if version is None:
+            version = self.factory.make_string("version")
+        kwargs = dict(
+            name=name, version=version, s_type=SourceType.phpcomposer
+        )
+        return php_composer.PHPComposer(**kwargs)
+
+    def test_get_source_url(self):
+        # The test data uses this package.
+        finder = self.make_finder("monolog/monolog", "2.3.5")
+        expected_url = self.factory.make_url()
+        expected_type = self.factory.random_choice(["zip", "tgz", "tar.gz"])
+        with open(self.testing / "monolog.json", "rb") as fp:
+            data = json.load(fp)
+        # Patch the real URL with our random one to make sure it's correctly
+        # picked up later.
+        for version in data["packages"]["monolog/monolog"]:
+            if version["version"] == "2.3.5":
+                version["dist"]["url"] = expected_url
+                version["dist"]["type"] = expected_type
+                break
+        get = self.patch_get_with_response(requests.codes.ok, json=data)
+
+        found_url, found_type = finder.get_source_url()
+        self.assertEqual(expected_url, found_url)
+        self.assertEqual(expected_type, found_type)
+        call = mock.call(
+            f"{php_composer.DEFAULT_INDEX}p2/{finder.name}.json",
+            timeout=30,
+        )
+        self.assertIn(call, get.call_args_list)
+
+    def test_get_source_url_source_not_found(self):
+        finder = self.make_finder()
+        self.patch_get_with_response(requests.codes.not_found)
+        self.assertRaises(exceptions.SourceNotFound, finder.get_source_url)
+
+    def test_get_source_version_not_found(self):
+        finder = self.make_finder("monolog/monolog")
+        with open(self.testing / "monolog.json", "rb") as fp:
+            data = json.load(fp)
+        self.patch_get_with_response(requests.codes.ok, json=data)
+        self.assertRaises(exceptions.SourceNotFound, finder.get_source_url)
+
+    def test_get_source_url_malformed_json(self):
+        finder = self.make_finder()
+        self.patch_get_with_response(requests.codes.ok, data="not json")
+        self.assertRaises(exceptions.DownloadError, finder.get_source_url)
+
+    def test_find(self):
+        finder = self.make_finder()
+        url = self.factory.make_url()
+        atype = self.factory.random_choice(["zip", "tgz", "tar.gz"])
+        self.patch(finder, "get_source_url").return_value = url, atype
+
+        disc_source = finder.find()
+        self.assertIsInstance(
+            disc_source, php_composer.PHPComposerDiscoveredSource
+        )
+        self.assertEqual([url], disc_source.urls)
+        self.assertEqual(f".{atype}", disc_source.archive_extension)
+
+
+class TestPHPPECLDiscoveredSource(base.TestCase):
+    def make_discovered_source(self, url=None, atype=None):
+        if url is None:
+            url = self.factory.make_url()
+        if atype is None:
+            atype = self.factory.random_choice([".zip", ".tgz", ".tar.gz"])
+        return php_composer.PHPComposerDiscoveredSource(
+            [url], archive_extension=atype
+        )
+
+    def test_repr(self):
+        url = self.factory.make_url()
+        atype = self.factory.random_choice([".zip", ".tgz", ".tar.gz"])
+        ds = self.make_discovered_source(url, atype)
+        self.assertEqual(f"{url}: {atype}", repr(ds))
+
+    def test_make_archive(self):
+        ds = self.make_discovered_source()
+        self.assertEqual(ds.make_archive, ds.remote_url_is_archive)

soufi/tests/test_factory.py

@@ -36,6 +36,7 @@ def test_supported_types(self):
             "java",
             "npm",
             "photon",
+            "phpcomposer",
             "phppecl",
             "python",
             "rhel",