`Output.from_input` deeply unwraps nested output values in dicts and lists, but doesn't currently do that for the more recently added "input types" (i.e. args classes). This leads to errors when using args classes with output values with `Provider` resources, which uses `Output.from_input` on each input property and then serializes the value to JSON in an `apply`. This changes fixes `Output.from_input` to recurse into values within args classes to properly unwrap any nested outputs.
175 lines
6.4 KiB
Python
175 lines
6.4 KiB
Python
# Copyright 2016-2021, Pulumi Corporation.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import unittest
|
|
from typing import Mapping, Optional, Sequence, cast
|
|
|
|
from pulumi.runtime import rpc, rpc_manager, settings
|
|
from pulumi import Output
|
|
import pulumi
|
|
|
|
|
|
def pulumi_test(coro):
|
|
wrapped = pulumi.runtime.test(coro)
|
|
def wrapper(*args, **kwargs):
|
|
settings.configure(settings.Settings())
|
|
rpc._RESOURCE_PACKAGES.clear()
|
|
rpc._RESOURCE_MODULES.clear()
|
|
rpc_manager.RPC_MANAGER = rpc_manager.RPCManager()
|
|
|
|
wrapped(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
class OutputSecretTests(unittest.TestCase):
|
|
@pulumi_test
|
|
async def test_secret(self):
|
|
x = Output.secret("foo")
|
|
is_secret = await x.is_secret()
|
|
self.assertTrue(is_secret)
|
|
|
|
@pulumi_test
|
|
async def test_unsecret(self):
|
|
x = Output.secret("foo")
|
|
x_is_secret = await x.is_secret()
|
|
self.assertTrue(x_is_secret)
|
|
|
|
y = Output.unsecret(x)
|
|
y_val = await y.future()
|
|
y_is_secret = await y.is_secret()
|
|
self.assertEqual(y_val, "foo")
|
|
self.assertFalse(y_is_secret)
|
|
|
|
|
|
class OutputFromInputTests(unittest.TestCase):
|
|
@pulumi_test
|
|
async def test_unwrap_dict(self):
|
|
x = Output.from_input({"hello": Output.from_input("world")})
|
|
x_val = await x.future()
|
|
self.assertEqual(x_val, {"hello": "world"})
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_dict_secret(self):
|
|
x = Output.from_input({"hello": Output.secret("world")})
|
|
x_val = await x.future()
|
|
self.assertEqual(x_val, {"hello": "world"})
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_dict_dict(self):
|
|
x = Output.from_input({"hello": {"foo": Output.from_input("bar")}})
|
|
x_val = await x.future()
|
|
self.assertEqual(x_val, {"hello": {"foo": "bar"}})
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_dict_list(self):
|
|
x = Output.from_input({"hello": ["foo", Output.from_input("bar")]})
|
|
x_val = await x.future()
|
|
self.assertEqual(x_val, {"hello": ["foo", "bar"]})
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_list(self):
|
|
x = Output.from_input(["hello", Output.from_input("world")])
|
|
x_val = await x.future()
|
|
self.assertEqual(x_val, ["hello", "world"])
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_list_list(self):
|
|
x = Output.from_input(["hello", ["foo", Output.from_input("bar")]])
|
|
x_val = await x.future()
|
|
self.assertEqual(x_val, ["hello", ["foo", "bar"]])
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_list_dict(self):
|
|
x = Output.from_input(["hello", {"foo": Output.from_input("bar")}])
|
|
x_val = await x.future()
|
|
self.assertEqual(x_val, ["hello", {"foo": "bar"}])
|
|
|
|
@pulumi.input_type
|
|
class FooArgs:
|
|
def __init__(self, *,
|
|
foo: Optional[pulumi.Input[str]] = None,
|
|
bar: Optional[pulumi.Input[Sequence[pulumi.Input[str]]]] = None,
|
|
baz: Optional[pulumi.Input[Mapping[str, pulumi.Input[str]]]] = None,
|
|
nested: Optional[pulumi.Input[pulumi.InputType['NestedArgs']]] = None):
|
|
if foo is not None:
|
|
pulumi.set(self, "foo", foo)
|
|
if bar is not None:
|
|
pulumi.set(self, "bar", bar)
|
|
if baz is not None:
|
|
pulumi.set(self, "baz", baz)
|
|
if nested is not None:
|
|
pulumi.set(self, "nested", nested)
|
|
|
|
@property
|
|
@pulumi.getter
|
|
def foo(self) -> Optional[pulumi.Input[str]]:
|
|
return pulumi.get(self, "foo")
|
|
|
|
@property
|
|
@pulumi.getter
|
|
def bar(self) -> Optional[pulumi.Input[Sequence[pulumi.Input[str]]]]:
|
|
return pulumi.get(self, "bar")
|
|
|
|
@property
|
|
@pulumi.getter
|
|
def baz(self) -> Optional[pulumi.Input[Mapping[str, pulumi.Input[str]]]]:
|
|
return pulumi.get(self, "baz")
|
|
|
|
@property
|
|
@pulumi.getter
|
|
def nested(self) -> Optional[pulumi.Input[pulumi.InputType['NestedArgs']]]:
|
|
return pulumi.get(self, "nested")
|
|
|
|
@pulumi.input_type
|
|
class NestedArgs:
|
|
def __init__(self, *,
|
|
hello: Optional[pulumi.Input[str]] = None):
|
|
if hello is not None:
|
|
pulumi.set(self, "hello", hello)
|
|
|
|
@property
|
|
@pulumi.getter
|
|
def hello(self) -> Optional[pulumi.Input[str]]:
|
|
return pulumi.get(self, "hello")
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_input_type(self):
|
|
x = Output.from_input(OutputFromInputTests.FooArgs(foo=Output.from_input("bar")))
|
|
x_val = cast(OutputFromInputTests.FooArgs, await x.future())
|
|
self.assertIsInstance(x_val, OutputFromInputTests.FooArgs)
|
|
self.assertEqual(x_val.foo, "bar")
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_input_type_list(self):
|
|
x = Output.from_input(OutputFromInputTests.FooArgs(bar=["a", Output.from_input("b")]))
|
|
x_val = cast(OutputFromInputTests.FooArgs, await x.future())
|
|
self.assertIsInstance(x_val, OutputFromInputTests.FooArgs)
|
|
self.assertEqual(x_val.bar, ["a", "b"])
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_input_type_dict(self):
|
|
x = Output.from_input(OutputFromInputTests.FooArgs(baz={"hello": Output.from_input("world")}))
|
|
x_val = cast(OutputFromInputTests.FooArgs, await x.future())
|
|
self.assertIsInstance(x_val, OutputFromInputTests.FooArgs)
|
|
self.assertEqual(x_val.baz, {"hello": "world"})
|
|
|
|
@pulumi_test
|
|
async def test_unwrap_input_type_nested(self):
|
|
nested = OutputFromInputTests.NestedArgs(hello=Output.from_input("world"))
|
|
x = Output.from_input(OutputFromInputTests.FooArgs(nested=nested))
|
|
x_val = cast(OutputFromInputTests.FooArgs, await x.future())
|
|
self.assertIsInstance(x_val, OutputFromInputTests.FooArgs)
|
|
self.assertIsInstance(x_val.nested, OutputFromInputTests.NestedArgs)
|
|
self.assertEqual(x_val.nested.hello, "world")
|