docstr: Tests no tasks generated when no inputs, regardless of min_count.
docstr_tok: ["Tests","no","tasks","generated","when","no","inputs",",","regardless","of","min_count","."]
code: def test_tasks_generation_when_no_inputs(self, min_count):
      """Tests no tasks generated when no inputs, regardless of min_count."""
  
      for node in self._pipeline.nodes:
          for v in node.pipeline_node.inputs.inputs.values():
              v.min_count = min_count
  
      # Note that "example gen" tasks will be generated since it has no declared
      # inputs, so it is okay to execute it even when there are no inputs.
      [
          update_example_gen_task,
          update_transform_task,
          update_trainer_task,
      ] = self._generate_and_test(
          use_task_queue=False,
          num_initial_executions=0,
          num_tasks_generated=3,
          num_new_executions=0,
          num_active_executions=0,
          expected_exec_nodes=[],
      )
  
      self.assertIsInstance(
          update_example_gen_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          pstate.NodeState.RUNNING, update_example_gen_task.state
      )
      self.assertIsInstance(
          update_transform_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          pstate.NodeState.STARTED, update_transform_task.state
      )
      self.assertIsInstance(
          update_trainer_task, task_lib.UpdateNodeStateTask
      )
      self.assertEqual(
          pstate.NodeState.STARTED, update_trainer_task.state
      )
  
imports: import os
  from absl.testing import parameterized
  from absl.testing.absltest import mock
  import tensorflow
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import async_pipeline_task_gen
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import service_jobs
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration.experimental.core import task_queue
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_async_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.utils import status
label_desc: Use image node_id 4 for calling the AsyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_tasks_generation_when_no_inputs(min_count) without return types