Merge pull request #34 from Tsanton/feat/database-roles

Feat/database roles

pyflake_client/models/describables/database_role.py

@@ -13,7 +13,7 @@
 
 @dataclass(frozen=True)
 class DatabaseRole(ISnowflakeDescribable, ISnowflakeGrantPrincipal):
-    """Role"""
+    """DatabaseRole"""
 
     name: str
     db_name: str

pyflake_client/models/describables/database_roles.py

@@ -0,0 +1,25 @@
+"""role"""
+from dataclasses import dataclass
+
+from dacite import Config
+
+from pyflake_client.models.describables.snowflake_describable_interface import (
+    ISnowflakeDescribable,
+)
+
+
+@dataclass(frozen=True)
+class DatabaseRoles(ISnowflakeDescribable):
+    """DatabaseRoles"""
+    db_name: str
+
+    def get_describe_statement(self) -> str:
+        """get_describe_statement"""
+        return f"SHOW DATABASE ROLES IN DATABASE {self.db_name};".upper()
+
+    def is_procedure(self) -> bool:
+        """is_procedure"""
+        return False
+
+    def get_dacite_config(self) -> Config:
+        return None

pyflake_client/models/describables/principal_ascendants.py

@@ -54,8 +54,8 @@ def show_all_roles_that_inherit_source_py(snowpark_session, principal_type: str,
     roles = show_grants_on_py(snowpark_session, principal_type, principal_identifier, links_removed)
     show_inheritance = []
     for role in roles:
+        result.append(role)
         if not role['grantee_name'] in roles_shown:
-            result.append(role)
             roles_shown.add(role['grantee_name'].upper())
             show_inheritance.append(role)
     for role in show_inheritance:
@@ -65,11 +65,7 @@ def show_all_roles_that_inherit_source_py(snowpark_session, principal_type: str,
 def main_py(snowpark_session, principal_type_py:str, principal_identifier_py:str):
     res = []
     show_all_roles_that_inherit_source_py(snowpark_session, principal_type_py, principal_identifier_py, 0, res)
-    return {
-        'principal_identifier': principal_identifier_py,
-        'principal_type': principal_type_py if principal_type_py != 'DATABASE ROLE' else 'DATABASE_ROLE',
-        'ascendants': res
-    }
+    return res
 $$
 call show_all_roles_that_inherit_source('%(s1)s', '%(s2)s');""" % {
             "s1": principal_type,

pyflake_client/models/describables/principal_descendants.py

@@ -46,11 +46,7 @@ def show_direct_descendants_from_principal_py(snowpark_session, principal_type_p
                 **row.as_dict(), 
                 **{'distance_from_source': 0 } 
             })
-    return {
-        'principal_identifier': principal_identifier_py,
-        'principal_type': principal_type_py if principal_type_py != 'DATABASE ROLE' else 'DATABASE_ROLE',
-        'descendants': res
-    }
+    return res
 $$
 call show_direct_descendants_from_principal('%(s1)s', '%(s2)s');""" % {
             "s1": principal_type,

pyflake_client/models/entities/principal_ascendants.py → pyflake_client/models/entities/principal_ascendant.py

@@ -33,19 +33,3 @@ def load_from_sf(cls, data: Dict[str, Any], config: Union[dacite.Config, None] =
         for old_key, new_key in cls.map_key_names().items():
             data[new_key] = data.pop(old_key)
         return PrincipalAscendant(**{k: data[k] for k in cls.__dataclass_fields__})
-
-@dataclass(frozen=True)
-class PrincipalAscendants(ISnowflakeEntity):
-    """PrincipalAscendants"""
-    principal_identifier: str
-    principal_type: str #TODO: enum
-    ascendants: List[PrincipalAscendant]
-
-    @classmethod
-    def load_from_sf(cls, data: Dict[str, Any], config: Union[dacite.Config, None]) -> PrincipalAscendants:
-        ascendants = [PrincipalAscendant.load_from_sf(c) for c in data["ascendants"]]
-        return PrincipalAscendants(
-            principal_identifier=data["principal_identifier"],
-            principal_type=data["principal_type"],
-            ascendants=ascendants
-        )

pyflake_client/models/enums/object_type.py

@@ -22,7 +22,7 @@ def __str__(self):
         elif self == ObjectType.FILE_FORMAT:
             return "FILE FORMAT"
         else:
-            return self
+            return self.value
 
     def singularize(self) -> str:
         """singularize"""

pyflake_client/tests/test_database_roles.py

@@ -0,0 +1,105 @@
+"""test_role"""
+import queue
+import uuid
+from datetime import date
+
+
+from pyflake_client.client import PyflakeClient
+from pyflake_client.models.assets.database_role import DatabaseRole
+from pyflake_client.models.describables.database_roles import DatabaseRoles
+from pyflake_client.models.entities.role import Role as EntitiesRole
+from pyflake_client.models.assets.role import Role as RoleAsset
+from pyflake_client.models.assets.database import Database as DatabaseAsset
+from pyflake_client.tests.utilities import find
+
+
+def test_get_database_role(flake: PyflakeClient, assets_queue: queue.LifoQueue):
+    """test_create_role"""
+    snowflake_comment:str = f"pyflake_client_test_{uuid.uuid4()}"
+    ### Arrange ###
+    database = DatabaseAsset("IGT_DEMO", snowflake_comment, owner=RoleAsset("SYSADMIN"))
+    role: DatabaseRole = DatabaseRole(
+        name="IGT_ROLE",
+        database_name=database.db_name,
+        comment=snowflake_comment,
+        owner=RoleAsset("USERADMIN"),
+    )
+
+    try:
+        flake.register_asset(database, assets_queue)
+        flake.create_asset(role)
+
+        ### Act ###
+        roles = flake.describe_many(DatabaseRoles(db_name=database.db_name), EntitiesRole)
+
+        ### Assert ###
+        assert roles is not None
+        assert len(roles) == 1
+        db_role = find(roles, lambda x: x.name == role.name)
+        assert db_role is not None
+        assert db_role.owner == "USERADMIN"
+    finally:
+        ### Cleanup ###
+        flake.delete_assets(assets_queue)
+
+
+def test_get_database_roles(flake: PyflakeClient, assets_queue: queue.LifoQueue):
+    """test_create_role"""
+    snowflake_comment:str = f"pyflake_client_test_{uuid.uuid4()}"
+    ### Arrange ###
+    database = DatabaseAsset("IGT_DEMO", snowflake_comment, owner=RoleAsset("SYSADMIN"))
+    role_1: DatabaseRole = DatabaseRole(
+        name="IGT_ROLE_1",
+        database_name=database.db_name,
+        comment=snowflake_comment,
+        owner=RoleAsset("USERADMIN"),
+    )
+    role_2: DatabaseRole = DatabaseRole(
+        name="IGT_ROLE_2",
+        database_name=database.db_name,
+        comment=snowflake_comment,
+        owner=RoleAsset("USERADMIN"),
+    )
+
+    try:
+        flake.register_asset(database, assets_queue)
+        flake.create_asset(role_1)
+        flake.create_asset(role_2)
+
+        ### Act ###
+        roles = flake.describe_many(DatabaseRoles(db_name=database.db_name), EntitiesRole)
+
+        ### Assert ###
+        assert roles is not None
+        assert len(roles) == 2
+
+        db_role_1 = find(roles, lambda x: x.name == role_1.name)
+        assert db_role_1 is not None
+        assert db_role_1.owner == "USERADMIN"
+
+        db_role_2 = find(roles, lambda x: x.name == role_2.name)
+        assert db_role_2 is not None
+        assert db_role_2.owner == "USERADMIN"
+    finally:
+        ### Cleanup ###
+        flake.delete_assets(assets_queue)
+
+
+
+def test_get_database_role_none_exist(flake: PyflakeClient, assets_queue: queue.LifoQueue):
+    """test_create_role"""
+    snowflake_comment:str = f"pyflake_client_test_{uuid.uuid4()}"
+    ### Arrange ###
+    database = DatabaseAsset("IGT_DEMO", snowflake_comment, owner=RoleAsset("SYSADMIN"))
+
+    try:
+        flake.register_asset(database, assets_queue)
+        ### Act ###
+        roles = flake.describe_many(DatabaseRoles(db_name=database.db_name), EntitiesRole)
+
+        ### Assert ###
+        assert roles is not None
+        assert len(roles) == 0
+    finally:
+        ### Cleanup ###
+        flake.delete_assets(assets_queue)

pyflake_client/tests/test_principal_ascendants.py

@@ -3,6 +3,7 @@
 # pylint: disable=too-many-locals
 
 import queue
+from typing import List
 import uuid
 
 from pyflake_client.client import PyflakeClient
@@ -16,23 +17,20 @@
 from pyflake_client.models.describables.database_role import DatabaseRole as DatabaseRoleDescribable
 from pyflake_client.models.describables.principal_ascendants import PrincipalAscendants as RoleAscendantsDescribable
 
-from pyflake_client.models.entities.principal_ascendants import PrincipalAscendants as RoleAscendantsEntity
-
-
+from pyflake_client.models.entities.principal_ascendant import PrincipalAscendant
+from pyflake_client.tests.utilities import find
 
 def test_get_principal_ascendants(flake: PyflakeClient):
     """test_get_principal_ascendants: we know that USERADMIN -> SECURITYADMIN -> ACCOUNTADMIN"""
     ### Act ###
-    snowflake_comment:str = f"pyflake_client_test_{uuid.uuid4()}"
-    hierarchy: RoleAscendantsEntity = flake.describe_one(RoleAscendantsDescribable(RoleDescribable("USERADMIN")), RoleAscendantsEntity)
+    ascendants: List[PrincipalAscendant] = flake.describe_many(RoleAscendantsDescribable(RoleDescribable("USERADMIN")), PrincipalAscendant)
 
-    sec_admin = next((r for r in hierarchy.ascendants if r.grantee_identifier == "SECURITYADMIN"), None)
-    acc_admin = next((r for r in hierarchy.ascendants if r.grantee_identifier == "ACCOUNTADMIN"), None)
-    sys_admin = next((r for r in hierarchy.ascendants if r.grantee_identifier == "SYSADMIN"), None)
+    sec_admin = find(ascendants, lambda x: x.grantee_identifier == "SECURITYADMIN" and x.granted_identifier == "USERADMIN")
+    acc_admin = find(ascendants, lambda x: x.grantee_identifier == "ACCOUNTADMIN" and x.granted_identifier == "SECURITYADMIN")
+    sys_admin = find(ascendants, lambda x: x.grantee_identifier == "SYSADMIN")
 
     ### Assert ###
-    assert hierarchy.principal_identifier == "USERADMIN"
-    assert hierarchy.principal_type == "ROLE"
+
     assert sys_admin is None
 
     assert sec_admin is not None
@@ -79,19 +77,17 @@ def test_create_role_ascendants(flake: PyflakeClient, assets_queue: queue.LifoQu
         flake.register_asset(sysadmin_great_grandparent_relationship, assets_queue)
 
         ### Act ###
-        hierarchy: RoleAscendantsEntity = flake.describe_one(RoleAscendantsDescribable(RoleDescribable("IGT_CHILD1_ROLE")), RoleAscendantsEntity)
-        parent = next((r for r in hierarchy.ascendants if r.grantee_identifier == parent_role.name), None)
-        grandparent1 = next((r for r in hierarchy.ascendants if r.grantee_identifier == grandparent1_role.name), None)
-        grandparent2 = next((r for r in hierarchy.ascendants if r.grantee_identifier == grandparent2_role.name), None)
-        great_grandparent = next((r for r in hierarchy.ascendants if r.grantee_identifier == great_grandparent_role.name), None)
-        sys_admin = next((r for r in hierarchy.ascendants if r.grantee_identifier == "SYSADMIN"), None)
-        acc_admin = next((r for r in hierarchy.ascendants if r.grantee_identifier == "ACCOUNTADMIN"), None)
+        ascendants: List[PrincipalAscendant] = flake.describe_many(RoleAscendantsDescribable(RoleDescribable("IGT_CHILD1_ROLE")), PrincipalAscendant)
+
+        parent = find(ascendants, lambda x: x.grantee_identifier == parent_role.name)
+        grandparent1 = find(ascendants, lambda x: x.grantee_identifier == grandparent1_role.name)
+        grandparent2 = find(ascendants, lambda x: x.grantee_identifier == grandparent2_role.name)
+        great_grandparent = find(ascendants, lambda x: x.grantee_identifier == great_grandparent_role.name)
+        sys_admin = find(ascendants, lambda x: x.grantee_identifier == "SYSADMIN")
+        acc_admin = find(ascendants, lambda x: x.grantee_identifier == "ACCOUNTADMIN" and x.granted_identifier == "SYSADMIN")
 
         ### Assert ###
         # {parent: 0, grandparent: 1, great_grandparent: 2, sysadmin: 3, accountadmin: 4}
-        assert hierarchy.principal_identifier == "IGT_CHILD1_ROLE"
-        assert hierarchy.principal_type == "ROLE"
-
         assert parent is not None
         assert parent.distance_from_source == 0
         assert parent.granted_identifier == child1_role.name
@@ -126,7 +122,7 @@ def test_broken_role_ascendants(flake: PyflakeClient, assets_queue: queue.LifoQu
     ### Arrange ###
     snowflake_comment:str = f"pyflake_client_test_{uuid.uuid4()}"
     user_admin_role = Role("USERADMIN")
-    sys_admin_role = Role("SYSADMIN")
+    # sys_admin_role = Role("SYSADMIN")
     child1_role = Role("IGT_CHILD1_ROLE", snowflake_comment, user_admin_role)
     child2_role = Role("IGT_CHILD2_ROLE", snowflake_comment, user_admin_role)
     parent_role = Role("IGT_PARENT_ROLE", snowflake_comment, user_admin_role)
@@ -156,18 +152,15 @@ def test_broken_role_ascendants(flake: PyflakeClient, assets_queue: queue.LifoQu
         flake.register_asset(sysadmin_great_grandparent_relationship, assets_queue)
 
         ### Act ###
-        hierarchy: RoleAscendantsEntity = flake.describe_one(RoleAscendantsDescribable(RoleDescribable("IGT_CHILD1_ROLE")), RoleAscendantsEntity)
-        parent = next((r for r in hierarchy.ascendants if r.grantee_identifier == parent_role.name), None)
-        grandparent = next((r for r in hierarchy.ascendants if r.grantee_identifier == grandparent_role.name), None)
-        great_grandparent = next((r for r in hierarchy.ascendants if r.grantee_identifier == great_grandparent_role.name), None)
-        sys_admin = next((r for r in hierarchy.ascendants if r.grantee_identifier == "SYSADMIN"), None)
-        acc_admin = next((r for r in hierarchy.ascendants if r.grantee_identifier == "ACCOUNTADMIN"), None)
+        ascendants: List[PrincipalAscendant] = flake.describe_many(RoleAscendantsDescribable(RoleDescribable("IGT_CHILD1_ROLE")), PrincipalAscendant)
+        parent = find(ascendants, lambda x: x.grantee_identifier == parent_role.name)
+        grandparent = find(ascendants, lambda x: x.grantee_identifier == grandparent_role.name)
+        great_grandparent = find(ascendants, lambda x: x.grantee_identifier == great_grandparent_role.name)
+        sys_admin = find(ascendants, lambda x: x.grantee_identifier == "SYSADMIN")
+        acc_admin = find(ascendants, lambda x: x.grantee_identifier == "ACCOUNTADMIN" and x.granted_identifier == "SYSADMIN")
 
         ### Assert ###
         # {parent: 0, grandparent: 1, great_grandparent: 2, sysadmin: 3, accountadmin: 4}
-        assert hierarchy.principal_identifier == "IGT_CHILD1_ROLE"
-        assert hierarchy.principal_type == "ROLE"
-
         assert parent is not None
         assert parent.distance_from_source == 0
         assert parent.granted_identifier == child1_role.name
@@ -217,17 +210,14 @@ def test_ascendants_with_database_roles(flake: PyflakeClient, assets_queue: queu
         flake.register_asset(rel4, assets_queue)
 
         ### Act ###
-        hierarchy: RoleAscendantsEntity = flake.describe_one(RoleAscendantsDescribable(DatabaseRoleDescribable(dr_r.name, dr_r.database_name)), RoleAscendantsEntity)
-        db_rw = next((r for r in hierarchy.ascendants if r.grantee_identifier == dr_rw.get_identifier()), None)
-        db_rwc = next((r for r in hierarchy.ascendants if r.grantee_identifier == dr_rwc.get_identifier()), None)
-        db_sysadmin = next((r for r in hierarchy.ascendants if r.grantee_identifier == dr_sys.get_identifier()), None)
-        sys_admin = next((r for r in hierarchy.ascendants if r.grantee_identifier == "SYSADMIN"), None)
-        acc_admin = next((r for r in hierarchy.ascendants if r.grantee_identifier == "ACCOUNTADMIN"), None)
+        ascendants: List[PrincipalAscendant] = flake.describe_many(RoleAscendantsDescribable(DatabaseRoleDescribable(dr_r.name, dr_r.database_name)),PrincipalAscendant)
+        db_rw = find(ascendants, lambda x: x.grantee_identifier == dr_rw.get_identifier())
+        db_rwc = find(ascendants, lambda x: x.grantee_identifier == dr_rwc.get_identifier())
+        db_sysadmin = find(ascendants, lambda x: x.grantee_identifier == dr_sys.get_identifier())
+        sys_admin = find(ascendants, lambda x: x.grantee_identifier == "SYSADMIN")
+        acc_admin = find(ascendants, lambda x: x.grantee_identifier == "ACCOUNTADMIN" and x.granted_identifier == "SYSADMIN")
 
         ### Assert ###
-        assert hierarchy.principal_identifier == dr_r.get_identifier()
-        assert hierarchy.principal_type == "DATABASE_ROLE"
-
         assert db_rw is not None
         assert db_rw.distance_from_source == 0
         assert db_rw.granted_identifier == dr_r.get_identifier()