Add allow_unsafe_triggers flag to the newly added tests.

Author: shunping

sdks/python/apache_beam/runners/portability/prism_runner_test.py

@@ -36,6 +36,7 @@
 from apache_beam.options.pipeline_options import DebugOptions
 from apache_beam.options.pipeline_options import PortableOptions
 from apache_beam.options.pipeline_options import StandardOptions
+from apache_beam.options.pipeline_options import TypeOptions
 from apache_beam.runners.portability import portable_runner_test
 from apache_beam.runners.portability import prism_runner
 from apache_beam.testing.util import assert_that
@@ -68,6 +69,7 @@ def __init__(self, *args, **kwargs):
     self.environment_config = None
     self.enable_commit = False
     self.streaming = False
+    self.allow_unsafe_triggers = False
 
   def setUp(self):
     self.enable_commit = False
@@ -180,6 +182,8 @@ def create_options(self):
         PortableOptions).environment_options = self.environment_options
 
     options.view_as(StandardOptions).streaming = self.streaming
+    options.view_as(
+        TypeOptions).allow_unsafe_triggers = self.allow_unsafe_triggers
     return options
 
   # Can't read host files from within docker, read a "local" file there.
@@ -234,6 +238,7 @@ def construct_timestamped(k, t):
     return window.TimestampedValue((k, t), t)
 
   def test_after_count_trigger_batch(self):
+    self.allow_unsafe_triggers = True
     with self.create_pipeline() as p:
       result = (
           p
@@ -257,6 +262,7 @@ def test_after_count_trigger_batch(self):
       # yapf: enable
 
   def test_after_count_trigger_streaming(self):
+    self.allow_unsafe_triggers = True
     self.streaming = True
     with self.create_pipeline() as p:
       result = (