Merge pull request #8493 from opsmill/pog-api-component-test-annotations

Fix type annotations

Author: ogenstad

backend/tests/component/api/conftest.py

@@ -6,19 +6,24 @@
 from fastapi.testclient import TestClient
 
 from infrahub import config
+from infrahub.core.branch import Branch
 from infrahub.core.constants import InfrahubKind
 from infrahub.core.initialization import create_branch
 from infrahub.core.manager import NodeManager
 from infrahub.core.node import Node
+from infrahub.core.schema.schema_branch import SchemaBranch
 from infrahub.core.timestamp import Timestamp
 from infrahub.database import InfrahubDatabase
 from infrahub.services.adapters.workflow.local import WorkflowLocalExecution
 from infrahub.workers.dependencies import build_database, build_message_bus, build_workflow
 from infrahub.workflows.initialization import setup_task_manager
+from tests.conftest import TestHelper
 
 
 @pytest.fixture
-def client(dependency_provider: Provider, nats, redis) -> Generator[TestClient, None, None]:
+def client(
+    dependency_provider: Provider, nats: dict[int, int] | None, redis: dict[int, int] | None
+) -> Generator[TestClient, None, None]:
     # In order to mock some methods later we can't load app by default because it will automatically load all import in main.py as well
     from infrahub.server import app
 
@@ -40,7 +45,7 @@ def admin_headers() -> dict[str, str]:
 
 
 @pytest.fixture
-def rpc_bus(helper, dependency_provider) -> Generator[Any, None, None]:
+def rpc_bus(helper: TestHelper, dependency_provider: Provider) -> Generator[Any, None, None]:
     original = config.OVERRIDE.message_bus
     bus = helper.get_message_bus_rpc()
     config.OVERRIDE.message_bus = bus
@@ -50,7 +55,7 @@ def rpc_bus(helper, dependency_provider) -> Generator[Any, None, None]:
 
 
 @pytest.fixture
-async def workflow_local(dependency_provider) -> AsyncGenerator[WorkflowLocalExecution, None]:
+async def workflow_local(dependency_provider: Provider) -> AsyncGenerator[WorkflowLocalExecution, None]:
     original = config.OVERRIDE.workflow
     workflow = WorkflowLocalExecution()
     await setup_task_manager()
@@ -62,7 +67,10 @@ async def workflow_local(dependency_provider) -> AsyncGenerator[WorkflowLocalExe
 
 @pytest.fixture
 async def car_person_data(
-    db: InfrahubDatabase, register_core_models_schema, car_person_schema, first_account
+    db: InfrahubDatabase,
+    register_core_models_schema: SchemaBranch,
+    car_person_schema: SchemaBranch,
+    first_account: Node,
 ) -> dict[str, Node]:
     p1 = await Node.init(db=db, schema="TestPerson")
     await p1.new(db=db, name="John", height=180)
@@ -148,7 +156,7 @@ async def car_person_data(
 
 @pytest.fixture
 async def car_person_data_generic_diff(
-    db: InfrahubDatabase, default_branch, car_person_data_generic, first_account
+    db: InfrahubDatabase, default_branch: Branch, car_person_data_generic: dict[str, Node], first_account: Node
 ) -> dict[str, Any]:
     branch2 = await create_branch(branch_name="branch2", db=db)
 
@@ -239,7 +247,7 @@ async def car_person_data_generic_diff(
 
 @pytest.fixture
 async def car_person_data_artifact_diff(
-    db: InfrahubDatabase, default_branch, car_person_data_generic_diff
+    db: InfrahubDatabase, default_branch: Branch, car_person_data_generic_diff: dict[str, Any]
 ) -> dict[str, Any]:
     query = """
     query {
@@ -368,7 +376,7 @@ async def car_person_data_artifact_diff(
 
 @pytest.fixture
 async def data_diff_attribute(
-    db: InfrahubDatabase, default_branch, car_person_data_generic, first_account
+    db: InfrahubDatabase, default_branch: Branch, car_person_data_generic: dict[str, Node], first_account: Node
 ) -> dict[str, Any]:
     branch2 = await create_branch(branch_name="branch2", db=db)
 

backend/tests/component/api/test_00_auth.py

@@ -3,6 +3,7 @@
 
 from infrahub import config
 from infrahub.core.branch import Branch
+from infrahub.core.node import Node
 from infrahub.database import InfrahubDatabase
 
 EXPIRED_ACCESS_TOKEN = (
@@ -22,7 +23,9 @@
 )
 
 
-async def test_password_based_login(db: InfrahubDatabase, default_branch, client, first_account) -> None:
+async def test_password_based_login(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
+) -> None:
     with client:
         response = client.post("/api/auth/login", json={"username": "First Account", "password": "FirstPassword123"})
 
@@ -37,7 +40,9 @@ async def test_password_based_login(db: InfrahubDatabase, default_branch, client
     assert first_account.id == decoded["sub"]
 
 
-async def test_refresh_with_invalidated_token(db: InfrahubDatabase, default_branch, client, first_account) -> None:
+async def test_refresh_with_invalidated_token(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
+) -> None:
     with client:
         response = client.post("/api/auth/login", json={"username": "First Account", "password": "FirstPassword123"})
 
@@ -64,7 +69,9 @@ async def test_refresh_with_invalidated_token(db: InfrahubDatabase, default_bran
     }
 
 
-async def test_refresh_access_token(db: InfrahubDatabase, default_branch, client, first_account) -> None:
+async def test_refresh_access_token(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
+) -> None:
     """Validate that it's possible to refresh an access token using a refresh token"""
     with client:
         login_response = client.post(
@@ -88,7 +95,9 @@ async def test_refresh_access_token(db: InfrahubDatabase, default_branch, client
     assert decoded_access["session_id"] == decoded_refresh["session_id"]
 
 
-async def test_refresh_access_token_with_cookies(db: InfrahubDatabase, default_branch, client, first_account) -> None:
+async def test_refresh_access_token_with_cookies(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
+) -> None:
     """Validate that it's possible to refresh an access token using a refresh token stored in cookies"""
     with client:
         login_response = client.post(
@@ -119,7 +128,7 @@ async def test_refresh_access_token_with_cookies(db: InfrahubDatabase, default_b
 
 
 async def test_fail_to_refresh_access_token_with_access_token(
-    db: InfrahubDatabase, default_branch, client, first_account
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
 ) -> None:
     """Validate that it's not possible to refresh an access token using an access token"""
     with client:
@@ -143,7 +152,9 @@ async def test_fail_to_refresh_access_token_with_access_token(
     }
 
 
-async def test_password_based_login_unknown_user(db: InfrahubDatabase, default_branch, client, first_account) -> None:
+async def test_password_based_login_unknown_user(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
+) -> None:
     with client:
         response = client.post("/api/auth/login", json={"username": "i-do-not-exist", "password": "something"})
 
@@ -155,7 +166,7 @@ async def test_password_based_login_unknown_user(db: InfrahubDatabase, default_b
 
 
 async def test_password_based_login_invalid_password(
-    db: InfrahubDatabase, default_branch, client, first_account
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
 ) -> None:
     with client:
         response = client.post("/api/auth/login", json={"username": "First Account", "password": "incorrect"})
@@ -190,7 +201,9 @@ async def test_refresh_access_token_with_expired_refresh_token(
     assert response.json() == {"data": None, "errors": [{"message": "Expired Signature", "extensions": {"code": 401}}]}
 
 
-async def test_access_resource_using_refresh_token(db: InfrahubDatabase, default_branch, client, first_account) -> None:
+async def test_access_resource_using_refresh_token(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
+) -> None:
     """It should not be possible to access a resource using a refresh token"""
     with client:
         login_response = client.post(
@@ -210,7 +223,9 @@ async def test_access_resource_using_refresh_token(db: InfrahubDatabase, default
     }
 
 
-async def test_generate_api_token(db: InfrahubDatabase, default_branch, client, create_test_admin) -> None:
+async def test_generate_api_token(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, create_test_admin: Node
+) -> None:
     """It should not be possible to generate an API token using a JWT token"""
     with client:
         login_response = client.post(

backend/tests/component/api/test_01_auth_cookies.py

@@ -1,10 +1,15 @@
 import jwt
+from fastapi.testclient import TestClient
 
 from infrahub import config
+from infrahub.core.branch import Branch
+from infrahub.core.node import Node
 from infrahub.database import InfrahubDatabase
 
 
-async def test_password_based_login(db: InfrahubDatabase, default_branch, client, first_account) -> None:
+async def test_password_based_login(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
+) -> None:
     with client:
         response = client.post("/api/auth/login", json={"username": "First Account", "password": "FirstPassword123"})
 
@@ -19,7 +24,9 @@ async def test_password_based_login(db: InfrahubDatabase, default_branch, client
     assert first_account.id == decoded["sub"]
 
 
-async def test_refresh_access_token(db: InfrahubDatabase, default_branch, client, first_account) -> None:
+async def test_refresh_access_token(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
+) -> None:
     """Validate that it's possible to refresh an access token using a refresh token"""
     with client:
         login_response = client.post(
@@ -43,7 +50,9 @@ async def test_refresh_access_token(db: InfrahubDatabase, default_branch, client
     assert decoded_access["session_id"] == decoded_refresh["session_id"]
 
 
-async def test_access_resource_using_refresh_token(db: InfrahubDatabase, default_branch, client, first_account) -> None:
+async def test_access_resource_using_refresh_token(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, first_account: Node
+) -> None:
     """It should not be possible to access a resource using a refresh token"""
     with client:
         login_response = client.post(
@@ -64,7 +73,9 @@ async def test_access_resource_using_refresh_token(db: InfrahubDatabase, default
     }
 
 
-async def test_generate_api_token(db: InfrahubDatabase, default_branch, client, create_test_admin) -> None:
+async def test_generate_api_token(
+    db: InfrahubDatabase, default_branch: Branch, client: TestClient, create_test_admin: Node
+) -> None:
     """It should not be possible to generate an API token using a JWT token"""
     with client:
         login_response = client.post(

backend/tests/component/api/test_03_menu.py

@@ -1,16 +1,19 @@
+from fastapi.testclient import TestClient
+
 from infrahub.core.branch import Branch
 from infrahub.core.initialization import create_default_menu
+from infrahub.core.node import Node
 from infrahub.core.schema import SchemaRoot
 from infrahub.database import InfrahubDatabase
 
 
 async def test_get_menu_not_admin(
     db: InfrahubDatabase,
-    client,
-    client_headers,
+    client: TestClient,
+    client_headers: dict[str, str],
     default_branch: Branch,
     car_person_schema_generics: SchemaRoot,
-    car_person_data_generic,
+    car_person_data_generic: dict[str, Node],
 ) -> None:
     await create_default_menu(db=db)
 
@@ -26,12 +29,12 @@ async def test_get_menu_not_admin(
 
 async def test_get_menu_admin(
     db: InfrahubDatabase,
-    client,
-    admin_headers,
-    authentication_base,
+    client: TestClient,
+    admin_headers: dict[str, str],
+    authentication_base: Node,
     default_branch: Branch,
     car_person_schema_generics: SchemaRoot,
-    car_person_data_generic,
+    car_person_data_generic: dict[str, Node],
 ) -> None:
     await create_default_menu(db=db)