diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md
index ef123ba08..dc84a0a0d 100644
--- a/CHANGELOG_PENDING.md
+++ b/CHANGELOG_PENDING.md
@@ -14,3 +14,5 @@
 
 ### Bug Fixes
 
+- [sdk/python] Fix mocks issue when passing a resource more than once.
+  [#6479](https://github.com/pulumi/pulumi/pull/6479)
diff --git a/sdk/python/lib/pulumi/runtime/mocks.py b/sdk/python/lib/pulumi/runtime/mocks.py
index de66c639a..74c118133 100644
--- a/sdk/python/lib/pulumi/runtime/mocks.py
+++ b/sdk/python/lib/pulumi/runtime/mocks.py
@@ -23,7 +23,7 @@ from typing import Dict, NamedTuple, Optional, Tuple, TYPE_CHECKING
 from google.protobuf import empty_pb2
 from . import rpc
 from .settings import Settings, configure, get_stack, get_project, get_root_resource
-from .sync_await import _sync_await
+from .sync_await import _ensure_event_loop, _sync_await
 from ..runtime.proto import engine_pb2, provider_pb2, resource_pb2
 from ..runtime.stack import Stack, run_pulumi_func
 
@@ -91,6 +91,9 @@ class MockMonitor:
         return "urn:pulumi:" + "::".join([get_stack(), get_project(), type_, name])
 
     def Invoke(self, request):
+        # Ensure we have an event loop on this thread because it's needed when deserializing resource references.
+        _ensure_event_loop()
+
         args = rpc.deserialize_properties(request.args)
 
         if request.tok == "pulumi:pulumi:getResource":
@@ -109,6 +112,9 @@ class MockMonitor:
         return provider_pb2.InvokeResponse(**fields)
 
     def ReadResource(self, request):
+        # Ensure we have an event loop on this thread because it's needed when deserializing resource references.
+        _ensure_event_loop()
+
         state = rpc.deserialize_properties(request.properties)
 
         id_, state = self.mocks.new_resource(request.type, request.name, state, request.provider, request.id)
@@ -127,6 +133,9 @@ class MockMonitor:
         if request.type == "pulumi:pulumi:Stack":
             return resource_pb2.RegisterResourceResponse(urn=urn)
 
+        # Ensure we have an event loop on this thread because it's needed when deserializing resource references.
+        _ensure_event_loop()
+
         inputs = rpc.deserialize_properties(request.object)
 
         id_, state = self.mocks.new_resource(request.type, request.name, inputs, request.provider, request.importId)
diff --git a/sdk/python/lib/pulumi/runtime/sync_await.py b/sdk/python/lib/pulumi/runtime/sync_await.py
index 67a0ff6c7..fd5ca34e9 100644
--- a/sdk/python/lib/pulumi/runtime/sync_await.py
+++ b/sdk/python/lib/pulumi/runtime/sync_await.py
@@ -45,15 +45,7 @@ def _sync_await(awaitable: Awaitable[Any]) -> Any:
     """
 
     # Fetch the current event loop and ensure a future.
-    loop = None
-    try:
-        loop = asyncio.get_event_loop()
-    except RuntimeError:
-        pass
-    if loop is None:
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-
+    loop = _ensure_event_loop()
     fut = asyncio.ensure_future(awaitable)
 
     # If the loop is not running, we can just use run_until_complete. Without this, we would need to duplicate a fair
@@ -93,3 +85,14 @@ def _sync_await(awaitable: Awaitable[Any]) -> Any:
 
     # Return the result of the future.
     return fut.result()
+
+
+def _ensure_event_loop():
+    """Ensures an asyncio event loop exists for the current thread."""
+    loop = None
+    try:
+        loop = asyncio.get_event_loop()
+    except RuntimeError:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+    return loop
diff --git a/sdk/python/lib/test_with_mocks/resources.py b/sdk/python/lib/test_with_mocks/resources.py
index fbb4c0786..c0572a3fa 100644
--- a/sdk/python/lib/test_with_mocks/resources.py
+++ b/sdk/python/lib/test_with_mocks/resources.py
@@ -56,6 +56,11 @@ myinstance = Instance("instance",
 mycustom = MyCustom("mycustom", {"instance": myinstance})
 invoke_result = do_invoke()
 
+# Pass myinstance several more times to ensure deserialization of the resource reference
+# works on other asyncio threads.
+for x in range(5):
+    MyCustom(f"mycustom{x}", {"instance": myinstance})
+
 pulumi.export("hello", "world")
 pulumi.export("outprop", mycomponent.outprop)
 pulumi.export("public_ip", myinstance.public_ip)
diff --git a/sdk/python/lib/test_with_mocks/test_testing_with_mocks.py b/sdk/python/lib/test_with_mocks/test_testing_with_mocks.py
index 020ef1add..82de2d05a 100644
--- a/sdk/python/lib/test_with_mocks/test_testing_with_mocks.py
+++ b/sdk/python/lib/test_with_mocks/test_testing_with_mocks.py
@@ -67,6 +67,9 @@ class TestingWithMocks(unittest.TestCase):
     def test_custom_resource_reference(self):
         def check_instance(instance):
             self.assertIsInstance(instance, resources.Instance)
+            def check_ip(ip):
+                self.assertEqual(ip, '203.0.113.12')
+            instance.public_ip.apply(check_ip)
         return resources.mycustom.instance.apply(check_instance)
 
     @pulumi.runtime.test