Set permissions and role for tests

backend/tests/helpers/test_app.py

@@ -9,8 +9,11 @@
 from infrahub.core.branch import Branch
 from infrahub.core.initialization import (
     create_account,
+    create_administrator_role,
+    create_administrators_group,
     create_default_branch,
     create_global_branch,
+    create_global_permissions,
     create_root_node,
     initialization,
 )
@@ -104,11 +107,11 @@ async def client(
     async def initialize_registry(
         self, db: InfrahubDatabase, register_core_schema: SchemaBranch, bus_simulator: BusSimulator, api_token: str
     ) -> None:
-        await create_account(
-            db=db,
-            name="admin",
-            password=config.SETTINGS.initial.admin_password,
-            token_value=api_token,
+        admin_account = await create_account(
+            db=db, name="admin", password=config.SETTINGS.initial.admin_password, token_value=api_token
         )
+        global_permissions = await create_global_permissions(db=db)
+        administrator_role = await create_administrator_role(db=db, global_permissions=global_permissions)
+        await create_administrators_group(db=db, role=administrator_role, admin_accounts=[admin_account])
 
         await initialization(db=db)

backend/tests/integration/conftest.py

@@ -89,18 +89,11 @@ async def create_token(self, account_name: Optional[str] = None) -> str:
         token = str(UUIDT())
         account_name = account_name or "admin"
         response = await NodeManager.query(
-            schema=InfrahubKind.ACCOUNT,
-            db=self.db,
-            filters={"name__value": account_name},
-            limit=1,
+            schema=InfrahubKind.ACCOUNT, db=self.db, filters={"name__value": account_name}, limit=1
         )
         account = response[0]
         account_token = await Node.init(db=self.db, schema=InfrahubKind.ACCOUNTTOKEN)
-        await account_token.new(
-            db=self.db,
-            token=token,
-            account=account,
-        )
+        await account_token.new(db=self.db, token=token, account=account)
         await account_token.save(db=self.db)
         return token
 

backend/tests/integration/diff/test_diff_update.py

@@ -51,11 +51,7 @@
 class TestDiffUpdateConflict(TestInfrahubApp):
     @pytest.fixture(scope="class")
     async def initial_dataset(
-        self,
-        db: InfrahubDatabase,
-        default_branch,
-        client: InfrahubClient,
-        bus_simulator: BusSimulator,
+        self, db: InfrahubDatabase, default_branch, client: InfrahubClient, bus_simulator: BusSimulator
     ) -> None:
         await load_schema(db, schema=CAR_SCHEMA)
         john = await Node.init(schema=TestKind.PERSON, db=db)

backend/tests/integration/git/test_git_repository.py

@@ -50,10 +50,7 @@ async def base_dataset(self, db: InfrahubDatabase):
         await initialization(db=db)
 
     @pytest.fixture(scope="class")
-    async def test_client(
-        self,
-        base_dataset,
-    ) -> InfrahubTestClient:
+    async def test_client(self, base_dataset) -> InfrahubTestClient:
         await app_initialization(app)
         return InfrahubTestClient(app=app)
 
@@ -67,11 +64,7 @@ async def client(self, test_client: InfrahubTestClient, integration_helper):
     @pytest.fixture(scope="class")
     async def query_99(self, db: InfrahubDatabase, test_client):
         obj = await Node.init(schema=InfrahubKind.GRAPHQLQUERY, db=db)
-        await obj.new(
-            db=db,
-            name="query99",
-            query="query query99 { CoreRepository { edges { node { id }}}}",
-        )
+        await obj.new(db=db, name="query99", query="query query99 { CoreRepository { edges { node { id }}}}")
         await obj.save(db=db)
         return obj
 

python_sdk/tests/integration/conftest.py

@@ -39,13 +39,7 @@ def _request(
         if payload:
             content = str(ujson.dumps(payload)).encode("UTF-8")
         with self as client:
-            return client.request(
-                method=method.value,
-                url=url,
-                headers=headers,
-                timeout=timeout,
-                content=content,
-            )
+            return client.request(method=method.value, url=url, headers=headers, timeout=timeout, content=content)
 
     async def async_request(
         self, url: str, method: HTTPMethod, headers: dict[str, Any], timeout: int, payload: Optional[dict] = None

python_sdk/tests/integration/test_infrahub_client_sync.py

@@ -33,11 +33,7 @@ async def test_client(self) -> InfrahubTestClient:
 
     @pytest.fixture
     def client(self, test_client: InfrahubTestClient):
-        config = Config(
-            username="admin",
-            password="infrahub",
-            sync_requester=test_client.sync_request,
-        )
+        config = Config(username="admin", password="infrahub", sync_requester=test_client.sync_request)
         return InfrahubClientSync(config=config)
 
     @pytest.fixture(scope="class")
@@ -63,21 +59,13 @@ async def base_dataset(self, db: InfrahubDatabase, test_client: InfrahubTestClie
 
         obj2 = await Node.init(schema="CoreRepository", db=db)
         await obj2.new(
-            db=db,
-            name="repository1",
-            description="test repository",
-            location="[email protected]:mock/test.git",
+            db=db, name="repository1", description="test repository", location="[email protected]:mock/test.git"
         )
         await obj2.save(db=db)
 
         obj3 = await Node.init(schema="CoreTransformJinja2", db=db)
         await obj3.new(
-            db=db,
-            name="rfile1",
-            description="test rfile",
-            template_path="mytemplate.j2",
-            repository=obj2,
-            query=obj1,
+            db=db, name="rfile1", description="test rfile", template_path="mytemplate.j2", repository=obj2, query=obj1
         )
         await obj3.save(db=db)
 
@@ -256,10 +244,7 @@ def test_recorder_with_playback(
         nodes = client.all(kind="CoreRepository")
 
         playback_config = JSONPlayback(directory=str(tmp_path))
-        config = Config(
-            address=client.config.address,
-            sync_requester=playback_config.sync_request,
-        )
+        config = Config(address=client.config.address, sync_requester=playback_config.sync_request)
         playback = InfrahubClientSync(config=config)
         recorded_nodes = playback.all(kind="CoreRepository")