Fix dask-distributed tests (#1293)

* Fix dask-distributed tests

This should resolve the failing tests on master.

Fixes #1292

* Use a context manager to ensure file is closed

* Allow clean-up failure on Window

Author: shoyer

xarray/tests/test_distributed.py

@@ -6,7 +6,7 @@
 da = pytest.importorskip('dask.array')
 from distributed.utils_test import cluster, loop
 
-from xarray.tests.test_backends import create_tmp_file
+from xarray.tests.test_backends import create_tmp_file, ON_WINDOWS
 from xarray.tests.test_dataset import create_test_data
 
 from . import assert_allclose, has_scipy, has_netCDF4, has_h5netcdf
@@ -24,11 +24,11 @@
 @pytest.mark.parametrize('engine', ENGINES)
 def test_dask_distributed_integration_test(loop, engine):
     with cluster() as (s, _):
-        with distributed.Client(('127.0.0.1', s['port']), loop=loop):
+        with distributed.Client(s['address'], loop=loop):
             original = create_test_data()
-            with create_tmp_file() as filename:
+            with create_tmp_file(allow_cleanup_failure=ON_WINDOWS) as filename:
                 original.to_netcdf(filename, engine=engine)
-                restored = xr.open_dataset(filename, chunks=3, engine=engine)
-                assert isinstance(restored.var1.data, da.Array)
-                computed = restored.compute()
-                assert_allclose(original, computed)
+                with xr.open_dataset(filename, chunks=3, engine=engine) as restored:
+                    assert isinstance(restored.var1.data, da.Array)
+                    computed = restored.compute()
+                    assert_allclose(original, computed)