addressing comments

.gitignore

@@ -19,7 +19,6 @@ develop-eggs
 .installed.cfg
 lib
 lib64
-venv
 __pycache__
 
 # Installer logs
@@ -51,6 +50,7 @@ docs.metadata
 
 # Virtual environment
 env/
+venv/
 coverage.xml
 sponge_log.xml
 

google/cloud/pubsublite/testing/test_utils.py

@@ -13,7 +13,8 @@
 # limitations under the License.
 
 import asyncio
-from typing import List, Union, Any, TypeVar, Generic, Optional
+import threading
+from typing import List, Union, Any, TypeVar, Generic, Optional, Callable
 
 from asynctest import CoroutineMock
 
@@ -62,3 +63,16 @@ def wire_queues(mock: CoroutineMock) -> QueuePair:
 
 class Box(Generic[T]):
     val: Optional[T]
+
+
+def run_on_thread(func: Callable[[], T]) -> T:
+    box = Box()
+
+    def set_box():
+        box.val = func()
+
+    # Initialize watcher on another thread with a different event loop.
+    thread = threading.Thread(target=set_box)
+    thread.start()
+    thread.join()
+    return box.val

tests/unit/pubsublite/internal/wire/partition_count_watcher_impl_test.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 import asyncio
 import queue
-import threading
 from asynctest.mock import MagicMock
 import pytest
 
@@ -22,8 +21,8 @@
     PartitionCountWatcherImpl,
 )
 from google.cloud.pubsublite.internal.wire.publisher import Publisher
-from google.cloud.pubsublite.testing.test_utils import Box
-from google.cloud.pubsublite.types import Partition, TopicPath, CloudZone, CloudRegion
+from google.cloud.pubsublite.testing.test_utils import run_on_thread
+from google.cloud.pubsublite.types import Partition, TopicPath
 from google.api_core.exceptions import GoogleAPICallError
 
 pytestmark = pytest.mark.asyncio
@@ -36,8 +35,7 @@ def mock_publishers():
 
 @pytest.fixture()
 def topic():
-    zone = CloudZone(region=CloudRegion("a"), zone_id="a")
-    return TopicPath(project_number=1, location=zone, name="c")
+    return TopicPath.parse("projects/1/locations/us-central1-a/topics/topic")
 
 
 @pytest.fixture()
@@ -48,16 +46,7 @@ def mock_admin():
 
 @pytest.fixture()
 def watcher(mock_admin, topic):
-    box = Box()
-
-    def set_box():
-        box.val = PartitionCountWatcherImpl(mock_admin, topic, 0.001)
-
-    # Initialize watcher on another thread with a different event loop.
-    thread = threading.Thread(target=set_box)
-    thread.start()
-    thread.join()
-    return box.val
+    return run_on_thread(lambda: PartitionCountWatcherImpl(mock_admin, topic, 0.001))
 
 
 async def test_init(watcher, mock_admin, topic):

tests/unit/pubsublite/internal/wire/partition_count_watching_publisher_test.py

@@ -11,7 +11,6 @@
 # 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 threading
 from asynctest.mock import MagicMock
 import pytest
 
@@ -23,7 +22,7 @@
 )
 from google.cloud.pubsublite.internal.wire.publisher import Publisher
 from google.cloud.pubsublite.internal.wire.routing_policy import RoutingPolicy
-from google.cloud.pubsublite.testing.test_utils import Box, wire_queues
+from google.cloud.pubsublite.testing.test_utils import wire_queues, run_on_thread
 from google.cloud.pubsublite.types import Partition
 from google.cloud.pubsublite_v1 import PubSubMessage
 from google.api_core.exceptions import GoogleAPICallError
@@ -49,18 +48,11 @@ def mock_watcher():
 
 @pytest.fixture()
 def publisher(mock_watcher, mock_publishers, mock_policies):
-    box = Box()
-
-    def set_box():
-        box.val = PartitionCountWatchingPublisher(
+    return run_on_thread(
+        lambda: PartitionCountWatchingPublisher(
             mock_watcher, lambda p: mock_publishers[p], lambda c: mock_policies[c]
         )
-
-    # Initialize publisher on another thread with a different event loop.
-    thread = threading.Thread(target=set_box)
-    thread.start()
-    thread.join()
-    return box.val
+    )
 
 
 async def test_init(mock_watcher, publisher):