tensorflow/tensorflow

File random_shuffle_queue_test.py has 1057 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# Copyright 2015 The TensorFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

RandomShuffleQueueTest has 64 functions (exceeds 20 allowed). Consider refactoring.

Open

class RandomShuffleQueueTest(test.TestCase):

  def setUp(self):
    # Useful for debugging when a test times out.
    super(RandomShuffleQueueTest, self).setUp()

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testDequeueUpToInDifferentOrders(self):
    with self.cached_session():
      # Specify seeds to make the test deterministic
      # (https://en.wikipedia.org/wiki/Taxicab_number).
      q1 = data_flow_ops.RandomShuffleQueue(

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testDequeueManyInDifferentOrders(self):
    with self.cached_session():
      # Specify seeds to make the test deterministic
      # (https://en.wikipedia.org/wiki/Taxicab_number).
      q1 = data_flow_ops.RandomShuffleQueue(

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testDequeueUpToWithTensorParameter(self):
    with self.cached_session():
      # Define a first queue that contains integer counts.
      dequeue_counts = [random.randint(1, 10) for _ in range(100)]
      count_q = data_flow_ops.RandomShuffleQueue(100, 0, dtypes_lib.int32)

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testDequeueManyWithTensorParameter(self):
    with self.cached_session():
      # Define a first queue that contains integer counts.
      dequeue_counts = [random.randint(1, 10) for _ in range(100)]
      count_q = data_flow_ops.RandomShuffleQueue(100, 0, dtypes_lib.int32)

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testBlockingDequeueMany(self):
    with self.cached_session() as sess:
      q = data_flow_ops.RandomShuffleQueue(10, 0, dtypes_lib.float32, ((),))
      elems = [10.0, 20.0, 30.0, 40.0]
      enqueue_op = q.enqueue_many((elems,))

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testBlockingDequeueUpTo(self):
    with self.cached_session() as sess:
      q = data_flow_ops.RandomShuffleQueue(10, 0, dtypes_lib.float32, ((),))
      elems = [10.0, 20.0, 30.0, 40.0]
      enqueue_op = q.enqueue_many((elems,))

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testParallelDequeueUpTo(self):
    with self.cached_session() as sess:
      q = data_flow_ops.RandomShuffleQueue(
          1000, 0, dtypes_lib.float32, shapes=())
      elems = [10.0 * x for x in range(1000)]

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testParallelDequeueMany(self):
    with self.cached_session() as sess:
      q = data_flow_ops.RandomShuffleQueue(
          1000, 0, dtypes_lib.float32, shapes=())
      elems = [10.0 * x for x in range(1000)]

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testBlockingDequeueManyFromClosedEmptyQueue(self):
    with self.cached_session() as sess:
      q = data_flow_ops.RandomShuffleQueue(10, 5, dtypes_lib.float32, ((),))
      close_op = q.close()
      dequeued_t = q.dequeue_many(4)

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testBlockingDequeueUpToFromClosedEmptyQueue(self):
    with self.cached_session() as sess:
      q = data_flow_ops.RandomShuffleQueue(10, 5, dtypes_lib.float32, ((),))
      close_op = q.close()
      dequeued_t = q.dequeue_up_to(4)

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testDequeueUpToNoBlocking(self):
    with self.cached_session():
      q = data_flow_ops.RandomShuffleQueue(10, 0, dtypes_lib.float32, ((),))
      elems = [10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0]
      enqueue_op = q.enqueue_many((elems,))

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testDequeueMany(self):
    with self.cached_session():
      q = data_flow_ops.RandomShuffleQueue(10, 0, dtypes_lib.float32, ((),))
      elems = [10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0]
      enqueue_op = q.enqueue_many((elems,))

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testEmptyDequeueManyWithNoShape(self):
    with self.cached_session():
      q = data_flow_ops.RandomShuffleQueue(10, 0, dtypes_lib.float32)
      enqueue_op = q.enqueue((constant_op.constant(
          [10.0, 20.0], shape=(1, 2)),))

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testEmptyDequeueUpToWithNoShape(self):
    with self.cached_session():
      q = data_flow_ops.RandomShuffleQueue(10, 0, dtypes_lib.float32)
      enqueue_op = q.enqueue((constant_op.constant(
          [10.0, 20.0], shape=(1, 2)),))

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testEmptyDequeueUpTo(self):
    with self.cached_session():
      q = data_flow_ops.RandomShuffleQueue(10, 0, dtypes_lib.float32, shapes=())
      enqueue_op = q.enqueue((10.0,))
      dequeued_t = q.dequeue_up_to(0)

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

  def testEmptyDequeueMany(self):
    with self.cached_session():
      q = data_flow_ops.RandomShuffleQueue(10, 0, dtypes_lib.float32, shapes=())
      enqueue_op = q.enqueue((10.0,))
      dequeued_t = q.dequeue_many(0)

• Read up
  Read up

Similar blocks of code found in 3 locations. Consider refactoring.

Open

      for _ in range(20):
        index = np.random.randint(num_queues)
        q = data_flow_ops.RandomShuffleQueue.from_list(index, qlist)
        q.enqueue((10.,)).run()
        self.assertEqual(q.dequeue().eval(), 10.0)

• Read up
  Read up

Identical blocks of code found in 2 locations. Consider refactoring.

Open

      def dequeue():
        results.extend(self.evaluate(dequeued_t))
        self.assertEqual(3, len(results))
        # min_after_dequeue is 2, we ask for 3 elements, and we end up only
        # getting the remaining 1.

• Read up
  Read up

Identical blocks of code found in 2 locations. Consider refactoring.

Open

      def dequeue():
        results.extend(self.evaluate(dequeued_t))
        self.assertEqual(3, len(results))
        results.extend(self.evaluate(dequeued_t))
        self.assertEqual(4, len(results))

• Read up
  Read up

Identical blocks of code found in 4 locations. Consider refactoring.

Open

      elems = np.array([[[[[x] * 4] * 4] * 4] * 4 for x in range(10)], np.int32)

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

      for _ in range(5):
        results[2].append(deq1.eval())
        results[3].append(deq2.eval())

• Read up
  Read up

Similar blocks of code found in 2 locations. Consider refactoring.

Open

      for _ in range(5):
        results[0].append(deq1.eval())
        results[1].append(deq2.eval())

• Read up
  Read up

Similar blocks of code found in 7 locations. Consider refactoring.

Open

      int_elems = [[1, 2], [3, 4], [5, 6], [7, 8], [9, 10], [11, 12], [13, 14],
                   [15, 16], [17, 18], [19, 20]]

• Read up
  Read up

Similar blocks of code found in 7 locations. Consider refactoring.

Open

      int_elems = [[1, 2], [3, 4], [5, 6], [7, 8], [9, 10], [11, 12], [13, 14],
                   [15, 16], [17, 18], [19, 20]]

• Read up
  Read up

Identical blocks of code found in 2 locations. Consider refactoring.

Open

      q.enqueue_many([[1, 2, 3, 4], [[1, 1], [2, 2], [3, 3], [4, 4]]]).run()

• Read up
  Read up

Identical blocks of code found in 3 locations. Consider refactoring.

Open

      def dequeue():
        for _ in range(len(elems)):
          results.append(self.evaluate(dequeued_t))

• Read up
  Read up