Add get_aws_accounts typed query (app-sre#3470)

* Add AWSAccounts gql definition
* Add get_aws_accounts typed query
* Update aws_accounts gql and typed_query to fit terraform_tgw_attachments

Signed-off-by: Di Wang <[email protected]>

reconcile/gql_definitions/fragments/aws_account_common.gql

@@ -0,0 +1,32 @@
+# qenerate: plugin=pydantic_v1
+
+fragment AWSAccountCommon on AWSAccount_v1 {
+  path
+  name
+  uid
+  terraformUsername
+  consoleUrl
+  resourcesDefaultRegion
+  supportedDeploymentRegions
+  providerVersion
+  accountOwners {
+    name
+    email
+  }
+  automationToken {
+    ... VaultSecret
+  }
+  garbageCollection
+  enableDeletion
+  deletionApprovals {
+    type
+    name
+    expiration
+  }
+  disable {
+    integrations
+  }
+  deleteKeys
+  premiumSupport
+  partition
+}

reconcile/gql_definitions/fragments/aws_account_common.py

@@ -0,0 +1,65 @@
+"""
+Generated by qenerate plugin=pydantic_v1. DO NOT MODIFY MANUALLY!
+"""
+from collections.abc import Callable  # noqa: F401 # pylint: disable=W0611
+from datetime import datetime  # noqa: F401 # pylint: disable=W0611
+from enum import Enum  # noqa: F401 # pylint: disable=W0611
+from typing import (  # noqa: F401 # pylint: disable=W0611
+    Any,
+    Optional,
+    Union,
+)
+
+from pydantic import (  # noqa: F401 # pylint: disable=W0611
+    BaseModel,
+    Extra,
+    Field,
+    Json,
+)
+
+from reconcile.gql_definitions.fragments.vault_secret import VaultSecret
+
+
+class ConfiguredBaseModel(BaseModel):
+    class Config:
+        smart_union = True
+        extra = Extra.forbid
+
+
+class OwnerV1(ConfiguredBaseModel):
+    name: str = Field(..., alias="name")
+    email: str = Field(..., alias="email")
+
+
+class DeletionApprovalV1(ConfiguredBaseModel):
+    q_type: str = Field(..., alias="type")
+    name: str = Field(..., alias="name")
+    expiration: str = Field(..., alias="expiration")
+
+
+class DisableClusterAutomationsV1(ConfiguredBaseModel):
+    integrations: Optional[list[str]] = Field(..., alias="integrations")
+
+
+class AWSAccountCommon(ConfiguredBaseModel):
+    path: str = Field(..., alias="path")
+    name: str = Field(..., alias="name")
+    uid: str = Field(..., alias="uid")
+    terraform_username: Optional[str] = Field(..., alias="terraformUsername")
+    console_url: str = Field(..., alias="consoleUrl")
+    resources_default_region: str = Field(..., alias="resourcesDefaultRegion")
+    supported_deployment_regions: Optional[list[str]] = Field(
+        ..., alias="supportedDeploymentRegions"
+    )
+    provider_version: str = Field(..., alias="providerVersion")
+    account_owners: list[OwnerV1] = Field(..., alias="accountOwners")
+    automation_token: VaultSecret = Field(..., alias="automationToken")
+    garbage_collection: Optional[bool] = Field(..., alias="garbageCollection")
+    enable_deletion: Optional[bool] = Field(..., alias="enableDeletion")
+    deletion_approvals: Optional[list[DeletionApprovalV1]] = Field(
+        ..., alias="deletionApprovals"
+    )
+    disable: Optional[DisableClusterAutomationsV1] = Field(..., alias="disable")
+    delete_keys: Optional[list[str]] = Field(..., alias="deleteKeys")
+    premium_support: bool = Field(..., alias="premiumSupport")
+    partition: Optional[str] = Field(..., alias="partition")

reconcile/gql_definitions/fragments/terraform_state.gql

@@ -0,0 +1,11 @@
+# qenerate: plugin=pydantic_v1
+
+fragment TerraformState on TerraformStateAWS_v1 {
+  provider
+  bucket
+  region
+  integrations {
+    key
+    integration
+  }
+}

reconcile/gql_definitions/fragments/terraform_state.py

@@ -0,0 +1,38 @@
+"""
+Generated by qenerate plugin=pydantic_v1. DO NOT MODIFY MANUALLY!
+"""
+from collections.abc import Callable  # noqa: F401 # pylint: disable=W0611
+from datetime import datetime  # noqa: F401 # pylint: disable=W0611
+from enum import Enum  # noqa: F401 # pylint: disable=W0611
+from typing import (  # noqa: F401 # pylint: disable=W0611
+    Any,
+    Optional,
+    Union,
+)
+
+from pydantic import (  # noqa: F401 # pylint: disable=W0611
+    BaseModel,
+    Extra,
+    Field,
+    Json,
+)
+
+
+class ConfiguredBaseModel(BaseModel):
+    class Config:
+        smart_union = True
+        extra = Extra.forbid
+
+
+class AWSTerraformStateIntegrationsV1(ConfiguredBaseModel):
+    key: str = Field(..., alias="key")
+    integration: str = Field(..., alias="integration")
+
+
+class TerraformState(ConfiguredBaseModel):
+    provider: str = Field(..., alias="provider")
+    bucket: str = Field(..., alias="bucket")
+    region: str = Field(..., alias="region")
+    integrations: list[AWSTerraformStateIntegrationsV1] = Field(
+        ..., alias="integrations"
+    )

reconcile/gql_definitions/terraform_tgw_attachments/aws_accounts.gql

@@ -0,0 +1,14 @@
+# qenerate: plugin=pydantic_v1
+
+query AWSAccounts($name: String) {
+  accounts: awsaccounts_v1
+  (
+    name: $name
+  )
+  {
+    ... AWSAccountCommon
+    terraformState {
+      ... TerraformState
+    }
+  }
+}

reconcile/gql_definitions/terraform_tgw_attachments/aws_accounts.py

@@ -0,0 +1,118 @@
+"""
+Generated by qenerate plugin=pydantic_v1. DO NOT MODIFY MANUALLY!
+"""
+from collections.abc import Callable  # noqa: F401 # pylint: disable=W0611
+from datetime import datetime  # noqa: F401 # pylint: disable=W0611
+from enum import Enum  # noqa: F401 # pylint: disable=W0611
+from typing import (  # noqa: F401 # pylint: disable=W0611
+    Any,
+    Optional,
+    Union,
+)
+
+from pydantic import (  # noqa: F401 # pylint: disable=W0611
+    BaseModel,
+    Extra,
+    Field,
+    Json,
+)
+
+from reconcile.gql_definitions.fragments.aws_account_common import AWSAccountCommon
+from reconcile.gql_definitions.fragments.terraform_state import TerraformState
+
+
+DEFINITION = """
+fragment AWSAccountCommon on AWSAccount_v1 {
+  path
+  name
+  uid
+  terraformUsername
+  consoleUrl
+  resourcesDefaultRegion
+  supportedDeploymentRegions
+  providerVersion
+  accountOwners {
+    name
+    email
+  }
+  automationToken {
+    ... VaultSecret
+  }
+  garbageCollection
+  enableDeletion
+  deletionApprovals {
+    type
+    name
+    expiration
+  }
+  disable {
+    integrations
+  }
+  deleteKeys
+  premiumSupport
+  partition
+}
+
+fragment TerraformState on TerraformStateAWS_v1 {
+  provider
+  bucket
+  region
+  integrations {
+    key
+    integration
+  }
+}
+
+fragment VaultSecret on VaultSecret_v1 {
+    path
+    field
+    version
+    format
+}
+
+query AWSAccounts($name: String) {
+  accounts: awsaccounts_v1
+  (
+    name: $name
+  )
+  {
+    ... AWSAccountCommon
+    terraformState {
+      ... TerraformState
+    }
+  }
+}
+"""
+
+
+class ConfiguredBaseModel(BaseModel):
+    class Config:
+        smart_union = True
+        extra = Extra.forbid
+
+
+class AWSAccountV1(AWSAccountCommon):
+    terraform_state: Optional[TerraformState] = Field(..., alias="terraformState")
+
+
+class AWSAccountsQueryData(ConfiguredBaseModel):
+    accounts: Optional[list[AWSAccountV1]] = Field(..., alias="accounts")
+
+
+def query(query_func: Callable, **kwargs: Any) -> AWSAccountsQueryData:
+    """
+    This is a convenience function which queries and parses the data into
+    concrete types. It should be compatible with most GQL clients.
+    You do not have to use it to consume the generated data classes.
+    Alternatively, you can also mime and alternate the behavior
+    of this function in the caller.
+
+    Parameters:
+        query_func (Callable): Function which queries your GQL Server
+        kwargs: optional arguments that will be passed to the query function
+
+    Returns:
+        AWSAccountsQueryData: queried data parsed into generated classes
+    """
+    raw_data: dict[Any, Any] = query_func(DEFINITION, **kwargs)
+    return AWSAccountsQueryData(**raw_data)

reconcile/test/test_typed_queries/terraform_tgw_attachments/test_get_aws_accounts.py

@@ -0,0 +1,101 @@
+from collections.abc import Callable
+from unittest.mock import ANY
+
+import pytest
+
+from reconcile.gql_definitions.terraform_tgw_attachments.aws_accounts import (
+    AWSAccountsQueryData,
+    AWSAccountV1,
+)
+from reconcile.typed_queries.terraform_tgw_attachments.aws_accounts import (
+    get_aws_accounts,
+)
+from reconcile.utils.gql import GqlApi
+
+
+@pytest.fixture
+def aws_account(
+    gql_class_factory: Callable[..., AWSAccountV1],
+) -> AWSAccountV1:
+    return gql_class_factory(
+        AWSAccountV1,
+        {
+            "name": "some-account",
+            "uid": "some-uid",
+            "accountOwners": [],
+            "automationToken": {},
+            "premiumSupport": False,
+        },
+    )
+
+
+@pytest.fixture
+def aws_accounts(
+    gql_class_factory: Callable[..., AWSAccountsQueryData],
+    aws_account: AWSAccountV1,
+) -> AWSAccountsQueryData:
+    return gql_class_factory(
+        AWSAccountsQueryData,
+        {
+            "accounts": [
+                aws_account.dict(by_alias=True),
+            ]
+        },
+    )
+
+
+@pytest.fixture
+def aws_accounts_with_no_data(
+    gql_class_factory: Callable[..., AWSAccountsQueryData],
+) -> AWSAccountsQueryData:
+    return gql_class_factory(
+        AWSAccountsQueryData,
+        {
+            "accounts": None,
+        },
+    )
+
+
+def test_get_aws_accounts_with_default_variables(
+    gql_api_builder: Callable[..., GqlApi],
+    aws_accounts: AWSAccountsQueryData,
+) -> None:
+    gql_api = gql_api_builder(aws_accounts.dict(by_alias=True))
+
+    result = get_aws_accounts(gql_api)
+
+    assert result == aws_accounts.accounts
+    expected_variables = {
+        "name": None,
+    }
+    gql_api.query.assert_called_once_with(ANY, variables=expected_variables)
+
+
+def test_get_aws_accounts_when_no_data(
+    gql_api_builder: Callable[..., GqlApi],
+    aws_accounts_with_no_data: AWSAccountsQueryData,
+) -> None:
+    gql_api = gql_api_builder(aws_accounts_with_no_data.dict(by_alias=True))
+
+    result = get_aws_accounts(gql_api)
+
+    assert result == []
+
+
+def test_get_aws_accounts_with_name(
+    gql_api_builder: Callable[..., GqlApi],
+    aws_accounts: AWSAccountsQueryData,
+) -> None:
+    gql_api = gql_api_builder(aws_accounts.dict(by_alias=True))
+    name = "some-name"
+
+    result = get_aws_accounts(
+        gql_api,
+        name=name,
+    )
+
+    assert result == aws_accounts.accounts
+    expected_variables = {
+        "name": name,
+    }
+    gql_api.query.assert_called_once_with(ANY, variables=expected_variables)