feat(FastAPI): add /packages (post) action: 'adopt'

Signed-off-by: Kevin Morris <kevr@0cost.org>

aurweb/routers/packages.py

@@ -995,6 +995,15 @@ async def pkgbase_disown_post(request: Request, name: str,
                             status_code=HTTPStatus.SEE_OTHER)
 
 
+def pkgbase_adopt_instance(request: Request, pkgbase: models.PackageBase):
+    with db.begin():
+        pkgbase.Maintainer = request.user
+
+    conn = db.ConnectionExecutor(db.get_engine().raw_connection())
+    notif = notify.AdoptNotification(conn, request.user.ID, pkgbase.ID)
+    notif.send()
+
+
 @router.post("/pkgbase/{name}/adopt")
 @auth_required(True, redirect="/pkgbase/{name}")
 async def pkgbase_adopt_post(request: Request, name: str):
@@ -1005,8 +1014,7 @@ async def pkgbase_adopt_post(request: Request, name: str):
         # If the user has credentials, they'll adopt the package regardless
         # of maintainership. Otherwise, we'll promote the user to maintainer
         # if no maintainer currently exists.
-        with db.begin():
-            pkgbase.Maintainer = request.user
+        pkgbase_adopt_instance(request, pkgbase)
 
     return RedirectResponse(f"/pkgbase/{name}",
                             status_code=HTTPStatus.SEE_OTHER)
@@ -1151,6 +1159,38 @@ async def packages_unnotify(request: Request, package_ids: List[int] = [],
     return (True, ["The selected packages' notifications have been removed."])
 
 
+async def packages_adopt(request: Request, package_ids: List[int] = [],
+                         confirm: bool = False, **kwargs):
+    if not package_ids:
+        return (False, ["You did not select any packages to adopt."])
+
+    if not confirm:
+        return (False, ["The selected packages have not been adopted, "
+                        "check the confirmation checkbox."])
+
+    bases = set()
+    package_ids = set(package_ids)
+    packages = db.query(models.Package).filter(
+        models.Package.ID.in_(package_ids)).all()
+
+    for pkg in packages:
+        if pkg.PackageBase not in bases:
+            bases.update({pkg.PackageBase})
+
+    # Check that the user has credentials for every package they selected.
+    for pkgbase in bases:
+        has_cred = request.user.has_credential("CRED_PKGBASE_ADOPT")
+        if not (has_cred or not pkgbase.Maintainer):
+            # TODO: This error needs to be translated.
+            return (False, ["You are not allowed to adopt one of the "
+                            "packages you selected."])
+
+    # Now, really adopt the bases.
+    for pkgbase in bases:
+        pkgbase_adopt_instance(request, pkgbase)
+
+    return (True, ["The selected packages have been adopted."])
+
 # A mapping of action string -> callback functions used within the
 # `packages_post` route below. We expect any action callback to
 # return a tuple in the format: (succeeded: bool, message: List[str]).
@@ -1158,6 +1198,7 @@ PACKAGE_ACTIONS = {
     "unflag": packages_unflag,
     "notify": packages_notify,
     "unnotify": packages_unnotify,
+    "adopt": packages_adopt,
 }
 
 

po/aurweb.pot

@@ -471,6 +471,12 @@ msgid ""
 "checkbox."
 msgstr ""
 
+#: aurweb/routers/packages.py
+msgid ""
+"The selected packages have not been adopted, check the confirmation "
+"checkbox."
+msgstr ""
+
 #: html/pkgbase.php lib/pkgreqfuncs.inc.php
 msgid "Cannot find package to merge votes and comments into."
 msgstr ""
@@ -1022,6 +1028,10 @@ msgstr ""
 msgid "You must be logged in before you can adopt packages."
 msgstr ""
 
+#: aurweb/routers/package.py
+msgid "You are not allowed to adopt one of the packages you selected."
+msgstr ""
+
 #: lib/pkgbasefuncs.inc.php
 msgid "You must be logged in before you can disown packages."
 msgstr ""

test/test_packages_routes.py

@@ -2203,3 +2203,59 @@ def test_packages_post_unnotify(client: TestClient, user: User,
     errors = get_errors(resp.text)
     expected = "A package you selected does not have notifications enabled."
     assert errors[0].text.strip() == expected
+
+
+def test_packages_post_adopt(client: TestClient, user: User,
+                             package: Package):
+
+    # Try to adopt an empty list of packages.
+    cookies = {"AURSID": user.login(Request(), "testPassword")}
+    with client as request:
+        resp = request.post("/packages", data={
+            "action": "adopt"
+        }, cookies=cookies)
+    assert resp.status_code == int(HTTPStatus.BAD_REQUEST)
+    errors = get_errors(resp.text)
+    expected = "You did not select any packages to adopt."
+    assert errors[0].text.strip() == expected
+
+    # Now, let's try to adopt a package that's already maintained.
+    with client as request:
+        resp = request.post("/packages", data={
+            "action": "adopt",
+            "IDs": [package.ID],
+            "confirm": True
+        }, cookies=cookies)
+    assert resp.status_code == int(HTTPStatus.BAD_REQUEST)
+    errors = get_errors(resp.text)
+    expected = "You are not allowed to adopt one of the packages you selected."
+    assert errors[0].text.strip() == expected
+
+    # Remove the maintainer from the DB.
+    with db.begin():
+        package.PackageBase.Maintainer = None
+    assert package.PackageBase.Maintainer is None
+
+    # Now, let's try to adopt without confirming.
+    with client as request:
+        resp = request.post("/packages", data={
+            "action": "adopt",
+            "IDs": [package.ID]
+        }, cookies=cookies)
+    assert resp.status_code == int(HTTPStatus.BAD_REQUEST)
+    errors = get_errors(resp.text)
+    expected = ("The selected packages have not been adopted, "
+                "check the confirmation checkbox.")
+    assert errors[0].text.strip() == expected
+
+    # Let's do it again now that there is no maintainer.
+    with client as request:
+        resp = request.post("/packages", data={
+            "action": "adopt",
+            "IDs": [package.ID],
+            "confirm": True
+        }, cookies=cookies)
+    assert resp.status_code == int(HTTPStatus.OK)
+    successes = get_successes(resp.text)
+    expected = "The selected packages have been adopted."
+    assert successes[0].text.strip() == expected