Revert "[GSoC-273] Feat: Integrate TestPubsubContext to prevent Pub/Sub resource leaks and expand stale cleaner scope"

sdks/python/apache_beam/examples/streaming_wordcount_debugging_it_test.py

@@ -32,7 +32,6 @@
 from apache_beam.testing import test_utils
 from apache_beam.testing.pipeline_verifiers import PipelineStateMatcher
 from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.pubsub_test_context import TestPubsubContext
 
 INPUT_TOPIC = 'wc_topic_input'
 OUTPUT_TOPIC = 'wc_topic_output'
@@ -61,7 +60,6 @@ class StreamingWordcountDebuggingIT(unittest.TestCase):
   def setUp(self):
     self.test_pipeline = TestPipeline(is_integration_test=True)
     self.project = self.test_pipeline.get_option('project')
-    self.pubsub_monitor = TestPubsubContext(project_id=self.project)
     self.setup_pubsub()
 
   def setup_pubsub(self):
@@ -85,10 +83,6 @@ def setup_pubsub(self):
             self.project, OUTPUT_SUB + self.uuid),
         topic=self.output_topic.name,
         ack_deadline_seconds=60)
-    self.pubsub_monitor.register_topic(self.input_topic.name)
-    self.pubsub_monitor.register_topic(self.output_topic.name)
-    self.pubsub_monitor.register_subscription(self.input_sub.name)
-    self.pubsub_monitor.register_subscription(self.output_sub.name)
 
   def _inject_data(self, topic, data):
     """Inject numbers as test data to PubSub."""
@@ -97,8 +91,10 @@ def _inject_data(self, topic, data):
       self.pub_client.publish(self.input_topic.name, str(n).encode('utf-8'))
 
   def tearDown(self):
-    with self.pubsub_monitor:
-      pass
+    test_utils.cleanup_subscriptions(
+        self.sub_client, [self.input_sub, self.output_sub])
+    test_utils.cleanup_topics(
+        self.pub_client, [self.input_topic, self.output_topic])
 
   @pytest.mark.it_postcommit
   @unittest.skip(

sdks/python/apache_beam/examples/streaming_wordcount_it_test.py

@@ -32,7 +32,6 @@
 from apache_beam.testing import test_utils
 from apache_beam.testing.pipeline_verifiers import PipelineStateMatcher
 from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.pubsub_test_context import TestPubsubContext
 
 INPUT_TOPIC = 'wc_topic_input'
 OUTPUT_TOPIC = 'wc_topic_output'
@@ -47,7 +46,6 @@ class StreamingWordCountIT(unittest.TestCase):
   def setUp(self):
     self.test_pipeline = TestPipeline(is_integration_test=True)
     self.project = self.test_pipeline.get_option('project')
-    self.pubsub_monitor = TestPubsubContext(project_id=self.project)
     self.uuid = str(uuid.uuid4())
 
     # Set up PubSub environment.
@@ -68,10 +66,6 @@ def setUp(self):
             self.project, OUTPUT_SUB + self.uuid),
         topic=self.output_topic.name,
         ack_deadline_seconds=60)
-    self.pubsub_monitor.register_topic(self.input_topic.name)
-    self.pubsub_monitor.register_topic(self.output_topic.name)
-    self.pubsub_monitor.register_subscription(self.input_sub.name)
-    self.pubsub_monitor.register_subscription(self.output_sub.name)
 
   def _inject_numbers(self, topic, num_messages):
     """Inject numbers as test data to PubSub."""
@@ -80,8 +74,10 @@ def _inject_numbers(self, topic, num_messages):
       self.pub_client.publish(self.input_topic.name, str(n).encode('utf-8'))
 
   def tearDown(self):
-    with self.pubsub_monitor:
-      pass
+    test_utils.cleanup_subscriptions(
+        self.sub_client, [self.input_sub, self.output_sub])
+    test_utils.cleanup_topics(
+        self.pub_client, [self.input_topic, self.output_topic])
 
   @pytest.mark.it_postcommit
   def test_streaming_wordcount_it(self):

sdks/python/apache_beam/io/gcp/pubsub_integration_test.py

@@ -36,7 +36,6 @@
 from apache_beam.testing import test_utils
 from apache_beam.testing.pipeline_verifiers import PipelineStateMatcher
 from apache_beam.testing.test_pipeline import TestPipeline
-from apache_beam.testing.pubsub_test_context import TestPubsubContext
 
 INPUT_TOPIC = 'psit_topic_input'
 OUTPUT_TOPIC = 'psit_topic_output'
@@ -138,7 +137,7 @@ def setUp(self):
     self.runner_name = type(self.test_pipeline.runner).__name__
     self.project = self.test_pipeline.get_option('project')
     self.uuid = str(uuid.uuid4())
-    self.pubsub_monitor = TestPubsubContext(project_id=self.project)
+
     # Set up PubSub environment.
     from google.cloud import pubsub
     self.pub_client = pubsub.PublisherClient()
@@ -156,19 +155,15 @@ def setUp(self):
         name=self.sub_client.subscription_path(
             self.project, OUTPUT_SUB + self.uuid),
         topic=self.output_topic.name)
-    # Register resources with the monitor immediately upon creation.
-    self.pubsub_monitor_register_topic(self.input_topic.name)
-    self.pubsub_monitor_register_topic(self.output_topic.name)
-    self.pubsub_monitor_register_subscription(self.input_sub.name)
-    self.pubsub_monitor_register_subscription(self.output_sub.name)
     # Add a 30 second sleep after resource creation to ensure subscriptions will
     # receive messages.
     time.sleep(30)
 
   def tearDown(self):
-    # The TestPubsubContext will automatically delete the topics and subscriptions
-    with self.pubsub_monitor:
-      pass
+    test_utils.cleanup_subscriptions(
+        self.sub_client, [self.input_sub, self.output_sub])
+    test_utils.cleanup_topics(
+        self.pub_client, [self.input_topic, self.output_topic])
 
   def _test_streaming(self, with_attributes):
     """Runs IT pipeline with message verifier.
@@ -334,15 +329,13 @@ def test_batch_write_with_ordering_key(self):
     ordering_topic = self.pub_client.create_topic(
         name=self.pub_client.topic_path(
             self.project, 'psit_topic_ordering' + self.uuid))
-    self.pubsub_monitor.register_topic(ordering_topic.name)
     ordering_sub = self.sub_client.create_subscription(
         request=Subscription(
             name=self.sub_client.subscription_path(
                 self.project, 'psit_sub_ordering' + self.uuid),
             topic=ordering_topic.name,
             enable_message_ordering=True,
         ))
-    self.pubsub_monitor.register_subscription(ordering_sub.name)
     time.sleep(10)
 
     try: