Materialize/misc/python/materialize/checks/all_checks/roles.py

# Copyright Materialize, Inc. and contributors. All rights reserved.
#
# Use of this software is governed by the Business Source License
# included in the LICENSE file at the root of this repository.
#
# As of the Change Date specified in that file, in accordance with
# the Business Source License, use of this software will be governed
# by the Apache License, Version 2.0.
from textwrap import dedent

from materialize.checks.actions import Testdrive
from materialize.checks.checks import TESTDRIVE_NOP, Check


class CreateRole(Check):
    def initialize(self) -> Testdrive:
        return Testdrive(TESTDRIVE_NOP)

    def manipulate(self) -> list[Testdrive]:
        return [
            Testdrive(dedent(s))
            for s in [
                """
                > CREATE ROLE create_role1;
                > GRANT create_role1 TO materialize;
                """,
                """
                > CREATE ROLE create_role2;
                > GRANT create_role2 TO materialize;
                """,
            ]
        ]

    def validate(self) -> Testdrive:
        return Testdrive(
            dedent(
                """
                > SELECT name FROM mz_roles WHERE name LIKE 'create_role%';
                create_role1
                create_role2
                # TODO(def-) Grantor information is currently not stable during
                # upgrades due to https://github.com/MaterializeInc/materialize/pull/18780
                # Reenable on next release
                > SELECT role.name, member.name from mz_role_members JOIN mz_roles role ON mz_role_members.role_id = role.id JOIN mz_roles member ON mz_role_members.member = member.id JOIN mz_roles grantor ON mz_role_members.grantor = grantor.id WHERE role.name LIKE 'create_role%';
                create_role1 materialize
                create_role2 materialize
                """
            )
        )


class DropRole(CreateRole):
    def initialize(self) -> Testdrive:
        return Testdrive(
            dedent(
                """
                > CREATE ROLE drop_role1;
                > GRANT drop_role1 TO materialize;
                """
            )
        )

    def manipulate(self) -> list[Testdrive]:
        return [
            Testdrive(dedent(s))
            for s in [
                """
                > REVOKE drop_role1 FROM materialize;
                > DROP ROLE drop_role1;
                > CREATE ROLE drop_role2;
                > GRANT drop_role2 TO materialize;
                """,
                """
                > REVOKE drop_role2 FROM materialize;
                > DROP ROLE drop_role2;
                """,
            ]
        ]

    def validate(self) -> Testdrive:
        return Testdrive(
            dedent(
                """
                > SELECT COUNT(*) FROM mz_roles WHERE name LIKE 'drop_role%';
                0
                > SELECT COUNT(*) FROM mz_role_members JOIN mz_roles ON mz_role_members.role_id = mz_roles.id WHERE name LIKE 'drop_role%';
                0
                """
            )
        )


class BuiltinRoles(CreateRole):
    def manipulate(self) -> list[Testdrive]:
        return [Testdrive(TESTDRIVE_NOP), Testdrive(TESTDRIVE_NOP)]

    def validate(self) -> Testdrive:
        return Testdrive(
            dedent(
                """
            $ skip-if
            SELECT mz_version_num() < 8300

            > SELECT name FROM mz_roles WHERE name IN ('mz_monitor', 'mz_monitor_redacted') ORDER BY name
            mz_monitor
            mz_monitor_redacted
            """
            )
        )