Clean up unittest deprecations and warnings

examples/ssh_remote_execution.py

@@ -23,7 +23,7 @@
 
 import luigi
 from luigi.contrib.ssh import RemoteContext, RemoteTarget
-from luigi.mock import MockFile
+from luigi.mock import MockTarget
 
 SSH_HOST = "some.accessible.host"
 
@@ -58,7 +58,7 @@ class ProcessRemoteData(luigi.Task):
  """
  Create a toplist of users based on how many running processes they have on a remote machine.
 
- In this example the processed data is stored in a MockFile.
+ In this example the processed data is stored in a MockTarget.
  """
 
  def requires(self):
@@ -96,4 +96,4 @@ def output(self):
  :return: the target output for this task.
  :rtype: object (:py:class:`~luigi.target.Target`)
  """
- return MockFile("output", mirror_on_stderr=True)
+ return MockTarget("output", mirror_on_stderr=True)

luigi/contrib/sqla.py

@@ -83,11 +83,11 @@ def rows(self):
  from sqlalchemy import String
  import luigi
  from luigi.contrib import sqla
- from luigi.mock import MockFile
+ from luigi.mock import MockTarget
 
  class BaseTask(luigi.Task):
  def output(self):
- return MockFile("BaseTask")
+ return MockTarget("BaseTask")
 
  def run(self):
  out = self.output().open("w")

luigi/lock.py

@@ -23,6 +23,8 @@
 
 import hashlib
 import os
+import sys
+from subprocess import Popen, PIPE
 
 from luigi import six
 
@@ -41,6 +43,20 @@ def getpcmd(pid):
  if lines:
  _, val = lines
  return val
+ elif sys.platform == "darwin":
+ # Use pgrep instead of /proc on macOS.
+ pidfile = ".%d.pid" % (pid, )
+ with open(pidfile, 'w') as f:
+ f.write(str(pid))
+ try:
+ p = Popen(['pgrep', '-lf', '-F', pidfile], stdout=PIPE)
+ stdout, _ = p.communicate()
+ line = stdout.decode('utf8').strip()
+ if line:
+ _, scmd = line.split(' ', 1)
+ return scmd
+ finally:
+ os.unlink(pidfile)
  else:
  # Use the /proc filesystem
  # At least on android there have been some issues with not all
@@ -49,7 +65,10 @@ def getpcmd(pid):
  # https://github.com/spotify/luigi/pull/1876
  try:
  with open('/proc/{0}/cmdline'.format(pid), 'r') as fh:
- return fh.read().replace('\0', ' ').rstrip()
+ if six.PY3:
+ return fh.read().replace('\0', ' ').rstrip()
+ else:
+ return fh.read().replace('\0', ' ').decode('utf8').rstrip()
  except IOError:
  # the system may not allow reading the command line
  # of a process owned by another user
@@ -65,12 +84,7 @@ def get_info(pid_dir, my_pid=None):
  my_pid = os.getpid()
 
  my_cmd = getpcmd(my_pid)
-
- if six.PY3:
- cmd_hash = my_cmd.encode('utf8')
- else:
- cmd_hash = my_cmd
-
+ cmd_hash = my_cmd.encode('utf8')
  pid_file = os.path.join(pid_dir, hashlib.md5(cmd_hash).hexdigest()) + '.pid'
 
  return my_pid, my_cmd, pid_file

test/contrib/bigquery_test.py

@@ -122,15 +122,15 @@ def test_bulk_complete(self):
  TestRunQueryTask.client = client
 
  complete = list(TestRunQueryTask.bulk_complete(parameters))
- self.assertEquals(complete, ['table2'])
+ self.assertEqual(complete, ['table2'])
 
  def test_dataset_doesnt_exist(self):
  client = MagicMock()
  client.dataset_exists.return_value = False
  TestRunQueryTask.client = client
 
  complete = list(TestRunQueryTask.bulk_complete(['table1']))
- self.assertEquals(complete, [])
+ self.assertEqual(complete, [])
 
  def test_query_property(self):
  task = TestRunQueryTask(table='table2')

test/contrib/dataproc_test.py

@@ -75,7 +75,7 @@ def test_3_submit_minimal_job(self):
  .list(projectId=PROJECT_ID, region=REGION, clusterName=CLUSTER_NAME).execute()
  lastJob = response['jobs'][0]['sparkJob']
 
- self.assertEquals(lastJob['mainClass'], "my.MinimalMainClass")
+ self.assertEqual(lastJob['mainClass'], "my.MinimalMainClass")
 
  def test_4_submit_spark_job(self):
  # The job itself will fail because the job files don't exist
@@ -95,9 +95,9 @@ def test_4_submit_spark_job(self):
  .list(projectId=PROJECT_ID, region=REGION, clusterName=CLUSTER_NAME).execute()
  lastJob = response['jobs'][0]['sparkJob']
 
- self.assertEquals(lastJob['mainClass'], "my.MainClass")
- self.assertEquals(lastJob['jarFileUris'], ["one.jar", "two.jar"])
- self.assertEquals(lastJob['args'], ["foo", "bar"])
+ self.assertEqual(lastJob['mainClass'], "my.MainClass")
+ self.assertEqual(lastJob['jarFileUris'], ["one.jar", "two.jar"])
+ self.assertEqual(lastJob['args'], ["foo", "bar"])
 
  def test_5_submit_pyspark_job(self):
  # The job itself will fail because the job files don't exist
@@ -117,9 +117,9 @@ def test_5_submit_pyspark_job(self):
  .list(projectId=PROJECT_ID, region=REGION, clusterName=CLUSTER_NAME).execute()
  lastJob = response['jobs'][0]['pysparkJob']
 
- self.assertEquals(lastJob['mainPythonFileUri'], "main_job.py")
- self.assertEquals(lastJob['pythonFileUris'], ["extra1.py", "extra2.py"])
- self.assertEquals(lastJob['args'], ["foo", "bar"])
+ self.assertEqual(lastJob['mainPythonFileUri'], "main_job.py")
+ self.assertEqual(lastJob['pythonFileUris'], ["extra1.py", "extra2.py"])
+ self.assertEqual(lastJob['args'], ["foo", "bar"])
 
  def test_6_delete_cluster(self):
  success = luigi.run(['--local-scheduler',

test/contrib/external_program_test.py

@@ -197,7 +197,7 @@ def test_venv_is_set_and_prepended_to_path(self, proc):
  self.assertTrue(proc_env['PATH'].startswith('/path/to/venv/bin'))
  self.assertTrue(proc_env['PATH'].endswith('/base/path'))
  self.assertIn('VIRTUAL_ENV', proc_env)
- self.assertEquals(proc_env['VIRTUAL_ENV'], '/path/to/venv')
+ self.assertEqual(proc_env['VIRTUAL_ENV'], '/path/to/venv')
 
  @patch.dict('os.environ', {}, clear=True)
  @patch('luigi.contrib.external_program.subprocess.Popen')

test/contrib/s3_test.py

@@ -288,7 +288,7 @@ def test_get(self):
  tmp_file_path = tmp_file.name
 
  s3_client.get('s3://mybucket/putMe', tmp_file_path)
- self.assertEquals(tmp_file.read(), self.tempFileContents)
+ self.assertEqual(tmp_file.read(), self.tempFileContents)
 
  tmp_file.close()
 
@@ -300,7 +300,7 @@ def test_get_as_string(self):
 
  contents = s3_client.get_as_string('s3://mybucket/putMe')
 
- self.assertEquals(contents, self.tempFileContents)
+ self.assertEqual(contents, self.tempFileContents)
 
  def test_get_key(self):
  s3_client = S3Client(AWS_ACCESS_KEY, AWS_SECRET_KEY)
@@ -438,15 +438,13 @@ def test_copy_empty_file(self):
  """
  self._run_copy_test(self.test_put_multipart_empty_file)
 
- @unittest.skip("bug in moto 0.4.21 causing failure: https://github.com/spulec/moto/issues/526")
  def test_copy_multipart_multiple_parts_non_exact_fit(self):
  """
  Test a multipart copy with two parts, where the parts are not exactly the split size.
  """
  # First, put a file into S3
  self._run_multipart_copy_test(self.test_put_multipart_multiple_parts_non_exact_fit)
 
- @unittest.skip("bug in moto 0.4.21 causing failure: https://github.com/spulec/moto/issues/526")
  def test_copy_multipart_multiple_parts_exact_fit(self):
  """
  Test a multipart copy with multiple parts, where the parts are exactly the split size.

test/contrib/salesforce_test.py

@@ -64,9 +64,13 @@ def mocked_open(*args, **kwargs):
 class TestSalesforceAPI(unittest.TestCase):
  # We patch 'requests.get' with our own method. The mock object is passed in to our test case method.
  @mock.patch('requests.get', side_effect=mocked_requests_get)
- def test_deprecated_results(self, mock_get):
+ def test_deprecated_results_warning(self, mock_get):
  sf = SalesforceAPI('xx', 'xx', 'xx')
- result_id = sf.get_batch_results('job_id', 'batch_id')
+ if PY3:
+ with self.assertWarnsRegex(UserWarning, r'get_batch_results is deprecated'):
+ result_id = sf.get_batch_results('job_id', 'batch_id')
+ else:
+ result_id = sf.get_batch_results('job_id', 'batch_id')
  self.assertEqual('1234', result_id)
 
  @mock.patch('requests.get', side_effect=mocked_requests_get)

test/contrib/sqla_test.py

@@ -29,7 +29,7 @@
 import luigi
 import sqlalchemy
 from luigi.contrib import sqla
-from luigi.mock import MockFile
+from luigi.mock import MockTarget
 from nose.plugins.attrib import attr
 from helpers import skipOnTravis
 
@@ -42,7 +42,7 @@ class BaseTask(luigi.Task):
  TASK_LIST = ["item%d\tproperty%d\n" % (i, i) for i in range(10)]
 
  def output(self):
- return MockFile("BaseTask", mirror_on_stderr=True)
+ return MockTarget("BaseTask", mirror_on_stderr=True)
 
  def run(self):
  out = self.output().open("w")
@@ -268,7 +268,7 @@ def test_column_row_separator(self):
  class ModBaseTask(luigi.Task):
 
  def output(self):
- return MockFile("ModBaseTask", mirror_on_stderr=True)
+ return MockTarget("ModBaseTask", mirror_on_stderr=True)
 
  def run(self):
  out = self.output().open("w")
@@ -302,7 +302,7 @@ def test_update_rows_test(self):
  class ModBaseTask(luigi.Task):
 
  def output(self):
- return MockFile("BaseTask", mirror_on_stderr=True)
+ return MockTarget("BaseTask", mirror_on_stderr=True)
 
  def run(self):
  out = self.output().open("w")

test/date_parameter_test.py

@@ -20,6 +20,7 @@
 
 import luigi
 import luigi.interface
+from luigi import six
 
 
 class DateTask(luigi.Task):
@@ -92,7 +93,11 @@ def test_parse_padding_zero(self):
  self.assertEqual(dm, datetime.datetime(2013, 2, 1, 18, 7, 0))
 
  def test_parse_deprecated(self):
- dm = luigi.DateMinuteParameter().parse('2013-02-01T18H42')
+ if six.PY3:
+ with self.assertWarnsRegex(DeprecationWarning, 'Using "H" between hours and minutes is deprecated, omit it instead.'):
+ dm = luigi.DateMinuteParameter().parse('2013-02-01T18H42')
+ else:
+ dm = luigi.DateMinuteParameter().parse('2013-02-01T18H42')
  self.assertEqual(dm, datetime.datetime(2013, 2, 1, 18, 42, 0))
 
  def test_serialize(self):

test/execution_summary_test.py

@@ -286,7 +286,7 @@ def requires(self):
  for i, line in enumerate(result):
  self.assertEqual(line, expected[i])
 
- @with_config({'execution_summary': {'summary-length': '1'}})
+ @with_config({'execution_summary': {'summary_length': '1'}})
  def test_config_summary_limit(self):
  class Bar(luigi.Task):
  num = luigi.IntParameter()

test/lock_test.py

@@ -50,10 +50,12 @@ def tearDown(self):
  os.rmdir(self.pid_dir)
 
  def test_get_info(self):
- p = subprocess.Popen(["yes", "à我ф"], stdout=subprocess.PIPE)
- pid, cmd, pid_file = luigi.lock.get_info(self.pid_dir, p.pid)
- p.kill()
- self.assertEqual(cmd, 'yes à我ф')
+ try:
+ p = subprocess.Popen(["yes", u"à我ф"], stdout=subprocess.PIPE)
+ pid, cmd, pid_file = luigi.lock.get_info(self.pid_dir, p.pid)
+ finally:
+ p.kill()
+ self.assertEqual(cmd, u'yes à我ф')
 
  def test_acquiring_free_lock(self):
  acquired = luigi.lock.acquire_for(self.pid_dir)

test/mock_test.py

@@ -19,6 +19,7 @@
 from helpers import unittest
 
 from luigi.mock import MockTarget, MockFileSystem
+from luigi import six
 
 
 class MockFileTest(unittest.TestCase):
@@ -98,4 +99,8 @@ class TestImportMockFile(unittest.TestCase):
 
  def test_mockfile(self):
  from luigi.mock import MockFile
- self.assertTrue(isinstance(MockFile('foo'), MockTarget))
+ if six.PY3:
+ with self.assertWarnsRegex(DeprecationWarning, r'MockFile has been renamed MockTarget'):
+ self.assertTrue(isinstance(MockFile('foo'), MockTarget))
+ else:
+ self.assertTrue(isinstance(MockFile('foo'), MockTarget))