Remote component py SDK (#6715)

* Python support for authoring resource providers for multi-lang * Support for passing prompt values to Python resource providers
2021-04-15 14:49:51 -04:00 · 2021-04-15 14:49:51 -04:00 · b77f32930c
parent 4dcc0d631e
commit b77f32930c
29 changed files with 973 additions and 149 deletions
--- a/CHANGELOG_PENDING.md
+++ b/CHANGELOG_PENDING.md
@ -41,12 +41,15 @@
 - [CLI] Remove `pulumi history` command. This was previously deprecated and replaced by `pulumi stack history`
  [#6724](https://github.com/pulumi/pulumi/pull/6724)

+- [sdk/python] Allow using Python to build resource providers for multi-lang components.
+  [#6715](https://github.com/pulumi/pulumi/pull/6715)
+
 ### Enhancements

 - [sdk/nodejs] Add support for multiple V8 VM contexts in closure serialization.
  [#6648](https://github.com/pulumi/pulumi/pull/6648)
-  
+
 - [sdk] Handle providers for RegisterResourceRequest
  [#6771](https://github.com/pulumi/pulumi/pull/6771)

-### Bug Fixes
+### Bug Fixes
--- a/build.proj
+++ b/build.proj
@ -67,7 +67,7 @@
          WorkingDirectory="$(DotNetSdkDirectory)" />
    <Exec Command="go install -ldflags &quot;-X github.com/pulumi/pulumi/sdk/v3/go/common/version.Version=$(Version)&quot; github.com/pulumi/pulumi/sdk/v3/dotnet/cmd/pulumi-language-dotnet"
          WorkingDirectory="$(SdkDirectory)"/>
-    
+
  </Target>

  <Target Name="DotNetInstallPlugin">
@ -77,7 +77,7 @@
    <Exec Command="go install -ldflags &quot;-X github.com/pulumi/pulumi/sdk/v3/go/common/version.Version=$(Version)&quot; github.com/pulumi/pulumi/sdk/v3/dotnet/cmd/pulumi-language-dotnet"
      EnvironmentVariables="GOBIN=$(PulumiBin)"
      WorkingDirectory="$(SdkDirectory)"/>
-    
+
  </Target>

  <Target Name="CopyNugetPackages">
@ -100,7 +100,7 @@
    </Exec>
    <Exec Command="go install -ldflags &quot;-X github.com/pulumi/pulumi/sdk/v3/go/common/version.Version=$(Version)&quot; github.com/pulumi/pulumi/sdk/v3/go/pulumi-language-go"
        WorkingDirectory="$(SdkDirectory)"/>
-    
+
  </Target>

  <Target Name="GoInstallPlugin">
@ -110,7 +110,7 @@
    <Exec Command="go install -ldflags &quot;-X github.com/pulumi/pulumi/sdk/v3/go/common/version.Version=$(Version)&quot; github.com/pulumi/pulumi/sdk/v3/go/pulumi-language-go"
          EnvironmentVariables="GOBIN=$(PulumiBin)"
          WorkingDirectory="$(SdkDirectory)"/>
-    
+
  </Target>

  <!-- This is where we build and install the NodeJS SDK -->
@ -145,7 +145,7 @@
          DestinationFolder="$(NodeJSSdkDirectory)\bin\tests\runtime\langhost\cases" />
    <Exec Command="go install -ldflags &quot;-X github.com/pulumi/pulumi/sdk/v3/go/common/version.Version=$(Version)&quot; github.com/pulumi/pulumi/sdk/v3/nodejs/cmd/pulumi-language-nodejs"
          WorkingDirectory="$(SdkDirectory)"/>
-    
+
  </Target>

  <Target Name="NodeJSInstallPlugin">
@ -160,7 +160,7 @@
    <Exec Command="go install -ldflags &quot;-X github.com/pulumi/pulumi/sdk/v3/go/common/version.Version=$(Version)&quot; github.com/pulumi/pulumi/sdk/v3/nodejs/cmd/pulumi-language-nodejs"
          EnvironmentVariables="GOBIN=$(PulumiBin)"
          WorkingDirectory="$(SdkDirectory)"/>
-    
+
  </Target>

  <Target Name="CopyNodeJSPackages">
@ -219,7 +219,7 @@
          WorkingDirectory="$(PythonSdkDirectory)\env\src" />
    <Exec Command="go install -ldflags &quot;-X github.com/pulumi/pulumi/sdk/v3/go/common/version.Version=$(Version)&quot; github.com/pulumi/pulumi/sdk/v3/python/cmd/pulumi-language-python"
    WorkingDirectory="$(SdkDirectory)"/>
-    
+
  </Target>

  <Target Name="PythonInstallPlugin">
@ -239,7 +239,7 @@
    <Exec Command="go install -ldflags &quot;-X github.com/pulumi/pulumi/sdk/v3/go/common/version.Version=$(Version)&quot; github.com/pulumi/pulumi/sdk/v3/python/cmd/pulumi-language-python"
          EnvironmentVariables="GOBIN=$(PulumiBin)"
          WorkingDirectory="$(SdkDirectory)"/>
-    
+
  </Target>

  <!-- Install the Pulumi SDK -->
@ -250,7 +250,7 @@
    <Exec Command="go install -ldflags &quot;-X github.com/pulumi/pulumi/pkg/v3/version.Version=v$(Version)&quot; github.com/pulumi/pulumi/pkg/v3/cmd/pulumi"
       EnvironmentVariables="GOBIN=$(PulumiBin)"
       WorkingDirectory="$(PkgDirectory)"/>
-    
+
  </Target>

  <!-- Build -->
@ -294,11 +294,16 @@
    <Exec Command="yarn run tsc" WorkingDirectory="$(TestsDirectory)\integration\construct_component\testcomponent" />
    <Exec Command="yarn run tsc" WorkingDirectory="$(TestsDirectory)\integration\construct_component_slow\testcomponent" />
    <Exec Command="yarn run tsc" WorkingDirectory="$(TestsDirectory)\integration\construct_component_plain\testcomponent" />
+
+    <!-- Install pulumi SDK into the venv managed by pipenv. -->
+    <Exec Command="pipenv run pip install -e ."
+          WorkingDirectory="$(PythonSdkDirectory)\env\src" />
  </Target>

  <!-- Tests -->
  <Target Name="Tests"
-	      DependsOnTargets="BuildTests">
+	  DependsOnTargets="BuildTests">
+
    <Exec Command="go test -timeout 5m -parallel $(TestParallelism) .\backend\..."
          IgnoreExitCode="true"
          WorkingDirectory="$(PkgDirectory)">
--- a/sdk/python/Makefile
+++ b/sdk/python/Makefile
@ -27,7 +27,7 @@ build_plugin::
 build:: build_package build_plugin

 lint::
-	pipenv run mypy ./lib/pulumi --config-file=mypy.ini
+	MYPYPATH=./stubs pipenv run mypy ./lib/pulumi --config-file=mypy.ini
 	pipenv run pylint ./lib/pulumi --rcfile=.pylintrc

 install_package:: build_package
--- a/sdk/python/Pipfile
+++ b/sdk/python/Pipfile
@ -6,7 +6,7 @@ name = "pypi"
 [packages]
 # Keep this list in sync with setup.py
 protobuf = ">=3.6.0"
-grpcio = ">=1.9.1,!=1.30.0"
+grpcio = ">=1.33.2"
 dill = ">=0.3.0"
 six = ">=1.12.0"
 semver = ">=2.8.1"
--- a/sdk/python/lib/pulumi/_async.py
+++ b/sdk/python/lib/pulumi/_async.py
@ -0,0 +1,50 @@
+# 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.
+
+"""Internal Async IO utilities, compat with Python 3.6."""
+
+import asyncio
+from typing import Any, Callable, TypeVar, cast
+
+
+_F = TypeVar('_F', bound=Callable[..., Any])
+
+
+def _asynchronized(func: _F) -> _F:
+    """Decorates a function to acquire and release a lock.
+
+    This makes sure that only one invocation of a function is active
+    on the current event loop at one time. Since this is an asyncio
+    lock, no real threads are blocked; only the invoking coroutine may
+    be blocked.
+
+    Usage:
+
+        class MyClass:
+
+            @_asynchronized
+            async def my_func(self, x, y=None):
+                ...
+
+    """
+    lock = asyncio.Lock()
+
+    async def sync_func(*args, **kw):
+        await lock.acquire()
+        try:
+            return await func(*args, **kw)
+        finally:
+            lock.release()
+
+    return cast(_F, sync_func)
--- a/sdk/python/lib/pulumi/dynamic/main.py
+++ b/sdk/python/lib/pulumi/dynamic/main.py
@ -107,7 +107,7 @@ class DynamicResourceProviderServicer(ResourceProviderServicer):
        props = rpc.deserialize_properties(request.properties)
        provider = get_provider(props)
        result = provider.create(props)
-        outs = result.outs
+        outs = result.outs if result.outs is not None else {}
        outs[PROVIDER_KEY] = props[PROVIDER_KEY]

        loop = asyncio.new_event_loop()
--- a/sdk/python/lib/pulumi/provider/init.py
+++ b/sdk/python/lib/pulumi/provider/init.py
@ -0,0 +1,22 @@
+# 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.
+
+from pulumi.provider.provider import Provider, ConstructResult
+from pulumi.provider.server import main
+
+__all__ = [
+    'Provider',
+    'ConstructResult',
+    'main'
+]
--- a/sdk/python/lib/pulumi/provider/provider.py
+++ b/sdk/python/lib/pulumi/provider/provider.py
@ -0,0 +1,58 @@
+# 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.
+
+from typing import Optional
+
+from pulumi import ResourceOptions, Input, Inputs
+
+
+class ConstructResult:
+    """ConstructResult represents the results of a call to
+`Provider.construct`.
+
+    """
+
+    urn: Input[str]
+    """The URN of the constructed resource."""
+
+    state: Inputs
+    """Any state that was computed during construction."""
+
+    def __init__(self, urn: Input[str], state: Inputs) -> None:
+        self.urn = urn
+        self.state = state
+
+
+class Provider:
+    """Provider represents an object that implements the resources and
+    functions for a particular Pulumi package.
+
+    """
+
+    version: str
+
+    def __init__(self, version: str) -> None:
+        self.version = version
+
+    def construct(self, name: str, resource_type: str, inputs: Inputs,
+                  options: Optional[ResourceOptions] = None) -> ConstructResult:
+        """Construct creates a new component resource.
+
+        :param name str: The name of the resource to create.
+        :param resource_type str: The type of the resource to create.
+        :param inputs Inputs: The inputs to the resource.
+        :param options Optional[ResourceOptions] The options for the resource.
+        """
+
+        raise Exception("Subclass of Provider must implement 'construct'")
--- a/sdk/python/lib/pulumi/provider/server.py
+++ b/sdk/python/lib/pulumi/provider/server.py
@ -0,0 +1,220 @@
+# 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.
+
+"""Define gRPC plumbing to expose a custom user-defined `Provider`
+instance as a gRPC server so that it can be used as a Pulumi plugin.
+
+"""
+
+from typing import Dict, List, Set, Optional, TypeVar, Any
+import argparse
+import asyncio
+import sys
+
+import grpc
+import grpc.aio
+
+from pulumi._async import _asynchronized
+from pulumi.provider.provider import Provider, ConstructResult
+from pulumi.resource import Resource, DependencyResource, DependencyProviderResource
+from pulumi.runtime import proto, rpc
+from pulumi.runtime.proto import provider_pb2_grpc, ResourceProviderServicer
+from pulumi.runtime.stack import wait_for_rpcs
+import pulumi
+import pulumi.resource
+import pulumi.runtime.config
+import pulumi.runtime.settings
+
+
+# _MAX_RPC_MESSAGE_SIZE raises the gRPC Max Message size from `4194304` (4mb) to `419430400` (400mb)
+_MAX_RPC_MESSAGE_SIZE = 1024 * 1024 * 400
+_GRPC_CHANNEL_OPTIONS = [('grpc.max_receive_message_length', _MAX_RPC_MESSAGE_SIZE)]
+
+
+class ProviderServicer(ResourceProviderServicer):
+    """Implements a subset of `ResourceProvider` methods to support
+    `Construct` and other methods invoked by the engine when the user
+    program creates a remote `ComponentResource` (with `remote=true`
+    in the constructor).
+
+    See `ResourceProvider` defined in `provider.proto`.
+
+    """
+
+    engine_address: str
+    provider: Provider
+    args: List[str]
+
+    # NOTE: remove @_asynchronized when we can avoid modifying globals in the method body.
+    @_asynchronized
+    async def Construct(self, request: proto.ConstructRequest, context) -> proto.ConstructResponse:  # pylint: disable=invalid-overridden-method
+        assert isinstance(request, proto.ConstructRequest), \
+            f'request is not ConstructRequest but is {type(request)} instead'
+
+        pulumi.runtime.settings.reset_options(
+            project=_empty_as_none(request.project),
+            stack=_empty_as_none(request.stack),
+            parallel=_zero_as_none(request.parallel),
+            engine_address=self.engine_address,
+            monitor_address=_empty_as_none(request.monitorEndpoint),
+            preview=request.dryRun)
+
+        pulumi.runtime.config.set_all_config(dict(request.config))
+
+        inputs = self._construct_inputs(request)
+
+        result = self.provider.construct(name=request.name,
+                                         resource_type=request.type,
+                                         inputs=inputs,
+                                         options=self._construct_options(request))
+
+        response = await self._construct_response(result)
+
+        # Wait for outstanding RPCs such as more provider Construct
+        # calls. This can happen if i.e. provider creates child
+        # resources but does not await their URN promises.
+        #
+        # Do not await all tasks as that starts hanging waiting for
+        # indefinite grpc.aio servier tasks.
+        await wait_for_rpcs(await_all_outstanding_tasks=False)
+
+        return response
+
+    @staticmethod
+    def _construct_inputs(request: proto.ConstructRequest) -> Dict[str, pulumi.Output]:
+
+        def deps(key: str) -> Set[Resource]:
+            return set(DependencyResource(urn) for urn in
+                       request.inputDependencies.get(
+                           key,
+                           proto.ConstructRequest.PropertyDependencies()
+                       ).urns)
+
+        return {
+            k: ProviderServicer._construct_output(the_input, deps=deps(k))
+            for k, the_input in
+            rpc.deserialize_properties(request.inputs, keep_unknowns=True).items()
+        }
+
+    @staticmethod
+    def _construct_output(the_input: Any, deps: Set[Resource]) -> Any:
+        is_secret = rpc.is_rpc_secret(the_input)
+
+        # If it's a prompt value, return it directly without wrapping
+        # it as an output.
+        if not is_secret and len(deps) == 0:
+            return the_input
+
+        # Otherwise, wrap it as an output so we can handle secrets
+        # and/or track dependencies.
+        return pulumi.Output(
+            resources=deps,
+            future=_as_future(rpc.unwrap_rpc_secret(the_input)),
+            is_known=_as_future(True),
+            is_secret=_as_future(is_secret))
+
+    @staticmethod
+    def _construct_options(request: proto.ConstructRequest) -> pulumi.ResourceOptions:
+        parent = None
+        if not _empty_as_none(request.parent):
+            parent = DependencyResource(request.parent)
+        return pulumi.ResourceOptions(
+            aliases=list(request.aliases),
+            depends_on=[DependencyResource(urn)
+                        for urn in request.dependencies],
+            protect=request.protect,
+            providers={pkg: DependencyProviderResource(ref)
+                       for pkg, ref in request.providers.items()},
+            parent=parent)
+
+    async def _construct_response(self, result: ConstructResult) -> proto.ConstructResponse:
+        urn = await pulumi.Output.from_input(result.urn).future()
+
+        # Note: property_deps is populated by rpc.serialize_properties.
+        property_deps: Dict[str, List[pulumi.resource.Resource]] = {}
+        state = await rpc.serialize_properties(
+            inputs={k: v for k, v in result.state.items() if k not in ['id', 'urn']},
+            property_deps=property_deps)
+
+        deps: Dict[str, proto.ConstructResponse.PropertyDependencies] = {}
+        for k, resources in property_deps.items():
+            urns = await asyncio.gather(*(r.urn.future() for r in resources))
+            deps[k] = proto.ConstructResponse.PropertyDependencies(urns=urns)
+
+        return proto.ConstructResponse(urn=urn,
+                                       state=state,
+                                       stateDependencies=deps)
+
+    async def Configure(self, request, context) -> proto.ConfigureResponse:  # pylint: disable=invalid-overridden-method
+        return proto.ConfigureResponse(acceptSecrets=True, acceptResources=True)
+
+    async def GetPluginInfo(self, request, context) -> proto.PluginInfo:  # pylint: disable=invalid-overridden-method
+        return proto.PluginInfo(version=self.provider.version)
+
+    def __init__(self, provider: Provider, args: List[str], engine_address: str) -> None:
+        super().__init__()
+        self.provider = provider
+        self.args = args
+        self.engine_address = engine_address
+
+
+def main(provider: Provider, args: List[str]) -> None:  # args not in use?
+    """For use as the `main` in programs that wrap a custom Provider
+    implementation into a Pulumi-compatible gRPC server.
+
+    :param provider: an instance of a Provider subclass
+
+    :args: command line arguiments such as os.argv[1:]
+
+    """
+
+    argp = argparse.ArgumentParser(description='Pulumi provider plugin (gRPC server)')
+    argp.add_argument('engine', help='Pulumi engine address')
+    engine_address: str = argp.parse_args().engine
+
+    async def serve() -> None:
+        server = grpc.aio.server(options=_GRPC_CHANNEL_OPTIONS)
+        servicer = ProviderServicer(provider, args, engine_address=engine_address)
+        provider_pb2_grpc.add_ResourceProviderServicer_to_server(servicer, server)
+        port = server.add_insecure_port(address='0.0.0.0:0')
+        await server.start()
+        sys.stdout.buffer.write(f'{port}\n'.encode())
+        sys.stdout.buffer.flush()
+        await server.wait_for_termination()
+
+    try:
+        loop = asyncio.get_event_loop()
+        try:
+            loop.run_until_complete(serve())
+        finally:
+            loop.close()
+    except KeyboardInterrupt:
+        pass
+
+
+T = TypeVar('T')  # pylint: disable=invalid-name
+
+
+def _as_future(value: T) -> 'asyncio.Future[T]':
+    fut: 'asyncio.Future[T]' = asyncio.Future()
+    fut.set_result(value)
+    return fut
+
+
+def _empty_as_none(text: str) -> Optional[str]:
+    return None if text == '' else text
+
+
+def _zero_as_none(value: int) -> Optional[int]:
+    return None if value == 0 else value
--- a/sdk/python/lib/pulumi/runtime/proto/plugin_pb2.pyi
+++ b/sdk/python/lib/pulumi/runtime/proto/plugin_pb2.pyi
@ -0,0 +1,19 @@
+# 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.
+
+class PluginInfo:
+
+    def __init__(self, version: str='') -> void: ...
+
+    version: string
--- a/sdk/python/lib/pulumi/runtime/proto/provider_pb2.pyi
+++ b/sdk/python/lib/pulumi/runtime/proto/provider_pb2.pyi
@ -0,0 +1,87 @@
+# 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.
+
+
+"""Manually constructed mypy typings. We should explore automated
+mypy typing generation from protobufs in the future.
+
+"""
+
+from typing import Dict, List, Optional
+from google.protobuf.struct_pb2 import Struct
+
+
+class ConstructRequest:
+    class PropertyDependencies:
+        urns: List[str]
+
+    project: str
+    stack: str
+    config: Dict[str,str]
+    dryRun: bool
+    parallel: int
+    monitorEndpoint: str
+    type: str
+    name: str
+    parent: str
+    inputs: Struct
+    inputDependencies: Dict[str,PropertyDependencies]
+    protect: bool
+    providers: Dict[str, str]
+    aliases: List[str]
+    dependencies: List[str]
+
+
+class ConstructResponse:
+
+    def __init__(self,
+                 urn: Optional[str]=None,
+                 state: Optional[Struct]=None,
+                 stateDependencies: Optional[Dict[str,PropertyDependencies]]=None) -> void:
+        pass
+
+    class PropertyDependencies:
+        urns: List[str]
+
+        def __init__(self, urns: List[str]) -> void:
+            pass
+
+    urn: str
+    state: Struct
+    stateDependencies: Dict[str,PropertyDependencies]
+
+
+class CheckResponse:
+    def __init__(self, inputs: Optional[Struct]=None, failures: List[CheckFailure]=[]) -> void:
+        pass
+
+    inputs: Struct
+    failures: List[CheckFailure]
+
+
+class CheckFailure:
+    property: str
+    reason: str
+
+
+class ConfigureResponse:
+
+    def __init__(self,
+                 acceptSecrets: bool=False,
+                 supportsPreview: bool=False,
+                 acceptResources: bool=False) -> void: ...
+
+    acceptSecrets: bool
+    supportsPreview: bool
+    acceptResources: bool
--- a/sdk/python/lib/pulumi/runtime/rpc.py
+++ b/sdk/python/lib/pulumi/runtime/rpc.py
@ -90,6 +90,14 @@ async def serialize_properties(inputs: 'Inputs',

    When `typ` is an input type, the metadata from the type is used to translate Python snake_case
    names to Pulumi camelCase names, rather than using the `input_transformer`.
+
+    Modifies given property_deps dict to collect discovered dependencies by property name.
+
+    :param Inputs inputs: The bag to serialize.
+
+    :param Dict[str, List[Resource]] property_deps: Dependencies are set here.
+
+    :param input_transfomer: Optional name translator.
    """

    # Default implementation of get_type that always returns None.
--- a/sdk/python/lib/pulumi/runtime/settings.py
+++ b/sdk/python/lib/pulumi/runtime/settings.py
@ -247,9 +247,9 @@ def grpc_error_to_exception(exn: grpc.RpcError) -> Optional[Exception]:


 def handle_grpc_error(exn: grpc.RpcError):
-    exn = grpc_error_to_exception(exn)
-    if exn is not None:
-        raise exn
+    exc = grpc_error_to_exception(exn)
+    if exc is not None:
+        raise exc

 async def monitor_supports_secrets() -> bool:
    return await monitor_supports_feature("secrets")
--- a/sdk/python/lib/pulumi/runtime/stack.py
+++ b/sdk/python/lib/pulumi/runtime/stack.py
@ -42,27 +42,36 @@ async def run_pulumi_func(func: Callable):
    try:
        func()
    finally:
-        log.debug("Waiting for outstanding RPCs to complete")
+        await wait_for_rpcs()

-        while True:
-            # Pump the event loop, giving all of the RPCs that we just queued up time to fully execute.
-            # The asyncio scheduler does not expose a "yield" primitive, so this will have to do.
-            #
-            # Note that "asyncio.sleep(0)" is the blessed way to do this:
-            # https://github.com/python/asyncio/issues/284#issuecomment-154180935
-            #
-            # We await each RPC in turn so that this loop will actually block rather than busy-wait.
-            while len(RPC_MANAGER.rpcs) > 0:
-                await asyncio.sleep(0)
-                log.debug(f"waiting for quiescence; {len(RPC_MANAGER.rpcs)} RPCs outstanding")
-                await RPC_MANAGER.rpcs.pop()
+        # By now, all tasks have exited and we're good to go.
+        log.debug("run_pulumi_func completed")

-            if RPC_MANAGER.unhandled_exception is not None:
-                raise RPC_MANAGER.unhandled_exception.with_traceback(RPC_MANAGER.exception_traceback)

-            log.debug("RPCs successfully completed")
+async def wait_for_rpcs(await_all_outstanding_tasks=True) -> None:
+    log.debug("Waiting for outstanding RPCs to complete")
+
+    while True:
+        # Pump the event loop, giving all of the RPCs that we just queued up time to fully execute.
+        # The asyncio scheduler does not expose a "yield" primitive, so this will have to do.
+        #
+        # Note that "asyncio.sleep(0)" is the blessed way to do this:
+        # https://github.com/python/asyncio/issues/284#issuecomment-154180935
+        #
+        # We await each RPC in turn so that this loop will actually block rather than busy-wait.
+        while len(RPC_MANAGER.rpcs) > 0:
+            await asyncio.sleep(0)
+            log.debug(f"waiting for quiescence; {len(RPC_MANAGER.rpcs)} RPCs outstanding")
+            await RPC_MANAGER.rpcs.pop()
+
+        if RPC_MANAGER.unhandled_exception is not None:
+            raise RPC_MANAGER.unhandled_exception.with_traceback(RPC_MANAGER.exception_traceback)
+
+        log.debug("RPCs successfully completed")
+
+        # If the RPCs have successfully completed, now await all remaining outstanding tasks.
+        if await_all_outstanding_tasks:

-            # If the RPCs have successfully completed, now await all remaining outstanding tasks.
            outstanding_tasks = _get_running_tasks()
            if len(outstanding_tasks) == 0:
                log.debug("No outstanding tasks to complete")
@ -86,13 +95,10 @@ async def run_pulumi_func(func: Callable):

                log.debug("All outstanding tasks completed.")

-            # Check to see if any more RPCs have been scheduled, and repeat the cycle if so.
-            # Break if no RPCs remain.
-            if len(RPC_MANAGER.rpcs) == 0:
-                break
-
-        # By now, all tasks have exited and we're good to go.
-        log.debug("run_pulumi_func completed")
+        # Check to see if any more RPCs have been scheduled, and repeat the cycle if so.
+        # Break if no RPCs remain.
+        if len(RPC_MANAGER.rpcs) == 0:
+            break


 async def run_in_stack(func: Callable):
--- a/sdk/python/lib/test/provider/test_server.py
+++ b/sdk/python/lib/test/provider/test_server.py
@ -0,0 +1,50 @@
+# 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.
+
+from typing import Dict, Any
+
+import pytest
+
+from pulumi.runtime.proto.provider_pb2 import ConstructRequest
+from pulumi.provider.server import ProviderServicer
+from pulumi.runtime import proto, rpc
+import google.protobuf.struct_pb2 as struct_pb2
+
+
+@pytest.mark.asyncio
+async def test_construct_inputs_parses_request():
+    value = 'foobar'
+    inputs = _as_struct({'echo': value})
+    req = ConstructRequest(inputs=inputs)
+    inputs = ProviderServicer._construct_inputs(req)
+    assert len(inputs) == 1
+    fut_v = await inputs['echo'].future()
+    assert fut_v == value
+
+
+@pytest.mark.asyncio
+async def test_construct_inputs_preserves_unknowns():
+    unknown = '04da6b54-80e4-46f7-96ec-b56ff0331ba9'
+    inputs = _as_struct({'echo': unknown})
+    req = ConstructRequest(inputs=inputs)
+    inputs = ProviderServicer._construct_inputs(req)
+    assert len(inputs) == 1
+    fut_v = await inputs['echo'].future()
+    assert fut_v is None
+
+
+def _as_struct(key_values: Dict[str, Any]) -> struct_pb2.Struct:
+    the_struct = struct_pb2.Struct()
+    the_struct.update(key_values)  # pylint: disable=no-member
+    return the_struct
--- a/sdk/python/mypy.ini
+++ b/sdk/python/mypy.ini
@ -4,9 +4,6 @@

 # Per-module options:

-[mypy-grpc]
-ignore_missing_imports = True
-
 [mypy-dill]
 ignore_missing_imports = True

@ -19,4 +16,3 @@ ignore_missing_imports = True
 # grpc generated
 [mypy-pulumi.runtime.proto.*]
 ignore_errors = True
-
--- a/sdk/python/stubs/grpc/init.pyi
+++ b/sdk/python/stubs/grpc/init.pyi
@ -0,0 +1,70 @@
+# 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 enum
+import threading
+import typing
+from concurrent import futures
+
+
+class Compression(enum.IntEnum):
+    NoCompression = ...
+    Deflate = ...
+    Gzip = ...
+
+
+class StatusCode(enum.Enum):
+    OK = ...
+    CANCELLED = ...
+    UNKNOWN = ...
+    INVALID_ARGUMENT = ...
+    DEADLINE_EXCEEDED = ...
+    NOT_FOUND = ...
+    ALREADY_EXISTS = ...
+    PERMISSION_DENIED = ...
+    UNAUTHENTICATED = ...
+    RESOURCE_EXHAUSTED = ...
+    FAILED_PRECONDITION = ...
+    ABORTED = ...
+    UNIMPLEMENTED = ...
+    INTERNAL = ...
+    UNAVAILABLE = ...
+    DATA_LOSS = ...
+
+
+class Channel:
+    pass
+
+
+class Server:
+    def add_insecure_port(self, address: str) -> int: ...
+    def start(self) -> None: ...
+    def stop(self, grace: typing.Optional[float] = None) -> threading.Event: ...
+
+
+class RpcError(Exception):
+    def code(self) -> StatusCode: ...
+    def details(self) -> str: ...
+
+
+
+def insecure_channel(
+    target: str,
+    options: typing.Any = None,
+    compression: typing.Optional[Compression] = None,
+) -> Channel:
+    ...
+
+
+def server(thread_pool: futures.ThreadPoolExecutor, options: typing.Any) -> Server: ...
--- a/sdk/python/stubs/grpc/aio.pyi
+++ b/sdk/python/stubs/grpc/aio.pyi
@ -0,0 +1,26 @@
+# 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.
+
+from typing import Optional, Sequence, Any
+import grpc
+
+
+class Server:
+    def add_insecure_port(self, address: str) -> int: ...
+    async def start(self) -> None: ...
+    async def stop(self, grace: Optional[float]) -> None: ...
+    async def wait_for_termination(self, timeout: Optional[float]=...) -> bool: ...
+
+
+def server(options: Any) -> Server: ...
--- a/tests/integration/construct_component/testcomponent-python/pulumi-resource-testcomponent
+++ b/tests/integration/construct_component/testcomponent-python/pulumi-resource-testcomponent
@ -0,0 +1,7 @@
+#!/bin/bash
+
+set -euo pipefail
+
+SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+
+exec "$PULUMI_RUNTIME_VIRTUALENV/bin/python" "$SCRIPT_DIR/testcomponent.py" "$@"
--- a/tests/integration/construct_component/testcomponent-python/pulumi-resource-testcomponent.cmd
+++ b/tests/integration/construct_component/testcomponent-python/pulumi-resource-testcomponent.cmd
@ -0,0 +1,4 @@
+@echo off
+setlocal
+set SCRIPT_DIR=%~dp0
+@%PULUMI_RUNTIME_VIRTUALENV%\Scripts\python.exe "%SCRIPT_DIR%/testcomponent.py" %*
--- a/tests/integration/construct_component/testcomponent-python/testcomponent.py
+++ b/tests/integration/construct_component/testcomponent-python/testcomponent.py
@ -0,0 +1,71 @@
+# 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.
+
+from typing import Any, Optional
+import sys
+
+from pulumi import Input, Inputs, ComponentResource, ResourceOptions
+import pulumi
+import pulumi.dynamic as dynamic
+import pulumi.provider as provider
+
+
+_ID = 0
+
+
+class MyDynamicProvider(dynamic.ResourceProvider):
+
+    def create(self, props: Any) -> dynamic.CreateResult:
+        global _ID
+        _ID = _ID + 1
+        return dynamic.CreateResult(id_=str(_ID))
+
+
+class Resource(dynamic.Resource):
+    def __init__(self, name: str, echo: Input[any], opts: Optional[ResourceOptions]=None):
+        super().__init__(MyDynamicProvider(), name, {'echo': echo}, opts)
+
+
+class Component(ComponentResource):
+    def __init__(self, name: str, echo: Input[any], opts: Optional[ResourceOptions]=None):
+        super().__init__('testcomponent:index:Component', name, {}, opts)
+        self.echo = pulumi.Output.from_input(echo)
+        resource = Resource('child-{}'.format(name), echo, ResourceOptions(parent=self))
+        self.child_id = resource.id
+
+
+class Provider(provider.Provider):
+    VERSION = "0.0.1"
+
+    def __init__(self):
+        super().__init__(Provider.VERSION)
+
+    def construct(self, name: str, resource_type: str, inputs: Inputs,
+                  options: Optional[ResourceOptions]=None) -> provider.ConstructResult:
+
+        if resource_type != 'testcomponent:index:Component':
+            raise Exception('unknown resource type {}'.format(resource_type))
+
+        component = Component(name, inputs['echo'], options)
+
+        return provider.ConstructResult(
+            urn=component.urn,
+            state={
+                'echo': component.echo,
+                'childId': component.child_id
+            })
+
+
+if __name__ == '__main__':
+    provider.main(Provider(), sys.argv[1:])
--- a/tests/integration/construct_component_plain/testcomponent-python/pulumi-resource-testcomponent
+++ b/tests/integration/construct_component_plain/testcomponent-python/pulumi-resource-testcomponent
@ -0,0 +1,7 @@
+#!/bin/bash
+
+set -euo pipefail
+
+SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+
+exec "$PULUMI_RUNTIME_VIRTUALENV/bin/python" "$SCRIPT_DIR/testcomponent.py" "$@"
--- a/tests/integration/construct_component_plain/testcomponent-python/pulumi-resource-testcomponent.cmd
+++ b/tests/integration/construct_component_plain/testcomponent-python/pulumi-resource-testcomponent.cmd
@ -0,0 +1,4 @@
+@echo off
+setlocal
+set SCRIPT_DIR=%~dp0
+@%PULUMI_RUNTIME_VIRTUALENV%\Scripts\python.exe "%SCRIPT_DIR%/testcomponent.py" %*
--- a/tests/integration/construct_component_plain/testcomponent-python/testcomponent.py
+++ b/tests/integration/construct_component_plain/testcomponent-python/testcomponent.py
@ -0,0 +1,71 @@
+# 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.
+
+from typing import Any, Optional
+import sys
+
+from pulumi import Input, Inputs, ComponentResource, ResourceOptions
+import pulumi
+import pulumi.dynamic as dynamic
+import pulumi.provider as provider
+
+
+_ID = 0
+
+
+class MyDynamicProvider(dynamic.ResourceProvider):
+
+    def create(self, props: Any) -> dynamic.CreateResult:
+        global _ID
+        _ID = _ID + 1
+        return dynamic.CreateResult(id_=str(_ID))
+
+
+class Resource(dynamic.Resource):
+    def __init__(self, name: str, options: Optional[ResourceOptions]=None):
+        super().__init__(MyDynamicProvider(), name, {}, options)
+
+
+class Component(ComponentResource):
+    def __init__(self, name: str, children: int, options: Optional[ResourceOptions] = None):
+        super().__init__('testcomponent:index:Component', name, {}, options)
+
+        for i in range(0, children):
+            Resource(f'child-{name}-{i+1}', options=ResourceOptions(parent=self))
+
+
+class Provider(provider.Provider):
+    VERSION = "0.0.1"
+
+    def __init__(self):
+        super().__init__(Provider.VERSION)
+
+    def construct(self,
+                  name: str,
+                  resource_type: str,
+                  inputs: Inputs,
+                  options: Optional[ResourceOptions] = None) -> provider.ConstructResult:
+
+        if resource_type != 'testcomponent:index:Component':
+            raise Exception('unknown resource type {}'.format(resource_type))
+
+        component = Component(name,
+                              children=int(inputs.get('children', 0)),
+                              options=options)
+
+        return provider.ConstructResult(urn=component.urn, state={})
+
+
+if __name__ == '__main__':
+    provider.main(Provider(), sys.argv[1:])
--- a/tests/integration/integration_dotnet_test.go
+++ b/tests/integration/integration_dotnet_test.go
@ -196,11 +196,6 @@ func TestLargeResourceDotNet(t *testing.T) {

 // Test remote component construction in .NET.
 func TestConstructDotnet(t *testing.T) {
-	pathEnv, err := testComponentPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}
-
 	// TODO[pulumi/pulumi#5455]: Dynamic providers fail to load when used from multi-lang components.
 	// Until we've addressed this, set PULUMI_TEST_YARN_LINK_PULUMI, which tells the integration test
 	// module to run `yarn install && yarn link @pulumi/pulumi` in the .NET program's directory, allowing
@ -209,12 +204,14 @@ func TestConstructDotnet(t *testing.T) {
 	// test module should be removed.
 	const testYarnLinkPulumiEnv = "PULUMI_TEST_YARN_LINK_PULUMI=true"

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
-		Env:          []string{pathEnv, testYarnLinkPulumiEnv},
+	runtimeVenv := pulumiRuntimeVirtualEnv(t, filepath.Join("..", ".."))
+
+	opts := &integration.ProgramTestOptions{
+		Env:          []string{testYarnLinkPulumiEnv, runtimeVenv},
 		Dir:          filepath.Join("construct_component", "dotnet"),
 		Dependencies: []string{"Pulumi"},
 		Quick:        true,
+		NoParallel:   true, // avoid contention for Dir
 		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
 			assert.NotNil(t, stackInfo.Deployment)
 			if assert.Equal(t, 9, len(stackInfo.Deployment.Resources)) {
@ -242,15 +239,16 @@ func TestConstructDotnet(t *testing.T) {
 			}
 		},
 	}
-	integration.ProgramTest(t, opts)
+
+	runProgramSubTests(t, opts, map[string]string{
+		"WithNodeProvider":   componentPathEnv(t, "construct_component", "testcomponent"),
+		"WithPythonProvider": componentPathEnv(t, "construct_component", "testcomponent-python"),
+	})
 }

 // Test remote component construction with a child resource that takes a long time to be created, ensuring it's created.
 func TestConstructSlowDotnet(t *testing.T) {
-	pathEnv, err := testComponentSlowPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}
+	pathEnv := testComponentSlowPathEnv(t)

 	// TODO[pulumi/pulumi#5455]: Dynamic providers fail to load when used from multi-lang components.
 	// Until we've addressed this, set PULUMI_TEST_YARN_LINK_PULUMI, which tells the integration test
@ -260,8 +258,7 @@ func TestConstructSlowDotnet(t *testing.T) {
 	// test module should be removed.
 	const testYarnLinkPulumiEnv = "PULUMI_TEST_YARN_LINK_PULUMI=true"

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
+	opts := &integration.ProgramTestOptions{
 		Env:          []string{pathEnv, testYarnLinkPulumiEnv},
 		Dir:          filepath.Join("construct_component_slow", "dotnet"),
 		Dependencies: []string{"Pulumi"},
@ -281,10 +278,7 @@ func TestConstructSlowDotnet(t *testing.T) {

 // Test remote component construction with prompt inputs.
 func TestConstructPlainDotnet(t *testing.T) {
-	pathEnv, err := testComponentPlainPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}
+	runtimeVenv := pulumiRuntimeVirtualEnv(t, filepath.Join("..", ".."))

 	// TODO[pulumi/pulumi#5455]: Dynamic providers fail to load when used from multi-lang components.
 	// Until we've addressed this, set PULUMI_TEST_YARN_LINK_PULUMI, which tells the integration test
@ -294,18 +288,22 @@ func TestConstructPlainDotnet(t *testing.T) {
 	// test module should be removed.
 	const testYarnLinkPulumiEnv = "PULUMI_TEST_YARN_LINK_PULUMI=true"

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
-		Env:          []string{pathEnv, testYarnLinkPulumiEnv},
+	opts := &integration.ProgramTestOptions{
+		Env:          []string{testYarnLinkPulumiEnv, runtimeVenv},
 		Dir:          filepath.Join("construct_component_plain", "dotnet"),
 		Dependencies: []string{"Pulumi"},
+		NoParallel:   true, // avoid contention for Dir
 		Quick:        true,
 		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
 			assert.NotNil(t, stackInfo.Deployment)
 			assert.Equal(t, 9, len(stackInfo.Deployment.Resources))
 		},
 	}
-	integration.ProgramTest(t, opts)
+
+	runProgramSubTests(t, opts, map[string]string{
+		"WithNodeProvider":   componentPathEnv(t, "construct_component_plain", "testcomponent"),
+		"WithPythonProvider": componentPathEnv(t, "construct_component_plain", "testcomponent-python"),
+	})
 }

 func TestGetResourceDotnet(t *testing.T) {
--- a/tests/integration/integration_go_test.go
+++ b/tests/integration/integration_go_test.go
@ -121,10 +121,6 @@ func TestLargeResourceGo(t *testing.T) {

 // Test remote component construction in Go.
 func TestConstructGo(t *testing.T) {
-	pathEnv, err := testComponentPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}

 	// TODO[pulumi/pulumi#5455]: Dynamic providers fail to load when used from multi-lang components.
 	// Until we've addressed this, set PULUMI_TEST_YARN_LINK_PULUMI, which tells the integration test
@ -134,14 +130,16 @@ func TestConstructGo(t *testing.T) {
 	// test module should be removed.
 	const testYarnLinkPulumiEnv = "PULUMI_TEST_YARN_LINK_PULUMI=true"

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
-		Env: []string{pathEnv, testYarnLinkPulumiEnv},
+	runtimeVenv := pulumiRuntimeVirtualEnv(t, filepath.Join("..", ".."))
+
+	opts := &integration.ProgramTestOptions{
+		Env: []string{testYarnLinkPulumiEnv, runtimeVenv},
 		Dir: filepath.Join("construct_component", "go"),
 		Dependencies: []string{
 			"github.com/pulumi/pulumi/sdk/v3",
 		},
-		Quick: true,
+		Quick:      true,
+		NoParallel: true, // avoid contention for Dir
 		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
 			assert.NotNil(t, stackInfo.Deployment)
 			if assert.Equal(t, 9, len(stackInfo.Deployment.Resources)) {
@ -172,15 +170,16 @@ func TestConstructGo(t *testing.T) {
 			}
 		},
 	}
-	integration.ProgramTest(t, opts)
+
+	runProgramSubTests(t, opts, map[string]string{
+		"WithNodeProvider":   componentPathEnv(t, "construct_component", "testcomponent"),
+		"WithPythonProvider": componentPathEnv(t, "construct_component", "testcomponent-python"),
+	})
 }

 // Test remote component construction with a child resource that takes a long time to be created, ensuring it's created.
 func TestConstructSlowGo(t *testing.T) {
-	pathEnv, err := testComponentSlowPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}
+	pathEnv := testComponentSlowPathEnv(t)

 	// TODO[pulumi/pulumi#5455]: Dynamic providers fail to load when used from multi-lang components.
 	// Until we've addressed this, set PULUMI_TEST_YARN_LINK_PULUMI, which tells the integration test
@ -190,8 +189,7 @@ func TestConstructSlowGo(t *testing.T) {
 	// test module should be removed.
 	const testYarnLinkPulumiEnv = "PULUMI_TEST_YARN_LINK_PULUMI=true"

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
+	opts := &integration.ProgramTestOptions{
 		Env: []string{pathEnv, testYarnLinkPulumiEnv},
 		Dir: filepath.Join("construct_component_slow", "go"),
 		Dependencies: []string{
@ -213,10 +211,7 @@ func TestConstructSlowGo(t *testing.T) {

 // Test remote component construction with prompt inputs.
 func TestConstructPlainGo(t *testing.T) {
-	pathEnv, err := testComponentPlainPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}
+	runtimeVenv := pulumiRuntimeVirtualEnv(t, filepath.Join("..", ".."))

 	// TODO[pulumi/pulumi#5455]: Dynamic providers fail to load when used from multi-lang components.
 	// Until we've addressed this, set PULUMI_TEST_YARN_LINK_PULUMI, which tells the integration test
@ -226,20 +221,24 @@ func TestConstructPlainGo(t *testing.T) {
 	// test module should be removed.
 	const testYarnLinkPulumiEnv = "PULUMI_TEST_YARN_LINK_PULUMI=true"

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
-		Env: []string{pathEnv, testYarnLinkPulumiEnv},
+	opts := &integration.ProgramTestOptions{
+		Env: []string{runtimeVenv, testYarnLinkPulumiEnv},
 		Dir: filepath.Join("construct_component_plain", "go"),
 		Dependencies: []string{
 			"github.com/pulumi/pulumi/sdk/v2",
 		},
-		Quick: true,
+		Quick:      true,
+		NoParallel: true, // avoid contention for Dir
 		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
 			assert.NotNil(t, stackInfo.Deployment)
 			assert.Equal(t, 9, len(stackInfo.Deployment.Resources))
 		},
 	}
-	integration.ProgramTest(t, opts)
+
+	runProgramSubTests(t, opts, map[string]string{
+		"WithNodeProvider":   componentPathEnv(t, "construct_component_plain", "testcomponent"),
+		"WithPythonProvider": componentPathEnv(t, "construct_component_plain", "testcomponent-python"),
+	})
 }

 func TestGetResourceGo(t *testing.T) {
--- a/tests/integration/integration_nodejs_test.go
+++ b/tests/integration/integration_nodejs_test.go
@ -60,6 +60,7 @@ func TestEngineEvents(t *testing.T) {
 		Dir:          "single_resource",
 		Dependencies: []string{"@pulumi/pulumi"},
 		Quick:        true,
+		NoParallel:   true, // avoid contention for Dir
 		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
 			// Ensure that we have a non-empty list of events.
 			assert.NotEmpty(t, stackInfo.Events)
@ -675,17 +676,13 @@ func TestConstructNode(t *testing.T) {
 	if runtime.GOOS == WindowsOS {
 		t.Skip("Temporarily skipping test on Windows")
 	}
-	pathEnv, err := testComponentPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
-		Env:          []string{pathEnv},
+	opts := &integration.ProgramTestOptions{
+		Env:          []string{pulumiRuntimeVirtualEnv(t, filepath.Join("..", ".."))},
 		Dir:          filepath.Join("construct_component", "nodejs"),
 		Dependencies: []string{"@pulumi/pulumi"},
 		Quick:        true,
+		NoParallel:   true,
 		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
 			assert.NotNil(t, stackInfo.Deployment)
 			if assert.Equal(t, 9, len(stackInfo.Deployment.Resources)) {
@ -713,15 +710,16 @@ func TestConstructNode(t *testing.T) {
 			}
 		},
 	}
-	integration.ProgramTest(t, opts)
+
+	runProgramSubTests(t, opts, map[string]string{
+		"WithNodeProvider":   componentPathEnv(t, "construct_component", "testcomponent"),
+		"WithPythonProvider": componentPathEnv(t, "construct_component", "testcomponent-python"),
+	})
 }

 // Test remote component construction with a child resource that takes a long time to be created, ensuring it's created.
 func TestConstructSlowNode(t *testing.T) {
-	pathEnv, err := testComponentSlowPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}
+	pathEnv := testComponentSlowPathEnv(t)

 	var opts *integration.ProgramTestOptions
 	opts = &integration.ProgramTestOptions{
@ -744,23 +742,21 @@ func TestConstructSlowNode(t *testing.T) {

 // Test remote component construction with prompt inputs.
 func TestConstructPlainNode(t *testing.T) {
-	pathEnv, err := testComponentPlainPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}
-
-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
-		Env:          []string{pathEnv},
+	opts := &integration.ProgramTestOptions{
+		Env:          []string{pulumiRuntimeVirtualEnv(t, filepath.Join("..", ".."))},
 		Dir:          filepath.Join("construct_component_plain", "nodejs"),
 		Dependencies: []string{"@pulumi/pulumi"},
 		Quick:        true,
+		NoParallel:   true,
 		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
 			assert.NotNil(t, stackInfo.Deployment)
 			assert.Equal(t, 9, len(stackInfo.Deployment.Resources))
 		},
 	}
-	integration.ProgramTest(t, opts)
+	runProgramSubTests(t, opts, map[string]string{
+		"WithNodeProvider":   componentPathEnv(t, "construct_component_plain", "testcomponent"),
+		"WithPythonProvider": componentPathEnv(t, "construct_component_plain", "testcomponent-python"),
+	})
 }

 func TestGetResourceNode(t *testing.T) {
--- a/tests/integration/integration_python_test.go
+++ b/tests/integration/integration_python_test.go
@ -378,10 +378,6 @@ func TestPythonResourceArgs(t *testing.T) {

 // Test remote component construction in Python.
 func TestConstructPython(t *testing.T) {
-	pathEnv, err := testComponentPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}

 	// TODO[pulumi/pulumi#5455]: Dynamic providers fail to load when used from multi-lang components.
 	// Until we've addressed this, set PULUMI_TEST_YARN_LINK_PULUMI, which tells the integration test
@ -391,14 +387,16 @@ func TestConstructPython(t *testing.T) {
 	// test module should be removed.
 	const testYarnLinkPulumiEnv = "PULUMI_TEST_YARN_LINK_PULUMI=true"

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
-		Env: []string{pathEnv, testYarnLinkPulumiEnv},
+	runtimeVenv := pulumiRuntimeVirtualEnv(t, filepath.Join("..", ".."))
+
+	opts := &integration.ProgramTestOptions{
+		Env: []string{testYarnLinkPulumiEnv, runtimeVenv},
 		Dir: filepath.Join("construct_component", "python"),
 		Dependencies: []string{
 			filepath.Join("..", "..", "sdk", "python", "env", "src"),
 		},
-		Quick: true,
+		Quick:      true,
+		NoParallel: true, // avoid contention for Dir
 		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
 			assert.NotNil(t, stackInfo.Deployment)
 			if assert.Equal(t, 9, len(stackInfo.Deployment.Resources)) {
@ -429,15 +427,16 @@ func TestConstructPython(t *testing.T) {
 			}
 		},
 	}
-	integration.ProgramTest(t, opts)
+
+	runProgramSubTests(t, opts, map[string]string{
+		"WithNodeProvider":   componentPathEnv(t, "construct_component", "testcomponent"),
+		"WithPythonProvider": componentPathEnv(t, "construct_component", "testcomponent-python"),
+	})
 }

 // Test remote component construction with a child resource that takes a long time to be created, ensuring it's created.
 func TestConstructSlowPython(t *testing.T) {
-	pathEnv, err := testComponentSlowPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}
+	pathEnv := testComponentSlowPathEnv(t)

 	// TODO[pulumi/pulumi#5455]: Dynamic providers fail to load when used from multi-lang components.
 	// Until we've addressed this, set PULUMI_TEST_YARN_LINK_PULUMI, which tells the integration test
@ -447,8 +446,7 @@ func TestConstructSlowPython(t *testing.T) {
 	// test module should be removed.
 	const testYarnLinkPulumiEnv = "PULUMI_TEST_YARN_LINK_PULUMI=true"

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
+	opts := &integration.ProgramTestOptions{
 		Env: []string{pathEnv, testYarnLinkPulumiEnv},
 		Dir: filepath.Join("construct_component_slow", "python"),
 		Dependencies: []string{
@ -470,10 +468,6 @@ func TestConstructSlowPython(t *testing.T) {

 // Test remote component construction with prompt inputs.
 func TestConstructPlainPython(t *testing.T) {
-	pathEnv, err := testComponentPlainPathEnv()
-	if err != nil {
-		t.Fatalf("failed to build test component PATH: %v", err)
-	}

 	// TODO[pulumi/pulumi#5455]: Dynamic providers fail to load when used from multi-lang components.
 	// Until we've addressed this, set PULUMI_TEST_YARN_LINK_PULUMI, which tells the integration test
@ -483,20 +477,26 @@ func TestConstructPlainPython(t *testing.T) {
 	// test module should be removed.
 	const testYarnLinkPulumiEnv = "PULUMI_TEST_YARN_LINK_PULUMI=true"

-	var opts *integration.ProgramTestOptions
-	opts = &integration.ProgramTestOptions{
-		Env: []string{pathEnv, testYarnLinkPulumiEnv},
+	runtimeVenv := pulumiRuntimeVirtualEnv(t, filepath.Join("..", ".."))
+
+	opts := &integration.ProgramTestOptions{
+		Env: []string{testYarnLinkPulumiEnv, runtimeVenv},
 		Dir: filepath.Join("construct_component_plain", "python"),
 		Dependencies: []string{
 			filepath.Join("..", "..", "sdk", "python", "env", "src"),
 		},
-		Quick: true,
+		Quick:      true,
+		NoParallel: true, // avoid contention for Dir
 		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
 			assert.NotNil(t, stackInfo.Deployment)
 			assert.Equal(t, 9, len(stackInfo.Deployment.Resources))
 		},
 	}
-	integration.ProgramTest(t, opts)
+
+	runProgramSubTests(t, opts, map[string]string{
+		"WithNodeProvider":   componentPathEnv(t, "construct_component_plain", "testcomponent"),
+		"WithPythonProvider": componentPathEnv(t, "construct_component_plain", "testcomponent-python"),
+	})
 }

 func TestGetResourcePython(t *testing.T) {
--- a/tests/integration/integration_test.go
+++ b/tests/integration/integration_test.go
@ -5,6 +5,7 @@ package ints
 import (
 	"fmt"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"runtime"
 	"strings"
@ -566,34 +567,80 @@ func TestConfigPaths(t *testing.T) {
 }

 //nolint:golint,deadcode
-func testComponentPathEnv() (string, error) {
-	return componentPathEnv("construct_component", "testcomponent")
+func testComponentPathEnv(t *testing.T) string {
+	return componentPathEnv(t, "construct_component", "testcomponent")
 }

 //nolint:golint,deadcode
-func testComponentSlowPathEnv() (string, error) {
-	return componentPathEnv("construct_component_slow", "testcomponent")
+func testComponentSlowPathEnv(t *testing.T) string {
+	return componentPathEnv(t, "construct_component_slow", "testcomponent")
 }

 //nolint:golint,deadcode
-func testComponentPlainPathEnv() (string, error) {
-	return componentPathEnv("construct_component_plain", "testcomponent")
+func testComponentPlainPathEnv(t *testing.T) string {
+	return componentPathEnv(t, "construct_component_plain", "testcomponent")
 }

-func componentPathEnv(integrationTest, componentDir string) (string, error) {
+func componentPathEnv(t *testing.T, integrationTest, componentDir string) string {
 	cwd, err := os.Getwd()
 	if err != nil {
-		return "", err
+		t.Fatal(err)
+		return ""
 	}
 	absCwd, err := filepath.Abs(cwd)
 	if err != nil {
-		return "", err
+		t.Fatal(err)
+		return ""
 	}
 	pluginDir := filepath.Join(absCwd, integrationTest, componentDir)
-
 	pathSeparator := ":"
 	if runtime.GOOS == "windows" {
 		pathSeparator = ";"
 	}
-	return "PATH=" + os.Getenv("PATH") + pathSeparator + pluginDir, nil
+	return "PATH=" + os.Getenv("PATH") + pathSeparator + pluginDir
+}
+
+// nolint: unused,deadcode
+func venvFromPipenv(relativeWorkdir string) (string, error) {
+	workdir, err := filepath.Abs(relativeWorkdir)
+	if err != nil {
+		return "", err
+	}
+	cmd := exec.Command("pipenv", "--venv")
+	cmd.Dir = workdir
+	dir, err := cmd.Output()
+	if err != nil {
+		return "", err
+	}
+	venv := strings.TrimRight(string(dir), "\r\n")
+	if _, err := os.Stat(venv); os.IsNotExist(err) {
+		return "", fmt.Errorf("Folder '%s' returned by 'pipenv --venv' from %s does not exist: %w",
+			venv, workdir, err)
+	}
+	return venv, nil
+}
+
+// nolint: unused,deadcode
+func pulumiRuntimeVirtualEnv(t *testing.T, pulumiRepoRootDir string) string {
+	venvFolder, err := venvFromPipenv(filepath.Join(pulumiRepoRootDir, "sdk", "python"))
+	if err != nil {
+		t.Fatal(fmt.Errorf("PULUMI_RUNTIME_VIRTUALENV guess failed: %w", err))
+		return ""
+	}
+	r := fmt.Sprintf("PULUMI_RUNTIME_VIRTUALENV=%s", venvFolder)
+	return r
+}
+
+// nolint: unused,deadcode
+func runProgramSubTests(t *testing.T, opts *integration.ProgramTestOptions, envExtensions map[string]string) {
+	extend := func(extraEnv string, opts integration.ProgramTestOptions) integration.ProgramTestOptions {
+		opts.Env = append(opts.Env, extraEnv)
+		return opts
+	}
+	for subTestName, extraEnv := range envExtensions {
+		t.Run(subTestName, func(t *testing.T) {
+			subTestOpts := extend(extraEnv, *opts)
+			integration.ProgramTest(t, &subTestOpts)
+		})
+	}
 }