Consistently apply sharding suffix to TextFileSink

Previously, the default behavior varied between the Direct and Dataflow
pipeline runner, with only the latter adding -SSSSS-of-NNNNN suffixes.

----Release Notes----
Consistently apply sharding suffix to TextFileSink
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=120885157

Author: robertwb

google/cloud/dataflow/examples/complete/estimate_pi_test.py

@@ -30,11 +30,11 @@ def create_temp_file(self, contents):
       return f.name
 
   def test_basics(self):
-    temp_path = self.create_temp_file('not used')
+    temp_path = self.create_temp_file('result')
     estimate_pi.run([
         '--output=%s' % temp_path])
     # Parse result file and compare.
-    with open(temp_path) as result_file:
+    with open(temp_path + '-00000-of-00001') as result_file:
       estimated_pi = json.loads(result_file.readline())[2]
       # Note: Probabilistically speaking this test can fail with a probability
       # that is very small (VERY) given that we run at least 10 million trials.

google/cloud/dataflow/examples/complete/juliaset/juliaset/juliaset_test.py

@@ -57,7 +57,8 @@ def test_output_file_format(self):
 
     # Parse the results from the file, and ensure it was written in the proper
     # format.
-    with open(self.test_files['output_coord_file_name']) as result_file:
+    with open(self.test_files['output_coord_file_name'] +
+              '-00000-of-00001') as result_file:
       output_lines = result_file.readlines()
 
       # Should have a line for each x-coordinate.

google/cloud/dataflow/examples/complete/tfidf_test.py

@@ -71,7 +71,8 @@ def test_basics(self):
         '--output', os.path.join(temp_folder, 'result')])
     # Parse result file and compare.
     results = []
-    with open(os.path.join(temp_folder, 'result')) as result_file:
+    with open(os.path.join(temp_folder,
+                           'result-00000-of-00001')) as result_file:
       for line in result_file:
         match = re.search(EXPECTED_LINE_RE, line)
         logging.info('Result line: %s', line)

google/cloud/dataflow/examples/cookbook/bigshuffle_test.py

@@ -39,17 +39,18 @@ def test_basics(self):
         '--checksum_output=%s.checksum' % temp_path])
     # Parse result file and compare.
     results = []
-    with open(temp_path + '.result') as result_file:
+    with open(temp_path + '.result-00000-of-00001') as result_file:
       for line in result_file:
         results.append(line.strip())
     expected = self.SAMPLE_TEXT.split('\n')
     self.assertEqual(sorted(results), sorted(expected))
     # Check the checksums
     input_csum = ''
-    with open(temp_path + '.checksum-input') as input_csum_file:
+    with open(temp_path + '.checksum-input-00000-of-00001') as input_csum_file:
       input_csum = input_csum_file.read().strip()
     output_csum = ''
-    with open(temp_path + '.checksum-output') as output_csum_file:
+    with open(temp_path +
+              '.checksum-output-00000-of-00001') as output_csum_file:
       output_csum = output_csum_file.read().strip()
     expected_csum = 'd629c1f6'
     self.assertEqual(input_csum, expected_csum)

google/cloud/dataflow/examples/cookbook/coders_test.py

@@ -42,7 +42,7 @@ def test_basics(self):
         '--output=%s.result' % temp_path])
     # Parse result file and compare.
     results = []
-    with open(temp_path + '.result') as result_file:
+    with open(temp_path + '.result-00000-of-00001') as result_file:
       for line in result_file:
         results.append(json.loads(line))
       logging.info('result: %s', results)

google/cloud/dataflow/examples/cookbook/custom_ptransform_test.py

@@ -42,9 +42,9 @@ def run_pipeline(self, count_implementation, factor=1):
     ])
 
     count_implementation(known_args, PipelineOptions(pipeline_args))
-    self.assertEqual(["""(u'CAT', %d)""" % (3 * factor),
-                      """(u'DOG', %d)""" % (2 * factor)],
-                     self.get_output(output_path))
+    self.assertEqual(["(u'CAT', %d)" % (3 * factor),
+                      "(u'DOG', %d)" % (2 * factor)],
+                     self.get_output(output_path + '-00000-of-00001'))
 
   def create_temp_file(self, contents=''):
     with tempfile.NamedTemporaryFile(delete=False) as f:

google/cloud/dataflow/examples/cookbook/group_with_coder_test.py

@@ -51,7 +51,7 @@ def test_basics_with_type_check(self):
         '--output=%s.result' % temp_path])
     # Parse result file and compare.
     results = []
-    with open(temp_path + '.result') as result_file:
+    with open(temp_path + '.result-00000-of-00001') as result_file:
       for line in result_file:
         name, points = line.split(',')
         results.append((name, int(points)))
@@ -72,7 +72,7 @@ def test_basics_without_type_check(self):
         '--output=%s.result' % temp_path])
     # Parse result file and compare.
     results = []
-    with open(temp_path + '.result') as result_file:
+    with open(temp_path + '.result-00000-of-00001') as result_file:
       for line in result_file:
         name, points = line.split(',')
         results.append((name, int(points)))

google/cloud/dataflow/examples/cookbook/mergecontacts_test.py

@@ -111,7 +111,7 @@ def test_mergecontacts(self):
         '--output_tsv=%s.tsv' % result_prefix,
         '--output_stats=%s.stats' % result_prefix], assert_results=(2, 1, 3))
 
-    with open('%s.tsv' % result_prefix) as f:
+    with open('%s.tsv-00000-of-00001' % result_prefix) as f:
       contents = f.read()
       self.assertEqual(self.EXPECTED_TSV, self.normalize_tsv_results(contents))
 

google/cloud/dataflow/examples/cookbook/multiple_output_pardo_test.py

@@ -52,14 +52,15 @@ def test_multiple_output_pardo(self):
         '--output=%s' % result_prefix])
 
     expected_char_count = len(''.join(self.SAMPLE_TEXT.split('\n')))
-    with open(result_prefix + '-chars') as f:
+    with open(result_prefix + '-chars-00000-of-00001') as f:
       contents = f.read()
       self.assertEqual(expected_char_count, int(contents))
 
-    short_words = self.get_wordcount_results(result_prefix + '-short-words')
+    short_words = self.get_wordcount_results(
+        result_prefix + '-short-words-00000-of-00001')
     self.assertEqual(sorted(short_words), sorted(self.EXPECTED_SHORT_WORDS))
 
-    words = self.get_wordcount_results(result_prefix + '-words')
+    words = self.get_wordcount_results(result_prefix + '-words-00000-of-00001')
     self.assertEqual(sorted(words), sorted(self.EXPECTED_WORDS))
 
 

google/cloud/dataflow/examples/snippets/snippets_test.py

@@ -322,9 +322,9 @@ def create_temp_file(self, contents=''):
       f.write(contents)
       return f.name
 
-  def get_output(self, path, sorted_output=True):
+  def get_output(self, path, sorted_output=True, suffix=''):
     lines = []
-    with open(path) as f:
+    with open(path + '-00000-of-00001' + suffix) as f:
       lines = f.readlines()
     if sorted_output:
       return sorted(s.rstrip('\n') for s in lines)
@@ -365,7 +365,7 @@ def test_model_textio(self):
     snippets.model_textio({'read': temp_path, 'write': result_path})
     self.assertEqual(
         ['aa', 'bb', 'bb', 'cc', 'cc', 'cc'],
-        self.get_output(result_path))
+        self.get_output(result_path, suffix='.csv'))
 
   def test_model_bigqueryio(self):
     # We cannot test BigQueryIO functionality in unit tests therefore we limit