Merge lp:~cjwatson/launchpad/snap-find-by-store-name into lp:launchpad

[Colin Watson]

Add SnapSet.findByStoreName.

This will be useful for the next iteration of build.snapcraft.io, and is handy for ad-hoc operational queries as well.

[William Grant (wgrant)]

review: Approve (code db)

=== added file 'database/schema/patch-2210-05-0.sql'
--- database/schema/patch-2210-05-0.sql	1970-01-01 00:00:00 +0000
+++ database/schema/patch-2210-05-0.sql	2019-08-22 10:39:42 +0000
@@ -0,0 +1,9 @@
+-- Copyright 2019 Canonical Ltd.  This software is licensed under the
+-- GNU Affero General Public License version 3 (see the file LICENSE).
+
+SET client_min_messages=ERROR;
+
+CREATE INDEX snap__git_repository_url__idx ON Snap (git_repository_url);
+CREATE INDEX snap__store_name__idx ON Snap (store_name);
+
+INSERT INTO LaunchpadDatabaseRevision VALUES (2210, 05, 0);

=== modified file 'lib/lp/snappy/interfaces/snap.py'
--- lib/lp/snappy/interfaces/snap.py	2019-06-21 11:30:26 +0000
+++ lib/lp/snappy/interfaces/snap.py	2019-08-22 10:39:42 +0000
@@ -1005,6 +1005,23 @@
             this user; otherwise, only return publicly-visible packages.
         """
 
+    @operation_parameters(
+        store_name=TextLine(
+            title=_("The registered store package name 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 findByStoreName(store_name, owner=None, visible_by_user=None):
+        """Return all snap packages with the given store package name.
+
+        :param store_name: A registered store package name.
+        :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.
+        """
+
     def preloadDataForSnaps(snaps, user):
         """Load the data related to a list of snap packages."""
 

=== modified file 'lib/lp/snappy/model/snap.py'
--- lib/lp/snappy/model/snap.py	2019-08-21 10:36:27 +0000
+++ lib/lp/snappy/model/snap.py	2019-08-22 10:39:42 +0000
@@ -1244,7 +1244,7 @@
             snaps.order_by(Desc(Snap.date_last_modified))
         return snaps
 
-    def _findByURLVisibilityClause(self, visible_by_user):
+    def _findSnapVisibilityClause(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 the numbers involved should be very small.
@@ -1266,7 +1266,7 @@
         clauses = [Snap.git_repository_url == url]
         if owner is not None:
             clauses.append(Snap.owner == owner)
-        clauses.append(self._findByURLVisibilityClause(visible_by_user))
+        clauses.append(self._findSnapVisibilityClause(visible_by_user))
         return IStore(Snap).find(Snap, *clauses)
 
     def findByURLPrefix(self, url_prefix, owner=None, visible_by_user=None):
@@ -1283,7 +1283,15 @@
         clauses = [Or(*prefix_clauses)]
         if owner is not None:
             clauses.append(Snap.owner == owner)
-        clauses.append(self._findByURLVisibilityClause(visible_by_user))
+        clauses.append(self._findSnapVisibilityClause(visible_by_user))
+        return IStore(Snap).find(Snap, *clauses)
+
+    def findByStoreName(self, store_name, owner=None, visible_by_user=None):
+        """See `ISnapSet`."""
+        clauses = [Snap.store_name == store_name]
+        if owner is not None:
+            clauses.append(Snap.owner == owner)
+        clauses.append(self._findSnapVisibilityClause(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	2019-08-21 10:36:27 +0000
+++ lib/lp/snappy/tests/test_snap.py	2019-08-22 10:39:42 +0000
@@ -1677,6 +1677,34 @@
             [snaps[0], snaps[2], snaps[4], snaps[6]],
             getUtility(ISnapSet).findByURLPrefixes(prefixes, owner=owners[0]))
 
+    def test_findByStoreName(self):
+        # ISnapSet.findByStoreName returns visible Snaps with the given
+        # store name.
+        store_names = ["foo", "bar"]
+        owners = [self.factory.makePerson() for i in range(2)]
+        snaps = []
+        for store_name in store_names:
+            for owner in owners:
+                for private in (False, True):
+                    snaps.append(self.factory.makeSnap(
+                        registrant=owner, owner=owner, private=private,
+                        store_name=store_name))
+        snaps.append(self.factory.makeSnap())
+        self.assertContentEqual(
+            [snaps[0], snaps[2]],
+            getUtility(ISnapSet).findByStoreName(store_names[0]))
+        with person_logged_in(owners[0]):
+            self.assertContentEqual(
+                snaps[:2],
+                getUtility(ISnapSet).findByStoreName(
+                    store_names[0], owner=owners[0],
+                    visible_by_user=owners[0]))
+            self.assertContentEqual(
+                [snaps[2]],
+                getUtility(ISnapSet).findByStoreName(
+                    store_names[0], owner=owners[1],
+                    visible_by_user=owners[0]))
+
     def test_getSnapcraftYaml_bzr_snap_snapcraft_yaml(self):
         def getInventory(unique_name, dirname, *args, **kwargs):
             if dirname == "snap":
@@ -2732,6 +2760,7 @@
         commercial_admin = (
             getUtility(ILaunchpadCelebrities).commercial_admin.teamowner)
         logout()
+
         # Anonymous requests can only see public snaps.
         anon_webservice = LaunchpadWebServiceCaller("test", "")
         response = anon_webservice.named_get(
@@ -2741,6 +2770,7 @@
         self.assertContentEqual(
             [ws_snaps[0]],
             [entry["self_link"] for entry in response.jsonBody()["entries"]])
+
         # persons[0] can see their own private snap as well, but not those
         # for other people.
         webservice = webservice_for_person(
@@ -2759,6 +2789,7 @@
         self.assertContentEqual(
             [ws_snaps[2]],
             [entry["self_link"] for entry in response.jsonBody()["entries"]])
+
         # Admins can see all snaps.
         commercial_admin_webservice = webservice_for_person(
             commercial_admin, permission=OAuthPermission.READ_PRIVATE)
@@ -2797,6 +2828,7 @@
         commercial_admin = (
             getUtility(ILaunchpadCelebrities).commercial_admin.teamowner)
         logout()
+
         # Anonymous requests can only see public snaps.
         anon_webservice = LaunchpadWebServiceCaller("test", "")
         response = anon_webservice.named_get(
@@ -2812,6 +2844,7 @@
         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(
@@ -2829,6 +2862,7 @@
         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)
@@ -2873,6 +2907,7 @@
             getUtility(ILaunchpadCelebrities).commercial_admin.teamowner)
         logout()
         prefix = "https://git.example.org/foo/"
+
         # Anonymous requests can only see public snaps.
         anon_webservice = LaunchpadWebServiceCaller("test", "")
         response = anon_webservice.named_get(
@@ -2889,6 +2924,7 @@
         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(
@@ -2907,6 +2943,7 @@
         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)
@@ -2955,6 +2992,7 @@
         logout()
         prefixes = [
             "https://git.example.org/foo/", "https://git.example.org/bar/"]
+
         # Anonymous requests can only see public snaps.
         anon_webservice = LaunchpadWebServiceCaller("test", "")
         response = anon_webservice.named_get(
@@ -2971,6 +3009,7 @@
         self.assertContentEqual(
             [ws_snaps[i] for i in (0, 4, 8, 12)],
             [entry["self_link"] for entry in response.jsonBody()["entries"]])
+
         # persons[0] can see all public snaps with any of these URL
         # prefixes, as well as their own matching private snaps.
         webservice = webservice_for_person(
@@ -2989,6 +3028,7 @@
         self.assertContentEqual(
             [ws_snaps[i] for i in (0, 1, 4, 5, 8, 9, 12, 13)],
             [entry["self_link"] for entry in response.jsonBody()["entries"]])
+
         # Admins can see all snaps with any of these URL prefixes.
         commercial_admin_webservice = webservice_for_person(
             commercial_admin, permission=OAuthPermission.READ_PRIVATE)
@@ -3007,6 +3047,81 @@
             [ws_snaps[i] for i in (0, 1, 4, 5, 8, 9, 12, 13)],
             [entry["self_link"] for entry in response.jsonBody()["entries"]])
 
+    def test_findByStoreName(self):
+        # lp.snaps.findByStoreName returns visible Snaps with the given
+        # store name.
+        persons = [self.factory.makePerson(), self.factory.makePerson()]
+        store_names = ["foo", "bar"]
+        snaps = []
+        for store_name in store_names:
+            for person in persons:
+                for private in (False, True):
+                    snaps.append(self.factory.makeSnap(
+                        registrant=person, owner=person, private=private,
+                        store_name=store_name))
+        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()
+
+        # Anonymous requests can only see public snaps.
+        anon_webservice = LaunchpadWebServiceCaller("test", "")
+        response = anon_webservice.named_get(
+            "/+snaps", "findByStoreName", store_name=store_names[0],
+            api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            [ws_snaps[0], ws_snaps[2]],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        response = anon_webservice.named_get(
+            "/+snaps", "findByStoreName", store_name=store_names[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 store name, as well
+        # as their own private snap.
+        webservice = webservice_for_person(
+            persons[0], permission=OAuthPermission.READ_PRIVATE)
+        response = webservice.named_get(
+            "/+snaps", "findByStoreName", store_name=store_names[0],
+            api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            ws_snaps[:3],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        response = webservice.named_get(
+            "/+snaps", "findByStoreName", store_name=store_names[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 store name.
+        commercial_admin_webservice = webservice_for_person(
+            commercial_admin, permission=OAuthPermission.READ_PRIVATE)
+        response = commercial_admin_webservice.named_get(
+            "/+snaps", "findByStoreName", store_name=store_names[0],
+            api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            ws_snaps[:4],
+            [entry["self_link"] for entry in response.jsonBody()["entries"]])
+        response = commercial_admin_webservice.named_get(
+            "/+snaps", "findByStoreName", store_name=store_names[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 setProcessors(self, user, snap, names):
         ws = webservice_for_person(
             user, permission=OAuthPermission.WRITE_PUBLIC)