From a77c9aa34b638bf287c3c8bcfc21cd4d36f80cdb Mon Sep 17 00:00:00 2001 From: Mark Liu Date: Tue, 5 Sep 2017 15:25:04 -0700 Subject: [PATCH] Fix TestUtilsTest that fail on Windows --- .../apache_beam/testing/test_utils_test.py | 34 ++++++++++--------- 1 file changed, 18 insertions(+), 16 deletions(-) diff --git a/sdks/python/apache_beam/testing/test_utils_test.py b/sdks/python/apache_beam/testing/test_utils_test.py index 093e1f175d25..811331079fcb 100644 --- a/sdks/python/apache_beam/testing/test_utils_test.py +++ b/sdks/python/apache_beam/testing/test_utils_test.py @@ -18,11 +18,10 @@ """Unittest for testing utilities,""" import logging +import os import tempfile import unittest -from mock import patch - from apache_beam.io.filesystem import BeamIOError from apache_beam.io.filesystems import FileSystems from apache_beam.testing import test_utils as utils @@ -35,20 +34,23 @@ def setUp(self): self.tmpdir = tempfile.mkdtemp() def test_delete_files_succeeds(self): - f = tempfile.NamedTemporaryFile(dir=self.tmpdir, delete=False) - assert FileSystems.exists(f.name) - utils.delete_files([f.name]) - assert not FileSystems.exists(f.name) - - @patch.object(FileSystems, 'delete', side_effect=BeamIOError('')) - def test_delete_files_fails_with_io_error(self, mocked_delete): - f = tempfile.NamedTemporaryFile(dir=self.tmpdir, delete=False) - assert FileSystems.exists(f.name) - - with self.assertRaises(BeamIOError): - utils.delete_files([f.name]) - self.assertTrue(mocked_delete.called) - self.assertEqual(mocked_delete.call_count, 4) + path = os.path.join(self.tmpdir, 'f1') + + with open(path, 'a') as f: + f.write('test') + + assert FileSystems.exists(path) + utils.delete_files([path]) + assert not FileSystems.exists(path) + + def test_delete_files_fails_with_io_error(self): + path = os.path.join(self.tmpdir, 'f2') + + with self.assertRaises(BeamIOError) as error: + utils.delete_files([path]) + self.assertTrue( + error.exception.message.startswith('Delete operation failed')) + self.assertEqual(error.exception.exception_details.keys(), [path]) def test_delete_files_fails_with_invalid_arg(self): with self.assertRaises(RuntimeError):