Running test with mocked pulumi and getting this error AttributeError: 'NoneType' object has no attribute 'get'

[zhukovgreen]

Hi, dear community! I have already spent a couple of hours and cannot resolve the following issue.

My iac.py code is:

import sys

from typing import cast

import pulumi

from environs import Env
from loguru import logger
from pulumi_aws import ec2, glue


pulumi_config = pulumi.Config()
env = Env(expand_vars=True)
env.read_env()

logger.add(
    sys.stderr,
    colorize=True,
    level="DEBUG" if env.bool("DEBUG", False) else "INFO",
)

# required pulumi configuration
ENV = pulumi_config.require("environment")
ACCOUNT = pulumi_config.require("account")
JDBC_CONNECTION_URL = pulumi_config.require("jdbc_connection_url")
JDBC_CONNECTION_PASSWORD = pulumi_config.require_secret(
    "jdbc_connection_password"
)
JDBC_CONNECTION_USERNAME = pulumi_config.require_secret(
    "jdbc_connection_username"
)
SUBNET_TAG = pulumi_config.require("subnet_tag_name")

# other constants
PROJECT_SLUG = "de-reportingstore-migration"
RESOURCE_PREFIX = f"{ENV}-{PROJECT_SLUG}"

# get resources from the dataplatform-network stack
network_stack = pulumi.StackReference(f"{ACCOUNT}-dataplatform-network")
glue_security_group_id = network_stack.get_output("GlueCrawlerSecurityGroup")
private_subnet_a_id = network_stack.require_output("PrivateSubnetAId")


def get_availability_zone(private_subnet_id: str) -> str:
    # TODO check if there is more straightforward way to get the
    #   availability_zone
    return cast(
        str,
        ec2.get_subnet(
            id=private_subnet_id,
            filters=[
                ec2.GetSubnetFilterArgs(
                    name="tag:Name",
                    values=[SUBNET_TAG],
                )
            ],
        ).availability_zone,
    )


def get_glue_connection(args: list[pulumi.Output[str]]) -> glue.Connection:
    availability_zone, subnet_id, security_group_id = args
    return glue.Connection(
        RESOURCE_PREFIX,
        connection_properties={
            "JDBC_CONNECTION_URL": JDBC_CONNECTION_URL,
            "JDBC_ENFORCE_SSL": "false",
            "PASSWORD": JDBC_CONNECTION_PASSWORD,
            "USERNAME": JDBC_CONNECTION_USERNAME,
        },
        physical_connection_requirements=(
            glue.ConnectionPhysicalConnectionRequirementsArgs(
                subnet_id=subnet_id,
                # TODO create own security group to satisfy the needs of only this
                #   repo
                security_group_id_lists=[security_group_id],
                availability_zone=availability_zone,
            )
        ),
    )


availability_zone = private_subnet_a_id.apply(get_availability_zone)

connection: pulumi.Output[glue.Connection] = pulumi.Output.all(
    availability_zone,
    private_subnet_a_id,
    glue_security_group_id,
).apply(get_glue_connection)

pulumi.export("connection_urn", connection.urn)

my test_iac.py code:

import pulumi
import pytest


class MockPulumiResources(pulumi.runtime.Mocks):
    def new_resource(self, args: pulumi.runtime.MockResourceArgs):
        outputs = args.inputs
        if args.typ == "pulumi:pulumi:StackReference":
            outputs = {
                **args.inputs,
                "GlueCrawlerSecurityGroup": "sg-0888888888850db48",
                "PrivateSubnetAId": "subnet-01e888e888c888ee8",
            }
        return [args.name + "_id", outputs]

    def call(self, args: pulumi.runtime.MockCallArgs):
        if args.token == "aws:ec2/getAmi:getAmi":
            return {
                "architecture": "x86_64",
                "id": "ami-0eb1f3cdeeb8eed2a",
            }
        return {}


pulumi.runtime.set_mocks(mocks=MockPulumiResources())
from reportingstore_migration import iac


@pulumi.runtime.test
def test_should_just_work():
    def inner(args):
        print(args)

    return pulumi.Output.all(
        iac.connection.physical_connection_requirements
    ).apply(inner)

when running this test, I am getting the following error:

test_iac.py::test_should_just_work FAILED
tests/test_iac.py:24 (test_should_just_work)
Traceback (most recent call last):
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/runtime/mocks.py", line 54, in wrapper
    _sync_await(
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/runtime/sync_await.py", line 54, in _sync_await
    return loop.run_until_complete(fut)
  File "/Users/username/.pyenv/versions/3.10.6/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/runtime/stack.py", line 49, in run_pulumi_func
    func()
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/runtime/mocks.py", line 56, in <lambda>
    lambda: _sync_await(Output.from_input(fn(*args, **kwargs)).future())
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/runtime/sync_await.py", line 87, in _sync_await
    return fut.result()
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/output.py", line 125, in get_value
    val = await self._future
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/output.py", line 171, in run
    value = await self._future
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/output.py", line 171, in run
    value = await self._future
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/output.py", line 447, in gather_futures
    return await asyncio.gather(*value_futures_list)
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/output.py", line 125, in get_value
    val = await self._future
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/output.py", line 171, in run
    value = await self._future
  File "/Users/username/Documents/company/dataplatform.reportingstore-migration/.venv/lib/python3.10/site-packages/pulumi/output.py", line 171, in run
    value = await self._future
AttributeError: 'NoneType' object has no attribute 'get'

I see that the code is reaching the MockPulumiResources:new_resource, but after that, it is failing with the provided error. Any hints will be very appreciated.

[zhukovgreen]

I found out that it is somehow related to those lines:

glue_security_group_id = network_stack.get_output("GlueCrawlerSecurityGroup")
private_subnet_a_id = network_stack.require_output("PrivateSubnetAId")

When I replace their value with some Output.from_value("some string"), then the test progress further.

I also see that the new_resource from the MockPulumiResources is called, and the proper outputs returned, however it doesn't help

[zhukovgreen]

Created a ticket which seems to be a root cause of the error:
#11037