format

Author: minitriga

infrahub_sdk/ctl/example_repo/tasks.py

@@ -1,7 +1,8 @@
-from invoke import Context, task
+import os
 from pathlib import Path
+
 import httpx
-import os
+from invoke import Context, task
 
 # If no version is indicated, we will take the latest
 VERSION = os.getenv("INFRAHUB_IMAGE_VER", None)
@@ -12,7 +13,7 @@ def start(context: Context) -> None:
     """
     Start the services using docker-compose in detached mode.
     """
-    compose_file = download_compose_file(context, override=False)
+    download_compose_file(context, override=False)
     context.run("docker compose up -d")
 
 
@@ -21,7 +22,7 @@ def destroy(context: Context) -> None:
     """
     Stop and remove containers, networks, and volumes.
     """
-    compose_file = download_compose_file(context, override=False)
+    download_compose_file(context, override=False)
     context.run("docker compose down -v")
 
 
@@ -30,7 +31,7 @@ def stop(context: Context) -> None:
     """
     Stop containers and remove networks.
     """
-    compose_file = download_compose_file(context, override=False)
+    download_compose_file(context, override=False)
     context.run("docker compose down")
 
 
@@ -64,7 +65,7 @@ def test(ctx: Context):
 
 
 @task(help={"override": "Redownload the compose file even if it already exists."})
-def download_compose_file(context: Context, override: bool = False) -> Path:
+def download_compose_file(context: Context, override: bool = False) -> Path:  # noqa ARG001
     """
     Download docker-compose.yml from InfraHub if missing or override is True.
     """

infrahub_sdk/ctl/example_repo/{% if package_mode %}lib{% endif %}/example.py

@@ -1,9 +1,8 @@
-from infrahub_sdk import InfrahubClient
-from infrahub_sdk.node import InfrahubNode
-from typing import List
 import logging
 
+from infrahub_sdk.node import InfrahubNode
+
 
-def print_nodes(client: InfrahubClient, log: logging.Logger, nodes: List[InfrahubNode]):
+def print_nodes(log: logging.Logger, nodes: list[InfrahubNode]):
     for node in nodes.keys():
         log.info(f"{node} present.")

infrahub_sdk/ctl/example_repo/{% if scripts %}scripts{% endif %}/example_script.py

@@ -1,14 +1,15 @@
 import logging
 
-from infrahub_sdk import InfrahubClient
 from lib.example import print_nodes
 
+from infrahub_sdk import InfrahubClient
+
 
 async def run(
     client: InfrahubClient,
     log: logging.Logger,
     branch: str,
 ):
-    log.info("Running example script...")
+    log.info(f"Running example script on {branch}...")
     nodes = await client.schema.all()
-    print_nodes(client, log, nodes)
+    print_nodes(log, nodes)

infrahub_sdk/ctl/example_repo/{% if tests %}tests{% endif %}/integration/conftest.py

@@ -1,7 +1,8 @@
-import pytest
 from pathlib import Path
-
 from typing import Any
+
+import pytest
+
 from infrahub_sdk.yaml import SchemaFile
 
 CURRENT_DIRECTORY = Path(__file__).parent.resolve()

infrahub_sdk/ctl/example_repo/{% if tests %}tests{% endif %}/integration/test_infrahub.py

@@ -1,14 +1,16 @@
-import pytest
 from pathlib import Path
+
+import pytest
+
 from infrahub_sdk import InfrahubClient
-from infrahub_sdk.testing.docker import TestInfrahubDockerClient
 from infrahub_sdk.protocols import CoreGenericRepository
+from infrahub_sdk.testing.docker import TestInfrahubDockerClient
 from infrahub_sdk.testing.repository import GitRepo
 
 
 class TestInfrahub(TestInfrahubDockerClient):
     @pytest.mark.asyncio
-    async def test_load_schema(self, default_branch: str, client: InfrahubClient, schemas):
+    async def test_load_schema(self, default_branch: str, client: InfrahubClient, schemas: list[dict]):
         await client.schema.wait_until_converged(branch=default_branch)
 
         resp = await client.schema.load(schemas=schemas, branch=default_branch, wait_until_converged=True)
@@ -34,7 +36,7 @@ async def test_load_repository(
 
         repos = await client.all(kind=CoreGenericRepository)
 
-        ### A breakpoint can be added to pause the tests from running and keep the test containers active
+        # A breakpoint can be added to pause the tests from running and keep the test containers active
         # breakpoint()
 
         assert repos

pyproject.toml

@@ -283,6 +283,11 @@ max-complexity = 17
     "ANN204", # Missing return type annotation for special method
 ]
 
+"infrahub_sdk/ctl/example_repo/**/*.py" = [
+    "S101",    # Use of assert detected
+    "N999",    # Invalid module name
+]
+
 "tests/unit/sdk/test_client.py" = [
     "W293", # Blank line contains whitespace (used within output check)
 ]

tests/unit/ctl/test_repository_app.py

@@ -293,9 +293,10 @@ def test_repo_list(self, mock_repositories_list) -> None:
         assert strip_color(result.stdout) == read_fixture("output.txt", "integration/test_infrahubctl/repository_list")
 
     def test_repo_init(self) -> None:
-        with tempfile.TemporaryDirectory() as temp_dst, tempfile.NamedTemporaryFile(
-            mode="w", suffix=".yml", delete=False, encoding="utf-8"
-        ) as temp_yaml:
+        with (
+            tempfile.TemporaryDirectory() as temp_dst,
+            tempfile.NamedTemporaryFile(mode="w", suffix=".yml", delete=False, encoding="utf-8") as temp_yaml,
+        ):
             dst = Path(temp_dst)
             yaml_path = Path(temp_yaml.name)