Merge pull request #1759 from davidmarin/error-while-reading-from

fix "error while reading from" in inline mode

mrjob/examples/mr_boom.py

@@ -1,3 +1,16 @@
+# Copyright 2016 Yelp
+#
+# 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
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
 """A job that always fails."""
 from mrjob.job import MRJob
 

mrjob/inline.py

@@ -2,6 +2,7 @@
 # Copyright 2011 Matthew Tai and Yelp
 # Copyright 2012-2016 Yelp and Contributors
 # Copyright 2017 Yelp
+# Copyright 2018 Yelp
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -94,9 +95,9 @@ def invoke_task(stdin, stdout, stderr, wd, env):
                         # read input path from stdin, add to args
                         line = stdin.readline().decode('utf_8')
                         input_uri = line.split('\t')[-1].rstrip()
-                        # input_uri is an absolute path, no need to copy to cwd
-                        input_path = input_uri
-                        args = list(args) + [input_path, input_uri]
+                        # input_uri is an absolute path, can serve
+                        # as path and uri both
+                        args = list(args) + [input_uri, input_uri]
 
                     task = self._mrjob_cls(args)
                     task.sandbox(stdin=stdin, stdout=stdout, stderr=stderr)
@@ -108,7 +109,7 @@ def invoke_task(stdin, stdout, stderr, wd, env):
                     # because then we lose the stacktrace (which is the whole
                     # point of the inline runner)
 
-                    if input_path:  # from manifest
+                    if input_uri:  # from manifest
                         self._error_while_reading_from = input_uri
                     else:
                         self._error_while_reading_from = self._task_input_path(

tests/test_inline.py

@@ -4,6 +4,7 @@
 # Copyright 2013 Yelp and Lyft
 # Copyright 2014 Marc Abramowitz
 # Copyright 2015-2017 Yelp
+# Copyright 2018 Yelp
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -28,13 +29,15 @@
 
 from mrjob.examples.mr_phone_to_url import MRPhoneToURL
 from mrjob.inline import InlineMRJobRunner
+from mrjob.job import MRJob
 
 from tests.examples.test_mr_phone_to_url import write_conversion_record
 from tests.mr_test_cmdenv import MRTestCmdenv
 from tests.mr_two_step_job import MRTwoStepJob
 from tests.sandbox import EmptyMrjobConfTestCase
 from tests.sandbox import SandboxedTestCase
 from tests.job import run_job
+from tests.py2 import patch
 from tests.test_sim import MRIncrementerJob
 
 
@@ -177,3 +180,42 @@ def test_input_manifest(self):
             run_job(MRPhoneToURL(['-r', self.RUNNER, wet2_gz_path, '-']),
                     raw_input=wet1.getvalue()),
             self.EXPECTED_OUTPUT)
+
+
+class MRNope(MRJob):
+    def mapper_init(self):
+        raise NotImplementedError
+
+
+class MRManifestNope(MRJob):
+    def mapper_raw(self, input_path, input_uri):
+        raise NotImplementedError
+
+
+class WhileReadingFromTestCase(SandboxedTestCase):
+    # mostly a regression test for #1758
+
+    def _test_reading_from(self, job_class, expect_input_path):
+        # check that we report the actual input file and not the manifest file
+        input_path = self.makefile('input.txt')
+
+        job = job_class([input_path])
+        job.sandbox()
+
+        log = self.start(patch('mrjob.inline.log'))
+
+        with job.make_runner() as runner:
+            self.assertRaises(NotImplementedError, runner.run)
+
+        error_log = ''.join(a[0][0] for a in log.error.call_args_list)
+
+        if expect_input_path:
+            self.assertIn(input_path, error_log)
+        else:
+            self.assertNotIn(input_path, error_log)
+
+    def test_regular_job(self):
+        self._test_reading_from(MRNope, expect_input_path=False)
+
+    def test_input_manifest(self):
+        self._test_reading_from(MRManifestNope, expect_input_path=True)