forked from iterative/dvc
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_import.py
118 lines (90 loc) · 4.11 KB
/
test_import.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
106
107
108
109
110
111
112
113
114
115
116
117
118
import os
import tempfile
from copy import copy
from dvc.command.import_file import CmdImportFile, ImportFileError
from tests.basic_env import DirHierarchyEnvironment
class TestCmdRemove(DirHierarchyEnvironment):
def setUp(self):
DirHierarchyEnvironment.init_environment(self)
def test_import_file_to_dir(self):
with tempfile.NamedTemporaryFile(mode='w') as temp:
content = 'Some data'
temp.write(content)
temp.flush()
dir = os.path.join('data', self.dir1)
settings = copy(self.settings)
settings.parse_args('import {} {}'.format(temp.name, dir))
cmd = CmdImportFile(settings)
cmd.run()
dvc_name = os.path.join(dir, os.path.basename(temp.name))
data_item = self.path_factory.existing_data_item(dvc_name)
self.assertTrue(os.path.exists(data_item.data.relative))
self.assertTrue(os.path.islink(data_item.data.relative))
self.assertEqual(open(data_item.data.relative).read(), content)
self.assertTrue(os.path.exists(data_item.cache.relative))
self.assertEqual(open(data_item.cache.relative).read(), content)
self.assertTrue(os.path.exists(data_item.state.relative))
pass
def test_import_file_to_file(self):
with tempfile.NamedTemporaryFile(mode='w') as temp:
content = 'Some data'
temp.write(content)
temp.flush()
target_file_name = 'targ.txt'
target_file = os.path.join('data', self.dir1, target_file_name)
settings = copy(self.settings)
settings.parse_args('import {} {}'.format(temp.name, target_file))
cmd = CmdImportFile(settings)
cmd.run()
self.assertTrue(os.path.exists(target_file))
self.assertTrue(open(target_file).read(), content)
pass
def test_import_to_existing_dir(self):
with tempfile.NamedTemporaryFile(mode='w') as temp:
content = 'Some data'
temp.write(content)
temp.flush()
dir = os.path.join('data', self.dir11)
settings = copy(self.settings)
settings.parse_args('import {} {}'.format(temp.name, dir))
cmd = CmdImportFile(settings)
cmd.run()
dvc_name = os.path.join(dir, os.path.basename(temp.name))
self.assertTrue(os.path.exists(dvc_name))
self.assertTrue(open(dvc_name).read(), content)
pass
def test_import_to_not_existing_dir(self):
with tempfile.NamedTemporaryFile(mode='w') as temp:
content = 'Some data'
temp.write(content)
temp.flush()
dir = os.path.join('data', self.dir1, 'not_existing_dir/')
settings = copy(self.settings)
settings.parse_args('import {} {}'.format(temp.name, dir))
cmd = CmdImportFile(settings)
with self.assertRaises(ImportFileError):
cmd.run()
dvc_name = os.path.join(dir, os.path.basename(temp.name))
self.assertFalse(os.path.exists(dir))
self.assertFalse(os.path.exists(dvc_name))
pass
def test_multiple_import(self):
with tempfile.NamedTemporaryFile(mode='w') as temp1,\
tempfile.NamedTemporaryFile(mode='w') as temp2:
content = 'Some data'
temp1.write(content)
temp1.flush()
temp2.write(content)
temp2.flush()
dir = os.path.join('data', self.dir11)
settings = copy(self.settings)
settings.parse_args('import {} {} {}'.format(temp1.name, temp2.name, dir))
cmd = CmdImportFile(settings)
cmd.run()
dvc_name1 = os.path.join(dir, os.path.basename(temp1.name))
dvc_name2 = os.path.join(dir, os.path.basename(temp2.name))
self.assertTrue(os.path.exists(dvc_name1))
self.assertTrue(open(dvc_name1).read(), content)
self.assertTrue(os.path.exists(dvc_name2))
self.assertTrue(open(dvc_name2).read(), content)
pass