Remove the confusing "dummy_test_module" directory (spotify#1756)

The purpose is simply to remove a confusing module from the luigi root
directory. It was only used from tests but it couldn't be put in the
test module for whatever technical reasons.

Author: Tarrasch

.coveragerc

@@ -1,6 +1,7 @@
 [report]
 omit =
     luigi/mrrunner.py
+    test/_test_time_generated_module*.py
     */python?.?/*
     */site-packages/nose/*
     *__init__*
@@ -9,7 +10,6 @@ omit =
     */.tox/*
     */setup.py
     */bin/luigidc
-    */dummy_test_module/*
     sitecustomize.py
     hadoop_test.py
     minicluster.py

dummy_test_module/__init__.py

@@ -15,13 +15,13 @@
 # limitations under the License.
 #
 
-from helpers import unittest
+import unittest
 import os
 import luigi
 import luigi.contrib.hdfs
 from luigi import six
 from luigi.mock import MockTarget
-from helpers import with_config
+from helpers import with_config, temporary_unloaded_module
 from luigi.contrib.external_program import ExternalProgramRunError
 from luigi.contrib.spark import SparkSubmitTask, PySparkTask
 from mock import patch, call, MagicMock
@@ -197,7 +197,6 @@ def test_run(self, proc):
         self.assertTrue(os.path.exists(proc_arg_list[7]))
         self.assertTrue(proc_arg_list[8].endswith('TestPySparkTask.pickle'))
 
-    @with_config({'spark': {'py-packages': 'dummy_test_module'}})
     @patch.dict('sys.modules', {'pyspark': MagicMock()})
     @patch('pyspark.SparkContext')
     def test_pyspark_runner(self, spark_context):
@@ -211,7 +210,8 @@ def mock_spark_submit(task):
 
         with patch.object(SparkSubmitTask, 'run', mock_spark_submit):
             job = TestPySparkTask()
-            job.run()
+            with temporary_unloaded_module(b'') as task_module:
+                with_config({'spark': {'py-packages': task_module}})(job.run)()
 
         sc.textFile.assert_called_with('input')
         sc.textFile.return_value.saveAsTextFile.assert_called_with('output')

dummy_test_module/not_imported.py

@@ -15,32 +15,25 @@
 # limitations under the License.
 #
 
-from helpers import LuigiTestCase
+from helpers import LuigiTestCase, temporary_unloaded_module
 
 import luigi
 import luigi.interface
-import tempfile
-import re
 
-_testing_glob_var = None
-
-
-class CmdlineTest(LuigiTestCase):
-
-    def test_dynamic_loading(self):
-        with tempfile.NamedTemporaryFile(dir='test/', prefix="_foo_module", suffix='.py') as temp_module_file:
-            temp_module_file.file.write(b'''
+CONTENTS = b'''
 import luigi
 
 class FooTask(luigi.Task):
     x = luigi.IntParameter()
 
     def run(self):
         luigi._testing_glob_var = self.x
-''')
-            temp_module_file.file.flush()
-            temp_module_path = temp_module_file.name
-            temp_module_name = re.search(r'/(_foo_module.*).py', temp_module_path).group(1)
-            luigi.interface.run(['--module', temp_module_name, 'FooTask', '--x', '123', '--local-scheduler', '--no-lock'])
+'''
 
+
+class CmdlineTest(LuigiTestCase):
+
+    def test_dynamic_loading(self):
+        with temporary_unloaded_module(CONTENTS) as temp_module_name:
+            luigi.interface.run(['--module', temp_module_name, 'FooTask', '--x', '123', '--local-scheduler', '--no-lock'])
             self.assertEqual(luigi._testing_glob_var, 123)

test/contrib/spark_test.py

@@ -17,6 +17,9 @@
 
 import functools
 import itertools
+import tempfile
+import re
+from contextlib import contextmanager
 
 import luigi
 import luigi.task_register
@@ -184,3 +187,21 @@ def wrapper(*args, **kwargs):
 def in_parse(cmds, deferred_computation):
     with CmdlineParser.global_instance(cmds) as cp:
         deferred_computation(cp.get_task_obj())
+
+
+@contextmanager
+def temporary_unloaded_module(python_file_contents):
+    """ Create an importable module
+
+    Return the name of importable module name given its file contents (source
+    code) """
+    with tempfile.NamedTemporaryFile(
+            dir='test/',
+            prefix="_test_time_generated_module",
+            suffix='.py') as temp_module_file:
+        temp_module_file.file.write(python_file_contents)
+        temp_module_file.file.flush()
+        temp_module_path = temp_module_file.name
+        temp_module_name = re.search(r'/(_test_time_generated_module.*).py',
+                                     temp_module_path).group(1)
+        yield temp_module_name

test/dynamic_import_test.py

@@ -25,7 +25,8 @@
 import threading
 import time
 import psutil
-from helpers import unittest, with_config, skipOnTravis, LuigiTestCase
+from helpers import (unittest, with_config, skipOnTravis, LuigiTestCase,
+                     temporary_unloaded_module)
 
 import luigi.notifications
 import luigi.worker
@@ -1098,6 +1099,15 @@ class Dummy3Task(Dummy2Task):
         self.assertEqual(list(self.sch.task_list('FAILED', '').keys()), [d.task_id])
 
     def test_unimported_job_type(self):
+        MODULE_CONTENTS = b'''
+import luigi
+
+
+class UnimportedTask(luigi.Task):
+    def complete(self):
+        return False
+'''
+
         class NotImportedTask(luigi.Task):
             task_family = 'UnimportedTask'
             task_module = None
@@ -1110,10 +1120,10 @@ class NotImportedTask(luigi.Task):
         self.assertEqual(list(self.sch.task_list('FAILED', '').keys()), [task.task_id])
 
         # check that it can import with the right module
-        task.task_module = 'dummy_test_module.not_imported'
-        self.w.add(task)
-        self.assertTrue(self.assistant.run())
-        self.assertEqual(list(self.sch.task_list('DONE', '').keys()), [task.task_id])
+        with temporary_unloaded_module(MODULE_CONTENTS) as task.task_module:
+            self.w.add(task)
+            self.assertTrue(self.assistant.run())
+            self.assertEqual(list(self.sch.task_list('DONE', '').keys()), [task.task_id])
 
 
 class ForkBombTask(luigi.Task):