Don’t use cached autoconfig client in config tests (#2120)

In hdfs_test.ConfigurationTest, calls to hdfs.get_autoconfig_client
return the version cached in the call to setUp. Because this is called
before the config is set by the helpers.with_config decorator, this
means that the tests aren’t actually testing different configurations.

Luckily, get_autoconfig_client takes an optional argument to specify its
cache. When passing a fresh cache, these will actually test what they’re
supposed to. This PR passes a fresh cache to get_autoconfig_client in
each test.

test/contrib/hdfs_test.py

@@ -19,6 +19,7 @@
 import re
 from helpers import unittest
 import random
+import threading
 import pickle
 
 import helpers
@@ -27,6 +28,9 @@
 import luigi.format
 from luigi.contrib import hdfs
 from luigi import six
+from luigi.contrib.hdfs import SnakebiteHdfsClient
+from luigi.contrib.hdfs.hadoopcli_clients import HdfsClient
+from luigi.contrib.target import CascadingClient
 from minicluster import MiniClusterTestCase
 from nose.plugins.attrib import attr
 import luigi.contrib.hdfs.clients
@@ -75,17 +79,22 @@ def test_when_not_specified(self):
 
     @helpers.with_config({"hdfs": {"client": "hadoopcli"}})
     def test_hadoopcli(self):
-        client = hdfs.get_autoconfig_client()
+        client = hdfs.get_autoconfig_client(threading.local())
+        self.assertTrue(isinstance(client, HdfsClient))
         self.tezt_rename_dont_move(client)
 
+    @unittest.skipIf(six.PY3, "snakebite doesn't work on Python 3 yet.")
     @helpers.with_config({"hdfs": {"client": "snakebite"}})
     def test_snakebite(self):
-        client = hdfs.get_autoconfig_client()
+        client = hdfs.get_autoconfig_client(threading.local())
+        self.assertTrue(isinstance(client, SnakebiteHdfsClient))
         self.tezt_rename_dont_move(client)
 
+    @unittest.skipIf(six.PY3, "snakebite doesn't work on Python 3 yet.")
     @helpers.with_config({"hdfs": {"client": "snakebite_with_hadoopcli_fallback"}})
     def test_snakebite_with_hadoopcli_fallback(self):
-        client = hdfs.get_autoconfig_client()
+        client = hdfs.get_autoconfig_client(threading.local())
+        self.assertTrue(isinstance(client, CascadingClient))
         self.tezt_rename_dont_move(client)