Implement databases.privileges.transfer_ownership RPC endpoint

db/roles/operations/ownership.py

@@ -0,0 +1,5 @@
+from db.connection import exec_msar_func
+
+
+def transfer_database_ownership(new_owner_oid, conn):
+    return exec_msar_func(conn, 'transfer_database_ownership', new_owner_oid).fetchone()[0]

db/sql/00_msar.sql

@@ -1221,7 +1221,7 @@ SELECT jsonb_build_object(
   'current_role_priv', msar.list_database_privileges_for_current_role(pgd.oid),
   'current_role_owns', pg_catalog.pg_has_role(pgd.datdba, 'USAGE')
 ) FROM pg_catalog.pg_database AS pgd
-WHERE pgd.datname = current_database();
+WHERE pgd.datname = pg_catalog.current_database();
 $$ LANGUAGE SQL STABLE RETURNS NULL ON NULL INPUT;
 
 
@@ -1308,7 +1308,7 @@ SELECT string_agg(
       ' %3$I'
     ),
     val,
-    current_database(),
+    pg_catalog.current_database(),
     msar.get_role_name(rol_id)
   ),
   E';\n'
@@ -1434,6 +1434,31 @@ END;
 $$ LANGUAGE plpgsql RETURNS NULL ON NULL INPUT;
 
 
+DROP FUNCTION IF EXISTS msar.transfer_database_ownership(regrole);
+CREATE OR REPLACE FUNCTION
+msar.transfer_database_ownership(new_owner_oid regrole) RETURNS jsonb AS $$/*
+Transfers ownership of the current database to a new owner.
+
+Args:
+  new_owner_oid: The OID of the role whom we want to be the new owner of the current database.
+
+NOTE: To successfully transfer ownership of a database to a new owner the current user must:
+  - Be a Superuser/Owner of the current database.
+  - Be a `MEMBER` of the new owning role. i.e. The current role should be able to `SET ROLE`
+    to the new owning role.
+  - Have `CREATEDB` privilege.
+*/
+BEGIN
+  EXECUTE format(
+    'ALTER DATABASE %I OWNER TO %I',
+    pg_catalog.current_database(),
+    msar.get_role_name(new_owner_oid)
+  );
+  RETURN msar.get_current_database_info();
+END;
+$$ LANGUAGE plpgsql RETURNS NULL ON NULL INPUT;
+
+
 ----------------------------------------------------------------------------------------------------
 ----------------------------------------------------------------------------------------------------
 -- ALTER SCHEMA FUNCTIONS

docs/docs/api/rpc.md

@@ -81,6 +81,7 @@ To use an RPC function:
       members:
       - list_direct
       - replace_for_roles
+      - transfer_ownership
       - DBPrivileges
 
 ## Database Setup

mathesar/rpc/databases/privileges.py

@@ -5,6 +5,8 @@
 
 from db.roles.operations.select import list_db_priv
 from db.roles.operations.update import replace_database_privileges_for_roles
+from db.roles.operations.ownership import transfer_database_ownership
+from mathesar.rpc.databases.base import DatabaseInfo
 from mathesar.rpc.utils import connect
 from mathesar.rpc.exceptions.handlers import handle_rpc_exceptions
 
@@ -64,7 +66,7 @@ def replace_for_roles(
     WARNING: Any privilege included in the `direct` list for a role
     is GRANTed, and any privilege not included is REVOKEd.
 
-    Args:
+    Attributes:
         privileges: The new privilege sets for roles.
         database_id: The Django id of the database.
 
@@ -78,3 +80,29 @@ def replace_for_roles(
             conn, [DBPrivileges.from_dict(i) for i in privileges]
         )
     return [DBPrivileges.from_dict(i) for i in raw_db_priv]
+
+
+@rpc_method(name="databases.privileges.transfer_ownership")
+@http_basic_auth_login_required
+@handle_rpc_exceptions
+def transfer_ownership(*, new_owner_oid: int, database_id: int, **kwargs) -> DatabaseInfo:
+    """
+    Transfers ownership of the current database to a new owner.
+
+    Attributes:
+        new_owner_oid: The OID of the role whom we want to be the new owner of the current database.
+        database_id: The Django id of the database whose ownership is to be transferred.
+
+    Note: To successfully transfer ownership of a database to a new owner the current user must:
+        - Be a Superuser/Owner of the current database.
+        - Be a `MEMBER` of the new owning role. i.e. The current role should be able to `SET ROLE`
+          to the new owning role.
+        - Have `CREATEDB` privilege.
+
+    Returns:
+        Information about the database, and the current user privileges.
+    """
+    user = kwargs.get(REQUEST_KEY).user
+    with connect(database_id, user) as conn:
+        db_info = transfer_database_ownership(new_owner_oid, conn)
+    return DatabaseInfo.from_dict(db_info)

mathesar/tests/rpc/test_database_privileges.py

@@ -51,3 +51,46 @@ def mock_replace_privileges(
         privileges=_privileges, database_id=_database_id, request=request
     )
     assert actual_response == expect_response
+
+
+def test_transfer_db_ownership(rf, monkeypatch):
+    _username = 'alice'
+    _password = 'pass1234'
+    _database_id = 2
+    _new_owner_oid = 443123
+    request = rf.post('/api/rpc/v0/', data={})
+    request.user = User(username=_username, password=_password)
+
+    @contextmanager
+    def mock_connect(database_id, user):
+        if database_id == _database_id and user.username == _username:
+            try:
+                yield True
+            finally:
+                pass
+        else:
+            raise AssertionError('incorrect parameters passed')
+
+    def mock_tansfer_db_ownership(
+            new_owner_oid,
+            conn
+    ):
+        if new_owner_oid != _new_owner_oid:
+            raise AssertionError('incorrect parameters passed')
+        return {
+            'oid': 1988103,
+            'name': 'mathesar',
+            'owner_oid': new_owner_oid,
+            'current_role_priv': ['CONNECT', 'CREATE', 'TEMPORARY'],
+            'current_role_owns': True
+        }
+
+    monkeypatch.setattr(privileges, 'connect', mock_connect)
+    monkeypatch.setattr(
+        privileges,
+        'transfer_database_ownership',
+        mock_tansfer_db_ownership
+    )
+    privileges.transfer_ownership(
+        new_owner_oid=_new_owner_oid, database_id=_database_id, request=request
+    )

mathesar/tests/rpc/test_endpoints.py

@@ -161,6 +161,11 @@
         "databases.privileges.replace_for_roles",
         [user_is_authenticated]
     ),
+    (
+        databases.privileges.transfer_ownership,
+        "databases.privileges.transfer_ownership",
+        [user_is_authenticated]
+    ),
 
     (
         databases.setup.create_new,