Move and refactor broken hardlink tests

Author: jessebrennan

test/unit/test_dss_client.py

@@ -186,21 +186,6 @@ def _assert_manifest_not_updated(self):
         for row in ManifestDownloadContext._parse_manifest(self.manifest_file):
             self.assertNotIn('file_path', row)
 
-    def _assert_link_fails(self, num_logs, download_func, *args, **kwargs):
-        """Test that download_func(*args, **kwargs) fails to link num_logs times"""
-        os_error = OSError()
-        os_error.errno = errno.EMLINK
-        for error in [os_error, ValueError(), RuntimeError()]:
-            with self.subTest(error=error):
-                with patch('os.link', side_effect=os_error):
-                    from hca.dss.util import log as log_
-                    with self.assertLogs(logger=log_, level=logging.WARNING) as logs:
-                        download_func(*args, **kwargs)
-                for record in logs.records:
-                    self.assertEqual(record.funcName, 'hardlink')
-                    self.assertEqual(record.msg, 'Failed to link source `%s` to destination `%s`; reverting to copying')
-                self.assertEqual(len(logs.records), num_logs)
-
 
 class TestManifestDownloadFilestore(DSSClientTestCase):
 
@@ -358,12 +343,6 @@ def _assert_bundle_json(self, prefix=''):
             actual_hash = hashlib.sha256(f.read()).hexdigest()
         self.assertEqual(actual_hash, expected_hash)
 
-    @unittest.skipIf(os.name is 'nt', 'os.link is not used on Windows')
-    def test_link_fails(self):
-        """If hardlinking fails, we should revert to copy"""
-        num_logs = 9  # 3 fake bundles with 3 data files each
-        self._assert_link_fails(num_logs, self._mock_download_manifest, self.manifest_file, 'aws', layout='bundle')
-
     def test_download_dir_empty(self):
         self._test_download_dir('')
 
@@ -498,11 +477,6 @@ def test_download_dir(self):
     def test_download_dir_dot_dir(self):
         self._test_download_dir(os.path.join('.', 'a_nested_dir'))
 
-    @unittest.skipIf(os.name is 'nt', 'os.link is not used on Windows')
-    def test_link_fails(self):
-        num_logs = 5  # 4 files in bundle, + bundle manifest
-        self._assert_link_fails(num_logs, self._mock_download, 'any_bundle_uuid', 'aws')
-
     @staticmethod
     def _fake_get_collection(collections):
         """Used for mocking :meth:`hca.dss.DSSClient.get_collection`"""

test/unit/util/__init__.py

@@ -1,8 +1,14 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
+import errno
 import os
 import sys
 import unittest
+from unittest.mock import patch
+
+from hca.dss.util import hardlink
+
+from test.unit import TmpDirTestCase
 
 pkg_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..'))  # noqa
 sys.path.insert(0, pkg_root)  # noqa
@@ -17,5 +23,34 @@ def test_dict_merge(self):
         dict3 = {'a': {'b': {'c': 1, 'd': 1}}, 'c': 2, 'd': 2}
         self.assertEqual(hca.util._merge_dict(dict1, dict2), dict3)
 
+
+class TestLinking(TmpDirTestCase):
+    """TmpDirTestCase will ensure any links / files we create are cleaned up"""
+    src = 'link_source'
+    dst = 'link_destination'
+
+    def test_link_limit(self):
+        """Ensure that we copy in the case that the link limit is reached"""
+        os_error = OSError()
+        os_error.errno = errno.EMLINK
+        self._link_with_error(self.src, self.dst, os_error)
+        source_stat = os.stat(self.src)
+        dest_stat = os.stat(self.dst)
+        self.assertTrue(source_stat.st_dev != dest_stat.st_dev or source_stat.st_ino != dest_stat.st_ino)
+
+    def test_link_fails(self):
+        """Ensure that linking fails in other cases"""
+        for error in (ValueError(), RuntimeError()):
+            with self.subTest(error=error):
+                with self.assertRaises(type(error)):
+                    self._link_with_error(self.src, self.dst, error)
+
+    def _link_with_error(self, src, dst, error):
+        with open(src, 'w'):
+            pass
+        with patch('os.link', side_effect=error):
+            hardlink(src, dst)
+
+
 if __name__ == "__main__":
     unittest.main()