Refactor tests

Now that the package is standalone it makes sense to have generic
tests that can then be run for each URI scheme. We also want each
scheme to be tested in a separate test file.

The big test file has been removed.

There are some complications from mypy.

tests/test_file.py

@@ -0,0 +1,152 @@
+# This file is part of lsst-resources.
+#
+# Developed for the LSST Data Management System.
+# This product includes software developed by the LSST Project
+# (https://www.lsst.org).
+# See the COPYRIGHT file at the top-level directory of this distribution
+# for details of code ownership.
+#
+# Use of this source code is governed by a 3-clause BSD-style
+# license that can be found in the LICENSE file.
+
+import os
+import pathlib
+import unittest
+import unittest.mock
+
+from lsst.resources import ResourcePath
+from lsst.resources.tests import GenericReadWriteTestCase, GenericTestCase
+
+TESTDIR = os.path.abspath(os.path.dirname(__file__))
+
+
+class FileTestCase(GenericTestCase, unittest.TestCase):
+    scheme = "file"
+    netloc = "localhost"
+
+    def test_env_var(self):
+        """Test that environment variables are expanded."""
+
+        with unittest.mock.patch.dict(os.environ, {"MY_TEST_DIR": "/a/b/c"}):
+            uri = ResourcePath("${MY_TEST_DIR}/d.txt")
+        self.assertEqual(uri.path, "/a/b/c/d.txt")
+        self.assertEqual(uri.scheme, "file")
+
+        # This will not expand
+        uri = ResourcePath("${MY_TEST_DIR}/d.txt", forceAbsolute=False)
+        self.assertEqual(uri.path, "${MY_TEST_DIR}/d.txt")
+        self.assertFalse(uri.scheme)
+
+    def test_ospath(self):
+        """File URIs have ospath property."""
+
+        file = ResourcePath(self._make_uri("a/test.txt"))
+        self.assertEqual(file.ospath, "/a/test.txt")
+        self.assertEqual(file.ospath, file.path)
+
+        # A Schemeless URI can take unquoted files but will be quoted
+        # when it becomes a file URI.
+        something = "/a#/???.txt"
+        file = ResourcePath(something, forceAbsolute=True)
+        self.assertEqual(file.scheme, "file")
+        self.assertEqual(file.ospath, something, "From URI: {file}")
+        self.assertNotIn("???", file.path)
+
+    def test_path_lib(self):
+        """File URIs can be created from pathlib"""
+        file = ResourcePath(self._make_uri("a/test.txt"))
+
+        path_file = pathlib.Path(file.ospath)
+        from_path = ResourcePath(path_file)
+        self.assertEqual(from_path.ospath, file.ospath)
+
+    def test_schemeless_root(self):
+        root = ResourcePath(self._make_uri("/root"))
+        via_root = ResourcePath("b.txt", root=root)
+        self.assertEqual(via_root.ospath, "/root/b.txt")
+
+        with self.assertRaises(ValueError):
+            # Scheme-less URIs are not allowed to support non-file roots
+            # at the present time. This may change in the future to become
+            # equivalent to ResourcePath.join()
+            ResourcePath("a/b.txt", root=ResourcePath("s3://bucket/a/b/"))
+
+
+class FileReadWriteTestCase(GenericReadWriteTestCase, unittest.TestCase):
+    scheme = "file"
+    netloc = "localhost"
+    testdir = TESTDIR
+    transfer_modes = ("move", "copy", "link", "hardlink", "symlink", "relsymlink")
+
+    def test_transfer_identical(self):
+        """Test overwrite of identical files.
+
+        Only relevant for local files.
+        """
+        dir1 = self.tmpdir.join("dir1", forceDirectory=True)
+        dir1.mkdir()
+        self.assertTrue(dir1.exists())
+        dir2 = self.tmpdir.join("dir2", forceDirectory=True)
+        # A symlink can't include a trailing slash.
+        dir2_ospath = dir2.ospath
+        if dir2_ospath.endswith("/"):
+            dir2_ospath = dir2_ospath[:-1]
+        os.symlink(dir1.ospath, dir2_ospath)
+
+        # Write a test file.
+        src_file = dir1.join("test.txt")
+        content = "0123456"
+        src_file.write(content.encode())
+
+        # Construct URI to destination that should be identical.
+        dest_file = dir2.join("test.txt")
+        self.assertTrue(dest_file.exists())
+        self.assertNotEqual(src_file, dest_file)
+
+        # Transfer it over itself.
+        dest_file.transfer_from(src_file, transfer="symlink", overwrite=True)
+        new_content = dest_file.read().decode()
+        self.assertEqual(content, new_content)
+
+    def test_local_temporary(self):
+        """Create temporary local file if no prefix specified."""
+        with ResourcePath.temporary_uri(suffix=".json") as tmp:
+            self.assertEqual(tmp.getExtension(), ".json", f"uri: {tmp}")
+            self.assertTrue(tmp.isabs(), f"uri: {tmp}")
+            self.assertFalse(tmp.exists(), f"uri: {tmp}")
+            tmp.write(b"abcd")
+            self.assertTrue(tmp.exists(), f"uri: {tmp}")
+            self.assertTrue(tmp.isTemporary)
+            self.assertTrue(tmp.isLocal)
+
+            # If we now ask for a local form of this temporary file
+            # it should still be temporary and it should not be deleted
+            # on exit.
+            with tmp.as_local() as loc:
+                self.assertEqual(tmp, loc)
+                self.assertTrue(loc.isTemporary)
+            self.assertTrue(tmp.exists())
+        self.assertFalse(tmp.exists(), f"uri: {tmp}")
+
+    def test_transfers_from_local(self):
+        """Extra tests for local transfers."""
+
+        target = self.tmpdir.join("a/target.txt")
+        with ResourcePath.temporary_uri() as tmp:
+            tmp.write(b"")
+            self.assertTrue(tmp.isTemporary)
+            for transfer in ("symlink", "hardlink", "link", "relsymlink"):
+                with self.assertRaises(RuntimeError):
+                    target.transfer_from(tmp, transfer)
+
+            # Force the target directory to be created.
+            target.transfer_from(tmp, "move")
+            self.assertFalse(tmp.exists())
+
+            # Temporary file now gone so transfer should not work.
+            with self.assertRaises(FileNotFoundError):
+                target.transfer_from(tmp, "move", overwrite=True)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_http.py

@@ -0,0 +1,241 @@
+# This file is part of lsst-resources.
+#
+# Developed for the LSST Data Management System.
+# This product includes software developed by the LSST Project
+# (https://www.lsst.org).
+# See the COPYRIGHT file at the top-level directory of this distribution
+# for details of code ownership.
+#
+# Use of this source code is governed by a 3-clause BSD-style
+# license that can be found in the LICENSE file.
+
+import os.path
+import unittest
+
+import lsst.resources
+import responses
+from lsst.resources import ResourcePath
+from lsst.resources.tests import GenericTestCase
+from lsst.resources.utils import makeTestTempDir, removeTestTempDir
+
+TESTDIR = os.path.abspath(os.path.dirname(__file__))
+
+
+class GenericHttpTestCase(GenericTestCase, unittest.TestCase):
+    scheme = "http"
+    netloc = "server.example"
+
+
+# Mock required environment variables during tests
+@unittest.mock.patch.dict(
+    os.environ,
+    {
+        "LSST_BUTLER_WEBDAV_AUTH": "TOKEN",
+        "LSST_BUTLER_WEBDAV_TOKEN_FILE": os.path.join(TESTDIR, "data/webdav/token"),
+        "LSST_BUTLER_WEBDAV_CA_BUNDLE": "/path/to/ca/certs",
+    },
+)
+class HttpReadWriteTestCase(unittest.TestCase):
+    """Specialist test cases for WebDAV server.
+
+    The responses class requires that every possible request be explicitly
+    mocked out.  This currently makes it extremely inconvenient to subclass
+    the generic read/write tests shared by other URI schemes.  For now use
+    explicit standalone tests.
+    """
+
+    def setUp(self):
+        # Local test directory
+        self.tmpdir = ResourcePath(makeTestTempDir(TESTDIR))
+
+        serverRoot = "www.not-exists.orgx"
+        existingFolderName = "existingFolder"
+        existingFileName = "existingFile"
+        notExistingFileName = "notExistingFile"
+
+        self.baseURL = ResourcePath(f"https://{serverRoot}", forceDirectory=True)
+        self.existingFileResourcePath = ResourcePath(
+            f"https://{serverRoot}/{existingFolderName}/{existingFileName}"
+        )
+        self.notExistingFileResourcePath = ResourcePath(
+            f"https://{serverRoot}/{existingFolderName}/{notExistingFileName}"
+        )
+        self.existingFolderResourcePath = ResourcePath(
+            f"https://{serverRoot}/{existingFolderName}", forceDirectory=True
+        )
+        self.notExistingFolderResourcePath = ResourcePath(
+            f"https://{serverRoot}/{notExistingFileName}", forceDirectory=True
+        )
+
+        # Need to declare the options
+        responses.add(responses.OPTIONS, self.baseURL.geturl(), status=200, headers={"DAV": "1,2,3"})
+
+        # Used by HttpResourcePath.exists()
+        responses.add(
+            responses.HEAD,
+            self.existingFileResourcePath.geturl(),
+            status=200,
+            headers={"Content-Length": "1024"},
+        )
+        responses.add(responses.HEAD, self.notExistingFileResourcePath.geturl(), status=404)
+
+        # Used by HttpResourcePath.read()
+        responses.add(
+            responses.GET, self.existingFileResourcePath.geturl(), status=200, body=str.encode("It works!")
+        )
+        responses.add(responses.GET, self.notExistingFileResourcePath.geturl(), status=404)
+
+        # Used by HttpResourcePath.write()
+        responses.add(responses.PUT, self.existingFileResourcePath.geturl(), status=201)
+
+        # Used by HttpResourcePath.transfer_from()
+        responses.add(
+            responses.Response(
+                url=self.existingFileResourcePath.geturl(),
+                method="COPY",
+                headers={"Destination": self.existingFileResourcePath.geturl()},
+                status=201,
+            )
+        )
+        responses.add(
+            responses.Response(
+                url=self.existingFileResourcePath.geturl(),
+                method="COPY",
+                headers={"Destination": self.notExistingFileResourcePath.geturl()},
+                status=201,
+            )
+        )
+        responses.add(
+            responses.Response(
+                url=self.existingFileResourcePath.geturl(),
+                method="MOVE",
+                headers={"Destination": self.notExistingFileResourcePath.geturl()},
+                status=201,
+            )
+        )
+
+        # Used by HttpResourcePath.remove()
+        responses.add(responses.DELETE, self.existingFileResourcePath.geturl(), status=200)
+        responses.add(responses.DELETE, self.notExistingFileResourcePath.geturl(), status=404)
+
+        # Used by HttpResourcePath.mkdir()
+        responses.add(
+            responses.HEAD,
+            self.existingFolderResourcePath.geturl(),
+            status=200,
+            headers={"Content-Length": "1024"},
+        )
+        responses.add(responses.HEAD, self.baseURL.geturl(), status=200, headers={"Content-Length": "1024"})
+        responses.add(responses.HEAD, self.notExistingFolderResourcePath.geturl(), status=404)
+        responses.add(
+            responses.Response(url=self.notExistingFolderResourcePath.geturl(), method="MKCOL", status=201)
+        )
+        responses.add(
+            responses.Response(url=self.existingFolderResourcePath.geturl(), method="MKCOL", status=403)
+        )
+
+    def tearDown(self):
+        if self.tmpdir:
+            if self.tmpdir.isLocal:
+                removeTestTempDir(self.tmpdir.ospath)
+
+    @responses.activate
+    def test_exists(self):
+
+        self.assertTrue(self.existingFileResourcePath.exists())
+        self.assertFalse(self.notExistingFileResourcePath.exists())
+
+        self.assertEqual(self.existingFileResourcePath.size(), 1024)
+        with self.assertRaises(FileNotFoundError):
+            self.notExistingFileResourcePath.size()
+
+    @responses.activate
+    def test_remove(self):
+
+        self.assertIsNone(self.existingFileResourcePath.remove())
+        with self.assertRaises(FileNotFoundError):
+            self.notExistingFileResourcePath.remove()
+
+    @responses.activate
+    def test_mkdir(self):
+
+        # The mock means that we can't check this now exists
+        self.notExistingFolderResourcePath.mkdir()
+
+        # This should do nothing
+        self.existingFolderResourcePath.mkdir()
+
+        with self.assertRaises(ValueError):
+            self.notExistingFileResourcePath.mkdir()
+
+    @responses.activate
+    def test_read(self):
+
+        self.assertEqual(self.existingFileResourcePath.read().decode(), "It works!")
+        self.assertNotEqual(self.existingFileResourcePath.read().decode(), "Nope.")
+        with self.assertRaises(FileNotFoundError):
+            self.notExistingFileResourcePath.read()
+
+        # Run this twice to ensure use of cache in code coverag.
+        for _ in (1, 2):
+            with self.existingFileResourcePath.as_local() as local_uri:
+                self.assertTrue(local_uri.isLocal)
+                content = local_uri.read().decode()
+                self.assertEqual(content, "It works!")
+
+        # Check that the environment variable is being read.
+        lsst.resources.http._TMPDIR = None
+        with unittest.mock.patch.dict(os.environ, {"LSST_RESOURCES_TMPDIR": self.tmpdir.ospath}):
+            with self.existingFileResourcePath.as_local() as local_uri:
+                self.assertTrue(local_uri.isLocal)
+                content = local_uri.read().decode()
+                self.assertEqual(content, "It works!")
+                self.assertIsNotNone(local_uri.relative_to(self.tmpdir))
+
+    @responses.activate
+    def test_write(self):
+
+        self.assertIsNone(self.existingFileResourcePath.write(data=str.encode("Some content.")))
+        with self.assertRaises(FileExistsError):
+            self.existingFileResourcePath.write(data=str.encode("Some content."), overwrite=False)
+
+    @responses.activate
+    def test_transfer(self):
+
+        # Transferring to self should be no-op.
+        self.existingFileResourcePath.transfer_from(src=self.existingFileResourcePath)
+
+        self.assertIsNone(self.notExistingFileResourcePath.transfer_from(src=self.existingFileResourcePath))
+        # Should test for existence.
+        # self.assertTrue(self.notExistingFileResourcePath.exists())
+
+        # Should delete and try again with move.
+        # self.notExistingFileResourcePath.remove()
+        self.assertIsNone(
+            self.notExistingFileResourcePath.transfer_from(src=self.existingFileResourcePath, transfer="move")
+        )
+        # Should then check that it was moved.
+        # self.assertFalse(self.existingFileResourcePath.exists())
+
+        # Existing file resource should have been removed so this should
+        # trigger FileNotFoundError.
+        # with self.assertRaises(FileNotFoundError):
+        #    self.notExistingFileResourcePath.transfer_from(src=self.existingFileResourcePath)
+        with self.assertRaises(ValueError):
+            self.notExistingFileResourcePath.transfer_from(
+                src=self.existingFileResourcePath, transfer="unsupported"
+            )
+
+    def test_parent(self):
+
+        self.assertEqual(
+            self.existingFolderResourcePath.geturl(), self.notExistingFileResourcePath.parent().geturl()
+        )
+        self.assertEqual(self.baseURL.geturl(), self.baseURL.parent().geturl())
+        self.assertEqual(
+            self.existingFileResourcePath.parent().geturl(), self.existingFileResourcePath.dirname().geturl()
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()