Testing

Section 18 · From docs/INVENTORY_INTERNALS.md on GitHub

18.1 Layout

File	Covers
tests/test_inventory.py	Mapper, linker, dependency analysis, exports, organization discovery (moto).
tests/test_inventory_deep.py	Deep AWS collectors: containers, serverless, security, identity, Kubernetes mapping.
tests/test_aws_application.py, test_aws_data_ml.py, test_aws_governance.py, test_aws_network_ext.py	The newer AWS packages, with moto or stubbed clients.
tests/test_catalogs_cloudcontrol.py	Catalog loading, overlays, Cloud Control selection and redaction.
tests/test_azure_inventory.py	Resource Graph rows → assets and relations, hierarchy, SDK fallback.
tests/test_gcp_inventory.py	Cloud Asset Inventory records → assets, extractors, IAM, firewall evaluation, hierarchy.

18.2 moto and aiobotocore

tests/conftest.py has three autouse fixtures that make moto work with the async collectors:

• patch_aiobotocore_for_moto replaces aiobotocore's convert_to_response_dict with a version that accepts moto's synchronous response bodies;
• skip_ddb_crc32_under_moto disables DynamoDB's CRC32 response check, which awaits a synchronous body under moto;
• isolate_cloudcontrol_cache points $CLOUDG_CACHE_DIR at a temporary directory and resets the in-process Cloud Control type cache, so tests never read or write the user's cache.

A typical deep collector test:

import asyncio

import boto3
from moto import mock_aws

from cloudg.inventory import AWSDeepInventoryCollector, RelationshipLinker


@mock_aws
def test_function_runs_as_role():
    iam = boto3.client("iam", region_name="us-east-1")
    role = iam.create_role(RoleName="r", AssumeRolePolicyDocument="{}")["Role"]
    ...  # create the function with that role
    collector = AWSDeepInventoryCollector(
        boto3.Session(region_name="us-east-1"),
        "us-east-1",
        "123456789012",
        tagging_sweep=False,
        services=["identity", "serverless"],
        cloud_control=False,
    )
    assets = asyncio.run(collector.collect())
    edges = RelationshipLinker(assets).link()
    ...  # assert on the ASSUMES_ROLE edge

Narrow services and disable cloud_control / tagging_sweep to keep tests fast and focused. For services moto does not implement, patch _client with a stub that returns recorded responses.

18.3 Azure and GCP

No live cloud is needed: Azure tests feed Resource Graph rows (dicts shaped like ARM resources) into build_assets() or inject a graph_client_factory; GCP tests inject a fake AssetServiceClient through the client option returning recorded list_assets / search_all_iam_policies pages.

The API-backed sources get a GCPRestClient(session=FakeSession(routes)) through the rest_client option (tests/gcp_fakes.py): routes map a URL to a JSON body, an error response or a list of pages. tests/test_gcp_parity_extractors.py asserts one linked edge per new extractor; tests/test_gcp_sources.py covers every source, the coverage records, service filtering and the REST client.

The Azure depth tests share tests/azure_fakes.py:

• TypedFakeGraph(tables) answers Resource Graph queries by table and applies the type =~, type in~ and subscriptionId =~ clauses, as ARG does, so one table can hold rows for several queries. fail={...} makes tables raise.
• FakeArm(routes, forbidden=...) answers ARM REST, Microsoft Graph and data-plane calls by path, with 404 for unknown paths and 403 for paths ending in a forbidden suffix.
• collect_linked(tables, arm_routes, dataplane=..., **options) runs AzureDeepInventoryCollector against both and returns the linked _Graph and the coverage records.

test_azure_parity_graph.py covers the Resource Graph tables, coverage and filtering, test_azure_parity_rest.py the REST children, test_azure_parity_extractors.py the row extractors through AzureAssetBuilder, test_azure_entra.py Microsoft Graph, test_azure_kubernetes.py in-cluster AKS and test_azure_rest.py the REST client.

18.4 Capturing real shapes

INVENTORY_CATALOG.md was produced by running the suite with a small pytest plugin that wraps CloudAsset.__init__ (recording type, provider, metadata keys and value types, declared relations) and RelationshipLinker._add (recording the source type, edge type, relationship and target type of each linked edge), plus an end-to-end map of a simulated Organization. Rerun the same procedure after large collector changes to refresh the catalog.

18.5 Before opening a pull request

ruff format --check .
ruff check .
pytest -q