• s

    straight-restaurant-537

    1 month ago
    Hey guys, I'm new to Pulumi and Python, so please excuse my question, if it is very simplistic. I have a resource component which has a dict of pulumi objects created and I am asserting the outputs in a unit test. This code does work, but I was wondering if there was a better way to do it?
    @pulumi.runtime.test
      def test_policy_assignment_created(self):
          available_assignments = AvailableAssignments(assignments_json=self.object_mother.get_assignments_json())
          
          sut = PolicyAssignmentComponent(self.policy_assignments_model, available_assignments, self.opts)
      
          def check_policy_assignment(args):
              management_group_name, policy_assignment_objects = args
              
              assert management_group_name == self.policy_assignments_model.management_group_name
              assert len(policy_assignment_objects) == self.object_mother.get_assignments_json_count()
    
              def assert_policy_properties(args):
                id, enforcement_mode, location, policy_definition_id  = args
    
                expected_policy_assignment = self.object_mother.get_policy_assignment_by_policy_def_id(
                  policy_def_id=policy_definition_id
                )
                
                expected_id = f"{management_group_name}-{expected_policy_assignment.name}_id"
                
                assert expected_id == id
                assert enforcement_mode == expected_policy_assignment.enforcementMode
                assert location == expected_policy_assignment.location
                assert policy_definition_id == expected_policy_assignment.policy_definition_id
              
              for policy_assignment_object in policy_assignment_objects:
                pulumi.Output.all(
                  policy_assignment_object.id,
                  policy_assignment_object.enforcement_mode,
                  policy_assignment_object.location,
                  policy_assignment_object.policy_definition_id,
                ).apply(assert_policy_properties)
    
          return pulumi.Output.all(sut.management_group_name, sut.policy_assignment_objects).apply(check_policy_assignment)
  • w

    white-terabyte-21934

    1 month ago
    Hi , I am looking for a way to pulumi login via automation api in python , any sample ?
    w
    b
    2 replies
    Copy to Clipboard
  • k

    kind-france-51068

    1 month ago
    Hey everyone, this may or may not be a dumb question but I'm working on a project right now where I have to use pulumi to grab a list of all of the snapshots in a project and find the most recently created to build a new host instance. Seems pretty simple but I'm having a hard time with grabbing a list of snapshots. I'm using the "google-native" provider with the 'getSnapshot' function. The documentation says "Returns the specified Snapshot resource. Gets a list of available snapshots by making a list() request." and for the life of me I can't figure out how to send that list request. I'm also seeing the same documentation for 'getDisk' and 'getInstance' and others so I'm feeling like I'm missing something simple. Anyway, this is the documentation I'm talking about:https://www.pulumi.com/registry/packages/google-native/api-docs/compute/v1/getsnapshot/ any help would be appreciated.
  • s

    shy-bird-55689

    1 month ago
    Is there are a resource for AWS Client VPN Endpoints? Cant seem to find one, although i can see other vpn options
    s
    b
    2 replies
    Copy to Clipboard
  • f

    full-window-21515

    1 month ago
    How would I do pulumi config set "aws😒kipMetadataApiCheck" false In Python?
    f
    e
    2 replies
    Copy to Clipboard
  • b

    busy-branch-95201

    1 month ago
    Hi, I want to migrate from one resource (EBS csi driver via Helm chart) to another (eks addon for ebs csi driver). However those two conflict with each other so creating the addon before deleting the helm chart does not work. Is there anything I could do to force pulumi to first delete the resources of the helm chart and after that install the eks addon?
    b
    l
    2 replies
    Copy to Clipboard
  • w

    worried-gold-55244

    1 month ago
    Hi Guys, is it possible to catch invoke exceptions ? here is the code
    try:
        key_pair = compute.get_keypair_output(args.region + '_key',
                                              opts=pulumi.ResourceOptions(provider=provider, parent=self))
    except:
        key_pair = compute.Keypair(args.region + '_key',
                                   name=args.region + '_key',
                                   public_key=args.ssh_key_pub,
                                   opts=pulumi.ResourceOptions(provider=provider, parent=self))
    and error:
    error: Program failed with an unhandled exception:
        error: Traceback (most recent call last):
          File "/usr/local/bin/pulumi-language-python-exec", line 107, in <module>
            loop.run_until_complete(coro)
          File "/usr/local/Cellar/python@3.9/3.9.13_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 647, in run_until_complete
            return future.result()
          File "/Users/amuntean/Projects/new_netlog/venv/lib/python3.9/site-packages/pulumi/runtime/stack.py", line 126, in run_in_stack
            await run_pulumi_func(lambda: Stack(func))
          File "/Users/amuntean/Projects/new_netlog/venv/lib/python3.9/site-packages/pulumi/runtime/stack.py", line 51, in run_pulumi_func
            await wait_for_rpcs()
          File "/Users/amuntean/Projects/new_netlog/venv/lib/python3.9/site-packages/pulumi/runtime/stack.py", line 110, in wait_for_rpcs
            raise exception
          File "/Users/amuntean/Projects/new_netlog/venv/lib/python3.9/site-packages/pulumi/output.py", line 194, in run
            transformed: Input[U] = func(value)
          File "/Users/amuntean/Projects/new_netlog/venv/lib/python3.9/site-packages/pulumi_openstack/_utilities.py", line 232, in <lambda>
            }).apply(lambda resolved_args: func(*resolved_args['args'],
          File "/Users/amuntean/Projects/new_netlog/venv/lib/python3.9/site-packages/pulumi_openstack/compute/get_keypair.py", line 121, in get_keypair
            __ret__ = pulumi.runtime.invoke('openstack:compute/getKeypair:getKeypair', __args__, opts=opts, typ=GetKeypairResult).value
          File "/Users/amuntean/Projects/new_netlog/venv/lib/python3.9/site-packages/pulumi/runtime/invoke.py", line 144, in invoke
            raise invoke_error
        Exception: invoke of openstack:compute/getKeypair:getKeypair failed: invocation of openstack:compute/getKeypair:getKeypair returned an error: invoking openstack:compute/getKeypair:getKeypair: 1 error occurred:
            * Error retrieving openstack_compute_keypair_v2 iad-private-dev_key: Resource not found
        error: an unhandled error occurred: Program exited with non-zero exit code: 1
  • k

    kind-hamburger-15227

    1 month ago
    import pulumi
    import pulumi_aws as aws
    
    # Create an AWS provider for the us-east-1 region.
    useast1 = aws.Provider("useast1", region="us-east-1")
    
    # Create an ACM certificate in us-east-1.
    cert = aws.acm.Certificate("cert",
        domain_name="<http://foo.com|foo.com>",
        validation_method="EMAIL",
        __opts__=pulumi.ResourceOptions(provider=useast1))
    Trying to create my own provider, but it fails with
    File ~/.local/lib/python3.10/site-packages/pulumi/runtime/settings.py:202, in get_monitor()
        200 monitor = SETTINGS.monitor
        201 if not monitor:
    --> 202     require_test_mode_enabled()
        203 return monitor
    
    File ~/.local/lib/python3.10/site-packages/pulumi/runtime/settings.py:147, in require_test_mode_enabled()
        145 def require_test_mode_enabled():
        146     if not is_test_mode_enabled():
    --> 147         raise RunError(
        148             "Program run without the Pulumi engine available; re-run using the `pulumi` CLI"
        149         )
    
    RunError: Program run without the Pulumi engine available; re-run using the `pulumi` CLI
    The purpose of the provide to set config
    pulumi config set aws:skipCredentialsValidation true
    - I am trying to run flask api example for automation api: https://github.com/pulumi/automation-api-examples/blob/main/python/pulumi_over_http/app.py any hints? Original sample fails with
    Diagnostics:
      aws:s3:Bucket (s3-website-bucket):
        error: 1 error occurred:
        	* error configuring Terraform AWS Provider: AWS account ID not previously found and failed retrieving via all available methods. See <https://www.terraform.io/docs/providers/aws/index.html#skip_requesting_account_id> for workaround and implications. Errors: 2 errors occurred:
  • k

    kind-hamburger-15227

    1 month ago
    How to create default_target_group for load balancer in python?
    listener = awsx.lb.NetworkLoadBalancer("lb",default_target_group=['group',{ 'port' : 80, 'protocol':'TCP'}])
    I am trying to re-create typescript example in python
    const listener = new awsx.lb.NetworkLoadBalancer('lb')
      .createTargetGroup('group', { port: 80, protocol: 'TCP' })
      .createListener('listener', {
        port: 443,
        protocol: 'TLS',
        // ARN of the Amazon Cert Manager certificate for *.<http://mysite.com|mysite.com>
        certificateArn: certCertificate.arn,
      })
  • f

    few-tent-80031

    1 month ago
    Hi guys, my name is Ivan Indjic and I am working as a DevOps engineer in a Serbian based company. I'm learning Pulumi both for work and for my master thesis. I have one problem with my current stack. I created one rds instance and I want to use output of that resource ( rds address ) as an env var in my ecs container definition. So, I have this instance below.
    database = aws.rds.Instance(
        resource_name="master-thesis-rds",
        allocated_storage = 5,
        instance_class='db.t4g.micro',
        allow_major_version_upgrade=False,
        apply_immediately=False,
        auto_minor_version_upgrade=True,
        db_name='master',
        db_subnet_group_name='db-subnet-group',
        deletion_protection=False,
        enabled_cloudwatch_logs_exports=['audit', 'error', 'general', 'slowquery'],
        engine='mysql',
        performance_insights_enabled=True,
        port=5432,
        publicly_accessible=False,
        skip_final_snapshot=True,
        vpc_security_group_ids=[db_sg.id],
        username='',
        password=password.value
    )
    Then, I did this:
    rds_endpoint = database.address.apply(lambda address: f"http://{address}")
    And last, I have ecs container definition with these env vars:
    "environment": [
                    {
                        "name": "user", "value": ""
                    },
                    {
                        "name": "password", "value": f"{password.value}"
                    },
                    {
                        "name": "db_host_replica", "value": rds_endpoint
                    },
                    {
                        "name": "db_host", "value": rds_endpoint
                    }
                ],
    However, when I run this, I get
    TypeError: Object of type Output is not JSON serializable
    Do you guys know where is a mistake? I tried different ways but I cannot get right output converted to string. Thank you
    f
    b
    2 replies
    Copy to Clipboard