tests: migrate unittest to pytest

tests/basic_env.py

@@ -155,12 +155,6 @@ class TestDir(TestDirFixture, TestCase):
         TestCase.__init__(self, methodName)
 
 
-class TestGit(TestGitFixture, TestCase):
-    def __init__(self, methodName):
-        TestGitFixture.__init__(self)
-        TestCase.__init__(self, methodName)
-
-
 class TestDvc(TestDvcFixture, TestCase):
     def __init__(self, methodName):
         TestDvcFixture.__init__(self)

tests/func/test_odb.py

@@ -9,176 +9,168 @@ from dvc.data.db import ODBManager
 from dvc.hash_info import HashInfo
 from dvc.objects.errors import ObjectFormatError
 from dvc.utils import relpath
-from tests.basic_env import TestDir, TestDvc
 
 
-class TestCache(TestDvc):
-    def setUp(self):
-        super().setUp()
-        self.cache1_md5 = "123"
-        self.cache2_md5 = "234"
-        self.cache1 = os.path.join(
-            self.dvc.odb.local.cache_dir,
-            self.cache1_md5[0:2],
-            self.cache1_md5[2:],
-        )
-        self.cache2 = os.path.join(
-            self.dvc.odb.local.cache_dir,
-            self.cache2_md5[0:2],
-            self.cache2_md5[2:],
-        )
-        self.create(self.cache1, "1")
-        self.create(self.cache2, "2")
+def test_cache(tmp_dir, dvc):
+    cache1_md5 = "123"
+    cache2_md5 = "234"
+    cache1 = os.path.join(
+        dvc.odb.local.cache_dir,
+        cache1_md5[0:2],
+        cache1_md5[2:],
+    )
+    cache2 = os.path.join(
+        dvc.odb.local.cache_dir,
+        cache2_md5[0:2],
+        cache2_md5[2:],
+    )
+    tmp_dir.gen({cache1: "1", cache2: "2"})
 
-    def test_all(self):
-        md5_list = list(ODBManager(self.dvc).local.all())
-        self.assertEqual(len(md5_list), 2)
-        self.assertIn(self.cache1_md5, md5_list)
-        self.assertIn(self.cache2_md5, md5_list)
+    assert os.path.exists(cache1)
+    assert os.path.exists(cache2)
 
-    def test_get(self):
-        cache = ODBManager(self.dvc).local.hash_to_path(self.cache1_md5)
-        self.assertEqual(os.fspath(cache), self.cache1)
+    odb = ODBManager(dvc)
+
+    md5_list = list(odb.local.all())
+    assert len(md5_list) == 2
+    assert cache1_md5 in md5_list
+    assert cache2_md5 in md5_list
+
+    odb_cache1 = odb.local.hash_to_path(cache1_md5)
+    odb_cache2 = odb.local.hash_to_path(cache2_md5)
+    assert os.fspath(odb_cache1) == cache1
+    assert os.fspath(odb_cache2) == cache2
 
 
-class TestCacheLoadBadDirCache(TestDvc):
-    def _do_test(self, ret):
-        self.assertTrue(isinstance(ret, list))
-        self.assertEqual(len(ret), 0)
+def test_cache_load_bad_dir_cache(tmp_dir, dvc):
+    from dvc.data import load
 
-    def test(self):
-        from dvc.data import load
+    dir_hash = "123.dir"
+    fname = os.fspath(dvc.odb.local.hash_to_path(dir_hash))
+    tmp_dir.gen({fname: "<clearly>not,json"})
+    with pytest.raises(ObjectFormatError):
+        load(dvc.odb.local, HashInfo("md5", dir_hash))
 
-        dir_hash = "123.dir"
-        fname = os.fspath(self.dvc.odb.local.hash_to_path(dir_hash))
-        self.create(fname, "<clearly>not,json")
-        with pytest.raises(ObjectFormatError):
-            load(self.dvc.odb.local, HashInfo("md5", dir_hash))
-
-        dir_hash = "234.dir"
-        fname = os.fspath(self.dvc.odb.local.hash_to_path(dir_hash))
-        self.create(fname, '{"a": "b"}')
-        with pytest.raises(ObjectFormatError):
-            load(self.dvc.odb.local, HashInfo("md5", dir_hash))
+    dir_hash = "234.dir"
+    fname = os.fspath(dvc.odb.local.hash_to_path(dir_hash))
+    tmp_dir.gen({fname: '{"a": "b"}'})
+    with pytest.raises(ObjectFormatError):
+        load(dvc.odb.local, HashInfo("md5", dir_hash))
 
 
-class TestExternalCacheDir(TestDvc):
-    def test(self):
-        cache_dir = TestDvc.mkdtemp()
+def test_external_cache_dir(tmp_dir, dvc, make_tmp_dir):
+    cache_dir = make_tmp_dir("cache")
 
-        ret = main(["config", "cache.dir", cache_dir])
-        self.assertEqual(ret, 0)
+    with dvc.config.edit() as conf:
+        conf["cache"]["dir"] = cache_dir.fs_path
+    assert not os.path.exists(dvc.odb.local.cache_dir)
+    dvc.odb = ODBManager(dvc)
 
-        self.assertFalse(os.path.exists(self.dvc.odb.local.cache_dir))
+    tmp_dir.dvc_gen({"foo": "foo"})
 
-        ret = main(["add", self.FOO])
-        self.assertEqual(ret, 0)
+    tmp_dir.dvc_gen(
+        {
+            "data_dir": {
+                "data": "data_dir/data",
+                "data_sub_dir": {"data_sub": "data_dir/data_sub_dir/data_sub"},
+            }
+        }
+    )
 
-        ret = main(["add", self.DATA_DIR])
-        self.assertEqual(ret, 0)
-
-        self.assertFalse(os.path.exists(".dvc/cache"))
-        self.assertNotEqual(len(os.listdir(cache_dir)), 0)
-
-    def test_remote_references(self):
-        ssh_url = "ssh://user@localhost:23"
-        assert main(["remote", "add", "storage", ssh_url]) == 0
-        assert main(["remote", "add", "cache", "remote://storage/tmp"]) == 0
-        assert main(["config", "cache.ssh", "cache"]) == 0
-
-        self.dvc.__init__()
-
-        assert self.dvc.odb.ssh.fs_path == "/tmp"
+    assert not os.path.exists(".dvc/cache")
+    assert len(os.listdir(cache_dir)) != 0
 
 
-class TestSharedCacheDir(TestDir):
-    def test(self):
-        cache_dir = os.path.abspath(os.path.join(os.curdir, "cache"))
-        for d in ["dir1", "dir2"]:
-            os.mkdir(d)
-            os.chdir(d)
+def test_remote_cache_references(tmp_dir, dvc):
+    with dvc.config.edit() as conf:
+        conf["remote"]["storage"] = {"url": "ssh://user@localhost:23"}
+        conf["remote"]["cache"] = {"url": "remote://storage/tmp"}
+        conf["cache"]["ssh"] = "cache"
 
+    dvc.odb = ODBManager(dvc)
+
+    assert dvc.odb.ssh.fs_path == "/tmp"
+
+
+def test_shared_cache_dir(tmp_dir):
+    cache_dir = os.path.abspath(os.path.join(os.curdir, "cache"))
+    for d in ["dir1", "dir2"]:
+        os.mkdir(d)
+        with (tmp_dir / d).chdir():
             ret = main(["init", "--no-scm"])
-            self.assertEqual(ret, 0)
+            assert ret == 0
 
             ret = main(["config", "cache.dir", cache_dir])
-            self.assertEqual(ret, 0)
+            assert ret == 0
 
-            self.assertFalse(os.path.exists(os.path.join(".dvc", "cache")))
+            assert not os.path.exists(os.path.join(".dvc", "cache"))
 
-            with open("common", "w+", encoding="utf-8") as fd:
-                fd.write("common")
-
-            with open("unique", "w+", encoding="utf-8") as fd:
-                fd.write(d)
+            (tmp_dir / d).gen({"common": "common", "unique": d})
 
             ret = main(["add", "common", "unique"])
-            self.assertEqual(ret, 0)
+            assert ret == 0
 
-            os.chdir("..")
+    assert not os.path.exists(os.path.join("dir1", ".dvc", "cache"))
+    assert not os.path.exists(os.path.join("dir2", ".dvc", "cache"))
 
-        self.assertFalse(os.path.exists(os.path.join("dir1", ".dvc", "cache")))
-        self.assertFalse(os.path.exists(os.path.join("dir2", ".dvc", "cache")))
-
-        subdirs = list(
-            filter(
-                lambda x: os.path.isdir(os.path.join(cache_dir, x)),
-                os.listdir(cache_dir),
-            )
-        )
-        self.assertEqual(len(subdirs), 3)
-        self.assertEqual(
-            len(os.listdir(os.path.join(cache_dir, subdirs[0]))), 1
-        )
-        self.assertEqual(
-            len(os.listdir(os.path.join(cache_dir, subdirs[1]))), 1
-        )
-        self.assertEqual(
-            len(os.listdir(os.path.join(cache_dir, subdirs[2]))), 1
+    subdirs = list(
+        filter(
+            lambda x: os.path.isdir(os.path.join(cache_dir, x)),
+            os.listdir(cache_dir),
         )
+    )
+    assert len(subdirs) == 3
+    assert len(os.listdir(os.path.join(cache_dir, subdirs[0]))) == 1
+
+    assert len(os.listdir(os.path.join(cache_dir, subdirs[1]))) == 1
+    assert len(os.listdir(os.path.join(cache_dir, subdirs[2]))) == 1
 
 
-class TestCacheLinkType(TestDvc):
-    def test(self):
-        ret = main(["config", "cache.type", "reflink,copy"])
-        self.assertEqual(ret, 0)
+def test_cache_link_type(tmp_dir, scm, dvc):
+    with dvc.config.edit() as conf:
+        conf["cache"]["type"] = "reflink,copy"
+    dvc.odb = ODBManager(dvc)
 
-        ret = main(["add", self.FOO])
-        self.assertEqual(ret, 0)
+    stages = tmp_dir.dvc_gen({"foo": "foo"})
+    assert len(stages) == 1
+    assert (tmp_dir / "foo").read_text().strip() == "foo"
 
 
-class TestCmdCacheDir(TestDvc):
-    def test(self):
-        ret = main(["cache", "dir"])
-        self.assertEqual(ret, 0)
+def test_cmd_cache_dir(tmp_dir, scm, dvc):
+    ret = main(["cache", "dir"])
+    assert ret == 0
 
-    def test_abs_path(self):
-        dname = os.path.join(os.path.dirname(self._root_dir), "dir")
-        ret = main(["cache", "dir", dname])
-        self.assertEqual(ret, 0)
 
-        config = configobj.ConfigObj(self.dvc.config.files["repo"])
-        self.assertEqual(config["cache"]["dir"], dname)
+def test_cmd_cache_abs_path(tmp_dir, scm, dvc, make_tmp_dir):
+    cache_dir = make_tmp_dir("cache")
+    ret = main(["cache", "dir", cache_dir.fs_path])
+    assert ret == 0
 
-    def test_relative_path(self):
-        tmpdir = os.path.realpath(self.mkdtemp())
-        dname = relpath(tmpdir)
-        ret = main(["cache", "dir", dname])
-        self.assertEqual(ret, 0)
+    config = configobj.ConfigObj(dvc.config.files["repo"])
+    assert config["cache"]["dir"] == cache_dir.fs_path
 
-        # NOTE: we are in the repo's root and config is in .dvc/, so
-        # dir path written to config should be just one level above.
-        rel = os.path.join("..", dname)
-        config = configobj.ConfigObj(self.dvc.config.files["repo"])
-        self.assertEqual(config["cache"]["dir"], rel.replace("\\", "/"))
 
-        ret = main(["add", self.FOO])
-        self.assertEqual(ret, 0)
+def test_cmd_cache_relative_path(tmp_dir, scm, dvc, make_tmp_dir):
+    cache_dir = make_tmp_dir("cache")
+    dname = relpath(cache_dir)
+    ret = main(["cache", "dir", dname])
+    assert ret == 0
 
-        subdirs = os.listdir(tmpdir)
-        self.assertEqual(len(subdirs), 1)
-        files = os.listdir(os.path.join(tmpdir, subdirs[0]))
-        self.assertEqual(len(files), 1)
+    dvc.config.load()
+    dvc.odb = ODBManager(dvc)
+
+    # NOTE: we are in the repo's root and config is in .dvc/, so
+    # dir path written to config should be just one level above.
+    rel = os.path.join("..", dname)
+    config = configobj.ConfigObj(dvc.config.files["repo"])
+    assert config["cache"]["dir"] == rel.replace("\\", "/")
+
+    tmp_dir.dvc_gen({"foo": "foo"})
+
+    subdirs = os.listdir(cache_dir)
+    assert len(subdirs) == 1
+    files = os.listdir(os.path.join(cache_dir, subdirs[0]))
+    assert len(files) == 1
 
 
 def test_default_cache_type(dvc):

tests/unit/test_context.py

@@ -1,6 +1,5 @@
 from dataclasses import asdict
 from math import pi
-from unittest.mock import mock_open
 
 import pytest
 
@@ -219,7 +218,7 @@ def test_overwrite_with_setitem():
 def test_load_from(mocker):
     d = {"x": {"y": {"z": 5}, "lst": [1, 2, 3]}, "foo": "foo"}
     fs = mocker.Mock(
-        open=mock_open(read_data=dumps_yaml(d)),
+        open=mocker.mock_open(read_data=dumps_yaml(d)),
         **{"exists.return_value": True, "isdir.return_value": False},
     )
     file = "params.yaml"

tests/unit/utils/test_fs.py

@@ -1,7 +1,5 @@
 import filecmp
 import os
-from unittest import TestCase
-from unittest.mock import patch
 
 import pytest
 
@@ -22,74 +20,77 @@ from dvc.utils.fs import (
     remove,
     walk_files,
 )
-from tests.basic_env import TestDir
 
 
-class TestMtimeAndSize(TestDir):
-    def test(self):
-        fs = LocalFileSystem(url=self.root_dir)
-        file_time, file_size = get_mtime_and_size(self.DATA, fs)
-        dir_time, dir_size = get_mtime_and_size(self.DATA_DIR, fs)
+def test_mtime_and_size(tmp_dir):
+    tmp_dir.gen({"dir/data": "data", "dir/subdir/subdata": "subdata"})
+    fs = LocalFileSystem(url=tmp_dir)
+    file_time, file_size = get_mtime_and_size("dir/data", fs)
+    dir_time, dir_size = get_mtime_and_size("dir", fs)
 
-        actual_file_size = os.path.getsize(self.DATA)
-        actual_dir_size = os.path.getsize(self.DATA) + os.path.getsize(
-            self.DATA_SUB
-        )
+    actual_file_size = os.path.getsize("dir/data")
+    actual_dir_size = os.path.getsize("dir/data") + os.path.getsize(
+        "dir/subdir/subdata"
+    )
 
-        self.assertIs(type(file_time), str)
-        self.assertIs(type(file_size), int)
-        self.assertEqual(file_size, actual_file_size)
-        self.assertIs(type(dir_time), str)
-        self.assertIs(type(dir_size), int)
-        self.assertEqual(dir_size, actual_dir_size)
+    assert isinstance(file_time, str)
+    assert isinstance(file_size, int)
+    assert file_size == actual_file_size
+    assert isinstance(dir_time, str)
+    assert isinstance(dir_size, int)
+    assert dir_size == actual_dir_size
 
 
-class TestContainsLink(TestCase):
-    def test_should_raise_exception_on_base_path_not_in_path(self):
-        with self.assertRaises(BasePathNotInCheckedPathException):
-            contains_symlink_up_to(os.path.join("foo", "path"), "bar")
+def test_should_raise_exception_on_base_path_not_in_path():
+    with pytest.raises(BasePathNotInCheckedPathException):
+        contains_symlink_up_to(os.path.join("foo", "path"), "bar")
 
-    @patch.object(System, "is_symlink", return_value=True)
-    def test_should_return_true_on_symlink_in_path(self, _):
-        base_path = "foo"
-        path = os.path.join(base_path, "bar")
-        self.assertTrue(contains_symlink_up_to(path, base_path))
 
-    @patch.object(System, "is_symlink", return_value=False)
-    def test_should_return_false_on_path_eq_to_base_path(self, _):
-        path = "path"
-        self.assertFalse(contains_symlink_up_to(path, path))
+def test_should_return_true_on_symlink_in_path(mocker):
+    mocker.patch.object(System, "is_symlink", return_value=True)
+    base_path = "foo"
+    path = os.path.join(base_path, "bar")
+    assert contains_symlink_up_to(path, base_path)
 
-    @patch.object(System, "is_symlink", return_value=False)
-    @patch.object(os.path, "dirname", side_effect=lambda arg: arg)
-    def test_should_return_false_on_no_more_dirs_below_path(
-        self, dirname_patch, _
-    ):
-        self.assertFalse(
-            contains_symlink_up_to(os.path.join("foo", "path"), "foo")
-        )
-        dirname_patch.assert_called_once()
 
-    @patch.object(System, "is_symlink", return_value=True)
-    def test_should_return_false_when_base_path_is_symlink(self, _):
-        base_path = "foo"
-        target_path = os.path.join(base_path, "bar")
+def test_should_return_false_on_path_eq_to_base_path(mocker):
+    mocker.patch.object(System, "is_symlink", return_value=False)
+    path = "path"
+    assert not contains_symlink_up_to(path, path)
 
-        def base_path_is_symlink(path):
-            if path == base_path:
-                return True
-            return False
 
-        with patch.object(
-            System, "is_symlink", side_effect=base_path_is_symlink
-        ):
-            self.assertFalse(contains_symlink_up_to(target_path, base_path))
+def test_should_return_false_on_no_more_dirs_below_path(mocker):
+    mocker.patch.object(System, "is_symlink", return_value=False)
+    dirname_patch = mocker.patch.object(
+        os.path, "dirname", side_effect=lambda arg: arg
+    )
+    assert not contains_symlink_up_to(os.path.join("foo", "path"), "foo")
+    dirname_patch.assert_called_once()
 
-    def test_path_object_and_str_are_valid_arg_types(self):
-        base_path = "foo"
-        target_path = os.path.join(base_path, "bar")
-        self.assertFalse(contains_symlink_up_to(target_path, base_path))
-        self.assertFalse(contains_symlink_up_to(target_path, base_path))
+
+def test_should_return_false_when_base_path_is_symlink(mocker):
+    base_path = "foo"
+    target_path = os.path.join(base_path, "bar")
+
+    def base_path_is_symlink(path):
+        if path == base_path:
+            return True
+        return False
+
+    mocker.patch.object(
+        System,
+        "is_symlink",
+        return_value=True,
+        side_effect=base_path_is_symlink,
+    )
+    assert not contains_symlink_up_to(target_path, base_path)
+
+
+def test_path_object_and_str_are_valid_arg_types():
+    base_path = "foo"
+    target_path = os.path.join(base_path, "bar")
+    assert not contains_symlink_up_to(target_path, base_path)
+    assert not contains_symlink_up_to(target_path, base_path)
 
 
 def test_should_call_recursive_on_no_condition_matched(mocker):