class_docstr: Tests paired text data class.
      
class_docstr_tok: ["Tests","paired","text","data","class","."]
code: def _run_and_test(
      self, hparams, proc_shr=False, length_inc=None, discard_src=False
  ):
      # Construct database
      text_data = tx.data.PairedTextData(hparams)
      self.assertEqual(
          text_data.source_vocab.size,
          self._vocab_size + len(text_data.source_vocab.special_tokens),
      )
  
      iterator = text_data.dataset.make_initializable_iterator()
      text_data_batch = iterator.get_next()
  
      with self.test_session() as sess:
          sess.run(tf.global_variables_initializer())
          sess.run(tf.local_variables_initializer())
          sess.run(tf.tables_initializer())
          sess.run(iterator.initializer)
  
          if proc_shr:
              tgt_eos = b"<EOS>"
          else:
              tgt_eos = b"<TARGET_EOS>"
  
          while True:
              try:
                  # Run the logics
                  data_batch_ = sess.run(text_data_batch)
                  self.assertEqual(
                      set(data_batch_.keys()),
                      set(text_data.list_items()),
                  )
                  # Test matching
                  src_text = data_batch_["source_text"]
                  tgt_text = data_batch_["target_text"]
                  if proc_shr:
                      for src, tgt in zip(src_text, tgt_text):
                          np.testing.assert_array_equal(
                              src[:3], tgt[:3]
                          )
                  else:
                      for src, tgt in zip(src_text, tgt_text):
                          np.testing.assert_array_equal(
                              src[:3], tgt[1:4]
                          )
                  self.assertTrue(
                      tgt_eos in data_batch_["target_text"][0]
                  )
  
                  if length_inc:
                      for i in range(len(data_batch_["source_text"])):
                          text_ = data_batch_["source_text"][i].tolist()
                          self.assertEqual(
                              text_.index(b"<EOS>") + 1,
                              data_batch_["source_length"][i]
                              - length_inc[0],
                          )
                      for i in range(len(data_batch_["target_text"])):
                          text_ = data_batch_["target_text"][i].tolist()
                          self.assertEqual(
                              text_.index(tgt_eos) + 1,
                              data_batch_["target_length"][i]
                              - length_inc[1],
                          )
  
                  if discard_src:
                      src_hparams = text_data.hparams.source_dataset
                      max_l = src_hparams.max_seq_length
                      max_l += text_data._decoder[0].added_length
                      for l in data_batch_[
                          text_data.source_length_name
                      ]:
                          self.assertLessEqual(l, max_l)
  
              except tf.errors.OutOfRangeError:
                  print("Done -- epoch limit reached")
                  break
  
file_path: texar/tests/data/data/paired_text_data_test.py
imports: import tempfile
  import copy
  import numpy
  import tensorflow
  import texar.tf
  from texar.tf.data import SpecialTokens
label_desc: Use image node_id 2 for calling the PairedTextDataTest obj's underlying member method code with example usage: obj._run_and_test(hparams, proc_shr, length_inc, discard_src) without return types