Merge lp:~cjwatson/launchpad/snap-extend-find-by-url into lp:launchpad

[Colin Watson]

Allow limiting SnapSet.findByURL results by owner; add SnapSet.findByURLPrefix.

build.snapcraft.io is going to need to show a dashboard of a user's repositories, including which of them are already enabled. We'll of course need to do lots of caching, but we'll also need a bulk interface to let us discover which snaps exist in Launchpad for a set of GitHub repositories. I think it's best to just do this by URL prefix.

[William Grant (wgrant)]

review: Approve (code)

=== modified file 'lib/lp/snappy/interfaces/snap.py'
--- lib/lp/snappy/interfaces/snap.py	2017-02-01 06:31:30 +0000
+++ lib/lp/snappy/interfaces/snap.py	2017-02-03 14:33:36 +0000
@@ -655,12 +655,14 @@
         :raises BadSnapSearchContext: if the context is not understood.
         """
 
-    @operation_parameters(url=TextLine(title=_("The URL to search for.")))
+    @operation_parameters(
+        url=TextLine(title=_("The URL to search for.")),
+        owner=Reference(IPerson, title=_("Owner"), required=False))
     @call_with(visible_by_user=REQUEST_USER)
     @operation_returns_collection_of(ISnap)
     @export_read_operation()
     @operation_for_version("devel")
-    def findByURL(url, visible_by_user=None):
+    def findByURL(url, owner=None, visible_by_user=None):
         """Return all snap packages that build from the given URL.
 
         This currently only works for packages that build directly from a
@@ -668,6 +670,27 @@
         hosted in Launchpad.
 
         :param url: A URL.
+        :param owner: Only return packages owned by this user.
+        :param visible_by_user: If not None, only return packages visible by
+            this user; otherwise, only return publicly-visible packages.
+        """
+
+    @operation_parameters(
+        url_prefix=TextLine(title=_("The URL prefix to search for.")),
+        owner=Reference(IPerson, title=_("Owner"), required=False))
+    @call_with(visible_by_user=REQUEST_USER)
+    @operation_returns_collection_of(ISnap)
+    @export_read_operation()
+    @operation_for_version("devel")
+    def findByURLPrefix(url_prefix, owner=None, visible_by_user=None):
+        """Return all snap packages that build from a URL with this prefix.
+
+        This currently only works for packages that build directly from a
+        URL, rather than being linked to a Bazaar branch or Git repository
+        hosted in Launchpad.
+
+        :param url_prefix: A URL prefix.
+        :param owner: Only return packages owned by this user.
         :param visible_by_user: If not None, only return packages visible by
             this user; otherwise, only return publicly-visible packages.
         """

=== modified file 'lib/lp/snappy/model/snap.py'
--- lib/lp/snappy/model/snap.py	2017-01-04 22:59:04 +0000
+++ lib/lp/snappy/model/snap.py	2017-02-03 14:33:36 +0000
@@ -799,26 +799,37 @@
             snaps.order_by(Desc(Snap.date_last_modified))
         return snaps
 
-    def findByURL(self, url, visible_by_user=None):
-        """See `ISnapSet`."""
-        clauses = [Snap.git_repository_url == url]
+    def _findByURLVisibilityClause(self, visible_by_user):
         # XXX cjwatson 2016-11-25: This is in principle a poor query, but we
         # don't yet have the access grant infrastructure to do better, and
-        # in any case since we're querying for a single URL the numbers
-        # involved should be very small.
+        # in any case the numbers involved should be very small.
         if visible_by_user is None:
-            visibility_clause = Snap.private == False
+            return Snap.private == False
         else:
             roles = IPersonRoles(visible_by_user)
             if roles.in_admin or roles.in_commercial_admin:
-                visibility_clause = True
+                return True
             else:
-                visibility_clause = Or(
+                return Or(
                     Snap.private == False,
                     Snap.owner_id.is_in(Select(
                         TeamParticipation.teamID,
                         TeamParticipation.person == visible_by_user)))
-        clauses.append(visibility_clause)
+
+    def findByURL(self, url, owner=None, visible_by_user=None):
+        """See `ISnapSet`."""
+        clauses = [Snap.git_repository_url == url]
+        if owner is not None:
+            clauses.append(Snap.owner == owner)
+        clauses.append(self._findByURLVisibilityClause(visible_by_user))
+        return IStore(Snap).find(Snap, *clauses)
+
+    def findByURLPrefix(self, url_prefix, owner=None, visible_by_user=None):
+        """See `ISnapSet`."""
+        clauses = [Snap.git_repository_url.startswith(url_prefix)]
+        if owner is not None:
+            clauses.append(Snap.owner == owner)
+        clauses.append(self._findByURLVisibilityClause(visible_by_user))
         return IStore(Snap).find(Snap, *clauses)
 
     def preloadDataForSnaps(self, snaps, user=None):

=== modified file 'lib/lp/snappy/tests/test_snap.py'
--- lib/lp/snappy/tests/test_snap.py	2017-01-04 20:52:12 +0000
+++ lib/lp/snappy/tests/test_snap.py	2017-02-03 14:33:36 +0000
@@ -832,16 +832,48 @@
     def test_findByURL(self):
         # ISnapSet.findByURL returns visible Snaps with the given URL.
         urls = [u"https://git.example.org/foo", u"https://git.example.org/bar"]
-        snaps = []
-        for url in urls:
-            snaps.append(self.factory.makeSnap(
-                git_ref=self.factory.makeGitRefRemote(repository_url=url)))
-        snaps.append(
-            self.factory.makeSnap(branch=self.factory.makeAnyBranch()))
-        snaps.append(
-            self.factory.makeSnap(git_ref=self.factory.makeGitRefs()[0]))
-        self.assertContentEqual(
-            [snaps[0]], getUtility(ISnapSet).findByURL(urls[0]))
+        owners = [self.factory.makePerson() for i in range(2)]
+        snaps = []
+        for url in urls:
+            for owner in owners:
+                snaps.append(self.factory.makeSnap(
+                    registrant=owner, owner=owner,
+                    git_ref=self.factory.makeGitRefRemote(repository_url=url)))
+        snaps.append(
+            self.factory.makeSnap(branch=self.factory.makeAnyBranch()))
+        snaps.append(
+            self.factory.makeSnap(git_ref=self.factory.makeGitRefs()[0]))
+        self.assertContentEqual(
+            snaps[:2], getUtility(ISnapSet).findByURL(urls[0]))
+        self.assertContentEqual(
+            [snaps[0]],
+            getUtility(ISnapSet).findByURL(urls[0], owner=owners[0]))
+
+    def test_findByURLPrefix(self):
+        # ISnapSet.findByURLPrefix returns visible Snaps with the given URL
+        # prefix.
+        urls = [
+            u"https://git.example.org/foo/a",
+            u"https://git.example.org/foo/b",
+            u"https://git.example.org/bar",
+            ]
+        owners = [self.factory.makePerson() for i in range(2)]
+        snaps = []
+        for url in urls:
+            for owner in owners:
+                snaps.append(self.factory.makeSnap(
+                    registrant=owner, owner=owner,
+                    git_ref=self.factory.makeGitRefRemote(repository_url=url)))
+        snaps.append(
+            self.factory.makeSnap(branch=self.factory.makeAnyBranch()))
+        snaps.append(
+            self.factory.makeSnap(git_ref=self.factory.makeGitRefs()[0]))
+        prefix = u"https://git.example.org/foo/"
+        self.assertContentEqual(
+            snaps[:4], getUtility(ISnapSet).findByURLPrefix(prefix))
+        self.assertContentEqual(
+            [snaps[0], snaps[2]],
+            getUtility(ISnapSet).findByURLPrefix(prefix, owner=owners[0]))
 
     def test__findStaleSnaps(self):
         # Stale; not built automatically.
@@ -1325,8 +1357,10 @@
                 for private in (False, True):
                     ref = self.factory.makeGitRefRemote(repository_url=url)
                     snaps.append(self.factory.makeSnap(
-                        registrant=person, git_ref=ref, private=private))
+                        registrant=person, owner=person, git_ref=ref,
+                        private=private))
         with admin_logged_in():
+            person_urls = [api_url(person) for person in persons]
             ws_snaps = [
                 self.webservice.getAbsoluteUrl(api_url(snap))
                 for snap in snaps]
@@ -1341,6 +1375,13 @@
         self.assertContentEqual(
             [ws_snaps[0], ws_snaps[2]],
             [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        response = anon_webservice.named_get(
+            "/+snaps", "findByURL", url=urls[0], owner=person_urls[0],
+            api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            [ws_snaps[0]],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
         # persons[0] can see both public snaps with this URL, as well as
         # their own private snap.
         webservice = webservice_for_person(
@@ -1351,6 +1392,13 @@
         self.assertContentEqual(
             ws_snaps[:3],
             [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        response = webservice.named_get(
+            "/+snaps", "findByURL", url=urls[0], owner=person_urls[0],
+            api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            ws_snaps[:2],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
         # Admins can see all snaps with this URL.
         commercial_admin_webservice = webservice_for_person(
             commercial_admin, permission=OAuthPermission.READ_PRIVATE)
@@ -1360,6 +1408,92 @@
         self.assertContentEqual(
             ws_snaps[:4],
             [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        response = commercial_admin_webservice.named_get(
+            "/+snaps", "findByURL", url=urls[0], owner=person_urls[0],
+            api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            ws_snaps[:2],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
+
+    def test_findByURLPrefix(self):
+        # lp.snaps.findByURLPrefix returns visible Snaps with the given URL
+        # prefix.
+        self.pushConfig("launchpad", default_batch_size=10)
+        persons = [self.factory.makePerson(), self.factory.makePerson()]
+        urls = [
+            u"https://git.example.org/foo/a",
+            u"https://git.example.org/foo/b",
+            u"https://git.example.org/bar",
+            ]
+        snaps = []
+        for url in urls:
+            for person in persons:
+                for private in (False, True):
+                    ref = self.factory.makeGitRefRemote(repository_url=url)
+                    snaps.append(self.factory.makeSnap(
+                        registrant=person, owner=person, git_ref=ref,
+                        private=private))
+        with admin_logged_in():
+            person_urls = [api_url(person) for person in persons]
+            ws_snaps = [
+                self.webservice.getAbsoluteUrl(api_url(snap))
+                for snap in snaps]
+        commercial_admin = (
+            getUtility(ILaunchpadCelebrities).commercial_admin.teamowner)
+        logout()
+        prefix = u"https://git.example.org/foo/"
+        # Anonymous requests can only see public snaps.
+        anon_webservice = LaunchpadWebServiceCaller("test", "")
+        response = anon_webservice.named_get(
+            "/+snaps", "findByURLPrefix", url_prefix=prefix,
+            api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            [ws_snaps[i] for i in (0, 2, 4, 6)],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        response = anon_webservice.named_get(
+            "/+snaps", "findByURLPrefix", url_prefix=prefix,
+            owner=person_urls[0], api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            [ws_snaps[i] for i in (0, 4)],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        # persons[0] can see all public snaps with this URL prefix, as well
+        # as their own matching private snaps.
+        webservice = webservice_for_person(
+            persons[0], permission=OAuthPermission.READ_PRIVATE)
+        response = webservice.named_get(
+            "/+snaps", "findByURLPrefix", url_prefix=prefix,
+            api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            [ws_snaps[i] for i in (0, 1, 2, 4, 5, 6)],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        response = webservice.named_get(
+            "/+snaps", "findByURLPrefix", url_prefix=prefix,
+            owner=person_urls[0], api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            [ws_snaps[i] for i in (0, 1, 4, 5)],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        # Admins can see all snaps with this URL prefix.
+        commercial_admin_webservice = webservice_for_person(
+            commercial_admin, permission=OAuthPermission.READ_PRIVATE)
+        response = commercial_admin_webservice.named_get(
+            "/+snaps", "findByURLPrefix", url_prefix=prefix,
+            api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            ws_snaps[:8],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        response = commercial_admin_webservice.named_get(
+            "/+snaps", "findByURLPrefix", url_prefix=prefix,
+            owner=person_urls[0], api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            [ws_snaps[i] for i in (0, 1, 4, 5)],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
 
     def setProcessors(self, user, snap, names):
         ws = webservice_for_person(