Seeded accounts e2e data

Author: Robert Segal

pyproject.toml

@@ -113,7 +113,7 @@ per-file-ignores = [
   "mpt_api_client/http/mixins.py: WPS202 WPS204 WPS235",
   "mpt_api_client/resources/*: WPS215",
   "mpt_api_client/models/model.py: WPS215 WPS110",
-  "mpt_api_client/resources/accounts/*.py: WPS202 WPS215 WPS214 WPS235",
+  "mpt_api_client/resources/accounts/*.py: WPS202 WPS215 WPS214 WPS235 WPS453",
   "mpt_api_client/resources/billing/*.py: WPS202 WPS204 WPS214 WPS215",
   "mpt_api_client/resources/catalog/*.py: WPS110 WPS214 WPS215 WPS235",
   "mpt_api_client/resources/catalog/mixins.py: WPS110 WPS202 WPS214 WPS215 WPS235",
@@ -129,7 +129,8 @@ per-file-ignores = [
   "tests/e2e/accounts/*.py: WPS430 WPS202",
   "tests/e2e/catalog/*.py: WPS202 WPS421",
   "tests/e2e/catalog/items/*.py: WPS110 WPS202",
-  "tests/*: WPS432 WPS202"
+  "tests/*: WPS432 WPS202",
+  "seed/accounts/*.py: WPS453",
 ]
 
 [tool.ruff]

seed/accounts/accounts.py

@@ -0,0 +1,23 @@
+import logging
+
+from seed.accounts.api_tokens import seed_api_token
+from seed.accounts.buyer import seed_buyer
+from seed.accounts.licensee import seed_licensee
+from seed.accounts.module import seed_module
+from seed.accounts.seller import seed_seller
+from seed.accounts.user_group import seed_user_group
+
+logger = logging.getLogger(__name__)
+
+
+async def seed_accounts() -> None:  # noqa: WPS217
+    """Seed accounts data including account."""
+    logger.debug("Seeding accounts ...")
+    await seed_seller()
+    await seed_buyer()
+    await seed_module()
+    await seed_api_token()
+    await seed_user_group()
+    await seed_licensee()
+
+    logger.debug("Seeded accounts completed.")

seed/accounts/api_tokens.py

@@ -0,0 +1,75 @@
+import logging
+import os
+
+from dependency_injector.wiring import inject
+
+from mpt_api_client import AsyncMPTClient
+from mpt_api_client.resources.accounts.api_tokens import ApiToken
+from seed.context import Context
+from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_OPERATIONS
+
+logger = logging.getLogger(__name__)
+
+
+@inject
+async def get_api_token(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_ops: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+) -> ApiToken | None:
+    """Get API token from context or fetch from API."""
+    api_token_id = context.get_string("accounts.api_token.id")
+    if not api_token_id:
+        return None
+    try:
+        api_token = context.get_resource("accounts.api_token", api_token_id)
+    except ValueError:
+        api_token = None
+    if not isinstance(api_token, ApiToken):
+        api_token = await mpt_ops.accounts.api_tokens.get(api_token_id)
+        context.set_resource("accounts.api_token", api_token)
+        context["accounts.api_token.id"] = api_token.id
+        return api_token
+    return api_token
+
+
+@inject
+def build_api_token_data(
+    context: Context = DEFAULT_CONTEXT,
+) -> dict[str, object]:
+    """Get API token data dictionary for creation."""
+    account_id = os.getenv("CLIENT_ACCOUNT_ID")
+    module_id = context.get_string("accounts.module.id")
+    return {
+        "account": {"id": account_id},
+        "name": "E2E Seeded API Token",
+        "description": "This is a seeded API token for end-to-end testing.",
+        "icon": "",
+        "modules": [{"id": module_id}],
+    }
+
+
+@inject
+async def init_api_token(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_ops: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+) -> ApiToken:
+    """Get or create API token."""
+    api_token = await get_api_token(context=context, mpt_ops=mpt_ops)
+    if api_token is None:
+        logger.debug("Creating API token ...")
+        api_token_data = build_api_token_data(context=context)
+        api_token = await mpt_ops.accounts.api_tokens.create(api_token_data)
+        context.set_resource("accounts.api_token", api_token)
+        context["accounts.api_token.id"] = api_token.id
+        logger.info("API token created: %s", api_token.id)
+    else:
+        logger.info("API token found: %s", api_token.id)
+    return api_token
+
+
+@inject
+async def seed_api_token() -> None:
+    """Seed API token."""
+    logger.debug("Seeding API token ...")
+    await init_api_token()
+    logger.debug("Seeding API token completed.")

seed/accounts/buyer.py

@@ -0,0 +1,93 @@
+import logging
+import os
+import pathlib
+
+from dependency_injector.wiring import inject
+
+from mpt_api_client import AsyncMPTClient
+from mpt_api_client.resources.accounts.buyers import Buyer
+from seed.context import Context
+from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_OPERATIONS
+
+logger = logging.getLogger(__name__)
+
+icon = pathlib.Path(__file__).parent / "logo.png"
+
+
+@inject
+async def get_buyer(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+) -> Buyer | None:
+    """Get buyer from context or fetch from API."""
+    buyer_id = context.get_string("accounts.buyer.id")
+    if not buyer_id:
+        return None
+    try:
+        buyer = context.get_resource("accounts.buyer", buyer_id)
+    except ValueError:
+        buyer = None
+    if not isinstance(buyer, Buyer):
+        buyer = await mpt_operations.accounts.buyers.get(buyer_id)
+        context.set_resource("accounts.buyer", buyer)
+        context["accounts.buyer.id"] = buyer.id
+        return buyer
+    return buyer
+
+
+@inject
+def build_buyer_data(context: Context = DEFAULT_CONTEXT) -> dict[str, object]:
+    """Build buyer data dictionary for creation."""
+    buyer_account_id = os.getenv("CLIENT_ACCOUNT_ID")
+    if not buyer_account_id:
+        raise ValueError("CLIENT_ACCOUNT_ID environment variable is required")
+    seller_id = context.get_string("accounts.seller.id")
+    return {
+        "name": "E2E Seeded Buyer",
+        "account": {"id": buyer_account_id},
+        "sellers": [{"id": seller_id}],
+        "contact": {
+            "firstName": "first",
+            "lastName": "last",
+            "email": "created.buyer@example.com",
+        },
+        "address": {
+            "addressLine1": "123 Main St",
+            "city": "Los Angeles",
+            "state": "CA",
+            "postCode": "12345",
+            "country": "US",
+        },
+    }
+
+
+@inject
+async def init_buyer(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+) -> Buyer:
+    """Get or create buyer."""
+    buyer = await get_buyer(context=context, mpt_operations=mpt_operations)
+    if buyer is None:
+        buyer_data = build_buyer_data(context=context)
+        logger.debug("Creating buyer ...")
+        with pathlib.Path.open(icon, "rb") as icon_file:
+            created = await mpt_operations.accounts.buyers.create(buyer_data, file=icon_file)
+        if isinstance(created, Buyer):
+            buyer = await mpt_operations.accounts.buyers.get(created.id)
+            context.set_resource("accounts.buyer", buyer)
+            context["accounts.buyer.id"] = buyer.id
+            logger.info("Buyer created: %s", buyer.id)
+            return buyer
+        logger.warning("Buyer creation failed")
+        raise ValueError("Buyer creation failed")
+    logger.info("Buyer found: %s", buyer.id)
+    return buyer
+
+
+@inject
+async def seed_buyer() -> None:
+    """Seed buyer."""
+    logger.debug("Seeding buyer ...")
+    await init_buyer()
+    logger.debug("Seeding buyer completed.")

seed/accounts/licensee.py

@@ -0,0 +1,97 @@
+import logging
+import os
+import pathlib
+
+from dependency_injector.wiring import inject
+
+from mpt_api_client import AsyncMPTClient
+from mpt_api_client.resources.accounts.licensees import Licensee
+from seed.context import Context
+from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_CLIENT
+
+logger = logging.getLogger(__name__)
+
+icon = pathlib.Path(__file__).parent / "logo.png"
+
+
+@inject
+async def get_licensee(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_client: AsyncMPTClient = DEFAULT_MPT_CLIENT,
+) -> Licensee | None:
+    """Get licensee from context or fetch from API."""
+    licensee_id = context.get_string("accounts.licensee.id")
+    if not licensee_id:
+        return None
+    try:
+        licensee = context.get_resource("accounts.licensee", licensee_id)
+    except ValueError:
+        licensee = None
+    if not isinstance(licensee, Licensee):
+        licensee = await mpt_client.accounts.licensees.get(licensee_id)
+        context.set_resource("accounts.licensee", licensee)
+        context["accounts.licensee.id"] = licensee.id
+        return licensee
+    return licensee
+
+
+@inject
+def build_licensee_data(
+    context: Context = DEFAULT_CONTEXT,
+) -> dict[str, object]:
+    """Get licensee data dictionary for creation."""
+    account_id = os.getenv("CLIENT_ACCOUNT_ID")
+    seller_id = context.get_string("accounts.seller.id")
+    buyer_id = context.get_string("accounts.buyer.id")
+    group = context.get_resource("accounts.user_group")
+    licensee_type = "Client"
+    return {
+        "name": "E2E Seeded Licensee",
+        "address": {
+            "addressLine1": "123 Main St",
+            "city": "Los Angeles",
+            "state": "CA",
+            "postCode": "67890",
+            "country": "US",
+        },
+        "useBuyerAddress": False,
+        "seller": {"id": seller_id},
+        "buyer": {"id": buyer_id},
+        "account": {"id": account_id},
+        "eligibility": {"client": True, "partner": False},
+        "groups": [{"id": group.id}],
+        "type": licensee_type,
+        "status": "Enabled",
+        "defaultLanguage": "en-US",
+    }
+
+
+@inject
+async def init_licensee(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_client: AsyncMPTClient = DEFAULT_MPT_CLIENT,
+) -> Licensee:
+    """Get or create licensee."""
+    licensee = await get_licensee(context=context, mpt_client=mpt_client)
+    if licensee is None:
+        licensee_data = build_licensee_data(context=context)
+        logger.debug("Creating licensee ...")
+        with pathlib.Path.open(icon, "rb") as icon_file:
+            created = await mpt_client.accounts.licensees.create(licensee_data, file=icon_file)
+        if isinstance(created, Licensee):
+            context.set_resource("accounts.licensee", created)
+            context["accounts.licensee.id"] = created.id
+            logger.info("Licensee created: %s", created.id)
+            return created
+        logger.warning("Licensee creation failed")
+        raise ValueError("Licensee creation failed")
+    logger.info("Licensee found: %s", licensee.id)
+    return licensee
+
+
+@inject
+async def seed_licensee() -> None:
+    """Seed licensee."""
+    logger.debug("Seeding licensee ...")
+    await init_licensee()
+    logger.info("Seeding licensee completed.")

seed/accounts/logo.png

@@ -0,0 +1,72 @@
+import logging
+
+from dependency_injector.wiring import inject
+
+from mpt_api_client import AsyncMPTClient
+from mpt_api_client.resources.accounts.modules import Module
+from mpt_api_client.rql.query_builder import RQLQuery
+from seed.context import Context
+from seed.defaults import DEFAULT_CONTEXT, DEFAULT_MPT_OPERATIONS
+
+logger = logging.getLogger(__name__)
+
+
+@inject
+async def get_module(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+) -> Module | None:
+    """Get module from context or fetch from API."""
+    module_id = context.get_string("accounts.module.id")
+    if not module_id:
+        return None
+    try:
+        module = context.get_resource("accounts.module", module_id)
+    except ValueError:
+        module = None
+    if not isinstance(module, Module):
+        module = await mpt_operations.accounts.modules.get(module_id)
+        context.set_resource("accounts.module", module)
+        context["accounts.module.id"] = module.id
+        return module
+    return module
+
+
+@inject
+async def refresh_module(
+    context: Context = DEFAULT_CONTEXT,
+    mpt_operations: AsyncMPTClient = DEFAULT_MPT_OPERATIONS,
+) -> Module | None:
+    """Refresh module in context (always fetch)."""
+    module = await get_module(context=context, mpt_operations=mpt_operations)
+    if module is None:
+        filtered_modules = mpt_operations.accounts.modules.filter(
+            RQLQuery(name="Access Management")
+        )
+        modules = [mod async for mod in filtered_modules.iterate()]
+        if modules:
+            first_module = modules[0]
+            if isinstance(first_module, Module):
+                context["accounts.module.id"] = first_module.id
+                context.set_resource("accounts.module", first_module)
+                return first_module
+            logger.warning("First module is not a Module instance.")
+            return None
+        logger.warning("Module 'Access Management' not found.")
+        return None
+    return module
+
+
+@inject
+async def seed_module() -> Module:
+    """Seed module."""
+    logger.debug("Seeding module ...")
+    existing_module = await get_module()
+    if existing_module is None:
+        refreshed = await refresh_module()
+        logger.debug("Seeding module completed.")
+        if refreshed is None:
+            raise ValueError("Could not seed module: no valid Module found.")
+        return refreshed
+    logger.debug("Seeding module completed.")
+    return existing_module