Module `materialize.checks.all_checks.roles`

Expand source code Browse git

# 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
from materialize.checks.executors import Executor
from materialize.mz_version import MzVersion


class CreateRole(Check):
    def _can_run(self, e: Executor) -> bool:
        return self.base_version >= MzVersion.parse_mz("v0.45.0-dev")

    def _if_can_grant_revoke(self, text: str) -> str:
        if self.base_version >= MzVersion.parse_mz("v0.47.0-dev"):
            return text
        return ""

    def initialize(self) -> Testdrive:
        return Testdrive(TESTDRIVE_NOP)

    def manipulate(self) -> list[Testdrive]:
        return [
            Testdrive(dedent(s))
            for s in [
                """
                > CREATE ROLE create_role1;
                """
                + self._if_can_grant_revoke(
                    """
                > GRANT create_role1 TO materialize;
                """
                ),
                """
                > CREATE ROLE create_role2;
                """
                + self._if_can_grant_revoke(
                    """
                > 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
                + self._if_can_grant_revoke(
                    """
                > 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;
                """
                + self._if_can_grant_revoke(
                    """
                > GRANT drop_role1 TO materialize;
                """
                )
            )
        )

    def manipulate(self) -> list[Testdrive]:
        return [
            Testdrive(dedent(s))
            for s in [
                self._if_can_grant_revoke(
                    """
                > REVOKE drop_role1 FROM materialize;
                """
                )
                + """
                > DROP ROLE drop_role1;
                > CREATE ROLE drop_role2;
                """
                + self._if_can_grant_revoke(
                    """
                > GRANT drop_role2 TO materialize;
                """
                ),
                self._if_can_grant_revoke(
                    """
                > 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
                """
                + self._if_can_grant_revoke(
                    """
                > 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
            """
            )
        )

Classes

class BuiltinRoles (base_version: MzVersion, rng: random.Random | None)

Expand source code Browse git

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
            """
            )
        )

Ancestors

CreateRole
Check

Methods

def manipulate(self) ‑> list[Testdrive]

Expand source code Browse git

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

def validate(self) ‑> Testdrive

Expand source code Browse git

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
        """
        )
    )

class CreateRole (base_version: MzVersion, rng: random.Random | None)

Expand source code Browse git

class CreateRole(Check):
    def _can_run(self, e: Executor) -> bool:
        return self.base_version >= MzVersion.parse_mz("v0.45.0-dev")

    def _if_can_grant_revoke(self, text: str) -> str:
        if self.base_version >= MzVersion.parse_mz("v0.47.0-dev"):
            return text
        return ""

    def initialize(self) -> Testdrive:
        return Testdrive(TESTDRIVE_NOP)

    def manipulate(self) -> list[Testdrive]:
        return [
            Testdrive(dedent(s))
            for s in [
                """
                > CREATE ROLE create_role1;
                """
                + self._if_can_grant_revoke(
                    """
                > GRANT create_role1 TO materialize;
                """
                ),
                """
                > CREATE ROLE create_role2;
                """
                + self._if_can_grant_revoke(
                    """
                > 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
                + self._if_can_grant_revoke(
                    """
                > 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
                """
                )
            )
        )

Ancestors

Check

Methods

def initialize(self) ‑> Testdrive

Expand source code Browse git

def initialize(self) -> Testdrive:
    return Testdrive(TESTDRIVE_NOP)

def manipulate(self) ‑> list[Testdrive]

Expand source code Browse git

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

def validate(self) ‑> Testdrive

Expand source code Browse git

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
            + self._if_can_grant_revoke(
                """
            > 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 (base_version: MzVersion, rng: random.Random | None)

Expand source code Browse git

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

    def manipulate(self) -> list[Testdrive]:
        return [
            Testdrive(dedent(s))
            for s in [
                self._if_can_grant_revoke(
                    """
                > REVOKE drop_role1 FROM materialize;
                """
                )
                + """
                > DROP ROLE drop_role1;
                > CREATE ROLE drop_role2;
                """
                + self._if_can_grant_revoke(
                    """
                > GRANT drop_role2 TO materialize;
                """
                ),
                self._if_can_grant_revoke(
                    """
                > 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
                """
                + self._if_can_grant_revoke(
                    """
                > SELECT COUNT(*) FROM mz_role_members JOIN mz_roles ON mz_role_members.role_id = mz_roles.id WHERE name LIKE 'drop_role%';
                0
                """
                )
            )
        )

Ancestors

CreateRole
Check

Methods

def initialize(self) ‑> Testdrive

Expand source code Browse git

def initialize(self) -> Testdrive:
    return Testdrive(
        dedent(
            """
            > CREATE ROLE drop_role1;
            """
            + self._if_can_grant_revoke(
                """
            > GRANT drop_role1 TO materialize;
            """
            )
        )
    )

def manipulate(self) ‑> list[Testdrive]

Expand source code Browse git

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

def validate(self) ‑> Testdrive

Expand source code Browse git

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