forked from iterative/dvc
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_cmd_target.py
105 lines (80 loc) · 3.67 KB
/
test_cmd_target.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
import os
from dvc.command.target import CmdTarget
from dvc.config import ConfigI
from dvc.git_wrapper import GitWrapperI
from dvc.path.factory import PathFactory
from dvc.settings import Settings
from dvc.system import System
from tests.basic_env import BasicEnvironment
class TestCmdTarget(BasicEnvironment):
def setUp(self):
self._test_dir = os.path.join(os.path.sep, 'tmp', 'dvc_unit_test')
curr_dir = None
commit = 'abc1234'
BasicEnvironment.init_environment(self, self._test_dir, curr_dir)
self._commit = commit
self._git = GitWrapperI(git_dir=self._test_git_dir, commit=self._commit)
self._config = ConfigI('data', 'cache', 'state', '.target')
self.path_factory = PathFactory(self._git, self._config)
self.settings = Settings(['target'], self._git, self._config)
os.chdir(self._test_git_dir)
os.mkdir(os.path.join('data', 'dir1'))
os.mkdir(os.path.join('cache', 'dir1'))
self.file1_cache = os.path.join('cache', 'dir1', 'file1')
self.file1_data = os.path.join('data', 'dir1', 'file1')
open(self.file1_cache, 'w').write('ddfdff')
System.symlink(self.file1_cache, self.file1_data)
self.file2_cache = os.path.join('cache', 'file2')
self.file2_data = os.path.join('data', 'file2')
open(self.file2_cache, 'w').write('ddfdff')
System.symlink(self.file2_cache, self.file2_data)
def test_initial_default_target(self):
self.assertFalse(os.path.exists('.target'))
def test_single_file(self):
self.settings.parse_args('target {}'.format(self.file1_data))
cmd = CmdTarget(self.settings)
self.assertEqual(cmd.run(), 0)
self.assertEqual(open('.target').read(), self.file1_data)
def test_multiple_files(self):
self.settings.parse_args('target {}'.format(self.file1_data))
cmd = CmdTarget(self.settings)
cmd.run()
# Another target
self.settings.parse_args('target {}'.format(self.file2_data))
cmd = CmdTarget(self.settings)
self.assertEqual(cmd.run(), 0)
self.assertEqual(open('.target').read(), self.file2_data)
# Unset target
self.settings.parse_args('target --unset')
cmd = CmdTarget(self.settings)
self.assertEqual(cmd.run(), 0)
self.assertEqual(open('.target').read(), '')
def test_initial_unset(self):
self.settings.parse_args('target --unset')
cmd = CmdTarget(self.settings)
self.assertEqual(cmd.run(), 1)
self.assertFalse(os.path.exists('.target'))
def test_unset_existing_target(self):
self.settings.parse_args('target {}'.format(self.file1_data))
cmd = CmdTarget(self.settings)
self.assertEqual(cmd.run(), 0)
self.assertEqual(open('.target').read(), self.file1_data)
self.settings.parse_args('target --unset')
cmd = CmdTarget(self.settings)
self.assertEqual(cmd.run(), 0)
self.assertEqual(open('.target').read(), '')
def test_the_same(self):
self.settings.parse_args('target {}'.format(self.file1_data))
cmd = CmdTarget(self.settings)
self.assertEqual(cmd.run(), 0)
self.assertEqual(open('.target').read(), self.file1_data)
self.assertEqual(cmd.run(), 0)
self.assertEqual(open('.target').read(), self.file1_data)
def test_args_conflict(self):
self.settings.parse_args('target {} --unset'.format(self.file1_data))
cmd = CmdTarget(self.settings)
self.assertEqual(cmd.run(), 1)
def test_no_args(self):
self.settings.parse_args('target --unset')
cmd = CmdTarget(self.settings)
self.assertEqual(cmd.run(), 1)