Total Complexity | 75 |
Total Lines | 270 |
Duplicated Lines | 31.48 % |
Changes | 0 |
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like PathActionsTests often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
1 | # -*- coding: utf-8 -*- |
||
70 | class PathActionsTests(TestCase): |
||
71 | def setUp(self): |
||
72 | tempdirdir = gettempdir() |
||
73 | |||
74 | prefix_dirname = str(uuid4())[:4] + ' ' + str(uuid4())[:4] + '-prefix' |
||
75 | self.prefix = join(tempdirdir, prefix_dirname) |
||
76 | mkdir_p(self.prefix) |
||
77 | assert isdir(self.prefix) |
||
78 | |||
79 | pkgs_dirname = str(uuid4())[:4] + ' ' + str(uuid4())[:4] |
||
80 | self.pkgs_dir = join(tempdirdir, pkgs_dirname) |
||
81 | mkdir_p(self.pkgs_dir) |
||
82 | assert isdir(self.pkgs_dir) |
||
83 | |||
84 | def tearDown(self): |
||
85 | rm_rf(self.prefix) |
||
86 | if not (on_win and PY2): |
||
87 | # this assertion fails for the Softlink action windows tests |
||
88 | # line 141 in backoff_rmdir |
||
89 | # exp_backoff_fn(rmtree, path, onerror=retry, max_tries=max_tries) |
||
90 | # leaves a directory self.prefix\\Scripts that cannot be accessed or removed |
||
91 | assert not lexists(self.prefix) |
||
92 | rm_rf(self.pkgs_dir) |
||
93 | assert not lexists(self.pkgs_dir) |
||
94 | |||
95 | def test_CompilePycAction_generic(self): |
||
96 | package_info = AttrDict( |
||
97 | package_metadata=AttrDict( |
||
98 | noarch=AttrDict( |
||
99 | type=NoarchType.generic)) |
||
100 | ) |
||
101 | noarch = package_info.package_metadata and package_info.package_metadata.noarch |
||
102 | assert noarch.type == NoarchType.generic |
||
103 | axns = CompilePycAction.create_actions({}, package_info, self.prefix, None, ()) |
||
104 | assert axns == () |
||
105 | |||
106 | package_info = AttrDict(package_metadata=None) |
||
107 | axns = CompilePycAction.create_actions({}, package_info, self.prefix, None, ()) |
||
108 | assert axns == () |
||
109 | |||
110 | def test_CompilePycAction_noarch_python(self): |
||
111 | if not softlink_supported(__file__, self.prefix) and on_win: |
||
112 | pytest.skip("softlink not supported") |
||
113 | |||
114 | target_python_version = '%d.%d' % sys.version_info[:2] |
||
115 | sp_dir = get_python_site_packages_short_path(target_python_version) |
||
116 | transaction_context = { |
||
117 | 'target_python_version': target_python_version, |
||
118 | 'target_site_packages_short_path': sp_dir, |
||
119 | } |
||
120 | package_info = AttrDict(package_metadata=AttrDict(noarch=AttrDict(type=NoarchType.python))) |
||
121 | |||
122 | file_link_actions = [ |
||
123 | AttrDict( |
||
124 | source_short_path='site-packages/something.py', |
||
125 | target_short_path=get_python_noarch_target_path('site-packages/something.py', sp_dir), |
||
126 | ), |
||
127 | AttrDict( |
||
128 | # this one shouldn't get compiled |
||
129 | source_short_path='something.py', |
||
130 | target_short_path=get_python_noarch_target_path('something.py', sp_dir), |
||
131 | ), |
||
132 | ] |
||
133 | axns = CompilePycAction.create_actions(transaction_context, package_info, self.prefix, |
||
134 | None, file_link_actions) |
||
135 | |||
136 | assert len(axns) == 1 |
||
137 | axn = axns[0] |
||
138 | assert axn.source_full_path == join(self.prefix, win_path_ok(get_python_noarch_target_path('site-packages/something.py', sp_dir))) |
||
139 | assert axn.target_full_path == join(self.prefix, win_path_ok(pyc_path(get_python_noarch_target_path('site-packages/something.py', sp_dir), |
||
140 | target_python_version))) |
||
141 | |||
142 | # make .py file in prefix that will be compiled |
||
143 | mkdir_p(dirname(axn.source_full_path)) |
||
144 | with open(axn.source_full_path, 'w') as fh: |
||
145 | fh.write("value = 42\n") |
||
146 | |||
147 | # symlink the current python |
||
148 | python_full_path = join(self.prefix, get_python_short_path(target_python_version)) |
||
149 | mkdir_p(dirname(python_full_path)) |
||
150 | create_link(sys.executable, python_full_path, LinkType.softlink) |
||
151 | |||
152 | axn.execute() |
||
153 | assert isfile(axn.target_full_path) |
||
154 | |||
155 | # remove the source .py file so we're sure we're importing the pyc file below |
||
156 | rm_rf(axn.source_full_path) |
||
157 | assert not isfile(axn.source_full_path) |
||
158 | |||
159 | if (3,) > sys.version_info >= (3, 5): |
||
160 | # we're probably dropping py34 support soon enough anyway |
||
161 | imported_pyc_file = load_python_file(axn.target_full_path) |
||
162 | assert imported_pyc_file.value == 42 |
||
163 | |||
164 | axn.reverse() |
||
165 | assert not isfile(axn.target_full_path) |
||
166 | |||
167 | def test_CreatePythonEntryPointAction_generic(self): |
||
168 | package_info = AttrDict(package_metadata=None) |
||
169 | axns = CreatePythonEntryPointAction.create_actions({}, package_info, self.prefix, None) |
||
170 | assert axns == () |
||
171 | |||
172 | def test_CreatePythonEntryPointAction_noarch_python(self): |
||
173 | target_python_version = '%d.%d' % sys.version_info[:2] |
||
174 | transaction_context = { |
||
175 | 'target_python_version': target_python_version, |
||
176 | } |
||
177 | package_info = AttrDict(package_metadata=AttrDict(noarch=AttrDict( |
||
178 | type=NoarchType.python, |
||
179 | entry_points=( |
||
180 | 'command1=some.module:main', |
||
181 | 'command2=another.somewhere:go', |
||
182 | ), |
||
183 | ))) |
||
184 | |||
185 | axns = CreatePythonEntryPointAction.create_actions(transaction_context, package_info, |
||
186 | self.prefix, LinkType.hardlink) |
||
187 | grouped_axns = groupby(lambda ax: isinstance(ax, LinkPathAction), axns) |
||
188 | windows_exe_axns = grouped_axns.get(True, ()) |
||
189 | assert len(windows_exe_axns) == (2 if on_win else 0) |
||
190 | py_ep_axns = grouped_axns.get(False, ()) |
||
191 | assert len(py_ep_axns) == 2 |
||
192 | |||
193 | py_ep_axn = py_ep_axns[0] |
||
194 | |||
195 | command, module, func = parse_entry_point_def('command1=some.module:main') |
||
196 | assert command == 'command1' |
||
197 | if on_win: |
||
198 | target_short_path = "%s\\%s-script.py" % (get_bin_directory_short_path(), command) |
||
199 | else: |
||
200 | target_short_path = "%s/%s" % (get_bin_directory_short_path(), command) |
||
201 | assert py_ep_axn.target_full_path == join(self.prefix, target_short_path) |
||
202 | assert py_ep_axn.module == module == 'some.module' |
||
203 | assert py_ep_axn.func == func == 'main' |
||
204 | |||
205 | mkdir_p(dirname(py_ep_axn.target_full_path)) |
||
206 | py_ep_axn.execute() |
||
207 | assert isfile(py_ep_axn.target_full_path) |
||
208 | if not on_win: |
||
209 | assert is_executable(py_ep_axn.target_full_path) |
||
210 | with open(py_ep_axn.target_full_path) as fh: |
||
211 | lines = fh.readlines() |
||
212 | first_line = lines[0].strip() |
||
213 | last_line = lines[-1].strip() |
||
214 | if not on_win: |
||
215 | python_full_path = join(self.prefix, get_python_short_path(target_python_version)) |
||
216 | assert first_line == "#!%s" % python_full_path |
||
217 | assert last_line == "sys.exit(%s())" % func |
||
218 | |||
219 | py_ep_axn.reverse() |
||
220 | assert not isfile(py_ep_axn.target_full_path) |
||
221 | |||
222 | if on_win: |
||
223 | windows_exe_axn = windows_exe_axns[0] |
||
224 | target_short_path = "%s\\%s.exe" % (get_bin_directory_short_path(), command) |
||
225 | assert windows_exe_axn.target_full_path == join(self.prefix, target_short_path) |
||
226 | |||
227 | mkdir_p(dirname(windows_exe_axn.target_full_path)) |
||
228 | windows_exe_axn.verify() |
||
229 | windows_exe_axn.execute() |
||
230 | assert isfile(windows_exe_axn.target_full_path) |
||
231 | assert is_executable(windows_exe_axn.target_full_path) |
||
232 | |||
233 | src = compute_md5sum(join(context.conda_prefix, 'Scripts/conda.exe')) |
||
234 | assert src == compute_md5sum(windows_exe_axn.target_full_path) |
||
235 | |||
236 | windows_exe_axn.reverse() |
||
237 | assert not isfile(windows_exe_axn.target_full_path) |
||
238 | |||
239 | View Code Duplication | def test_simple_LinkPathAction_hardlink(self): |
|
|
|||
240 | source_full_path = make_test_file(self.pkgs_dir) |
||
241 | target_short_path = source_short_path = basename(source_full_path) |
||
242 | |||
243 | correct_sha256 = compute_sha256sum(source_full_path) |
||
244 | correct_size_in_bytes = getsize(source_full_path) |
||
245 | path_type = PathType.hardlink |
||
246 | |||
247 | source_path_data = PathDataV1( |
||
248 | _path = source_short_path, |
||
249 | path_type=path_type, |
||
250 | sha256=correct_sha256, |
||
251 | size_in_bytes=correct_size_in_bytes, |
||
252 | ) |
||
253 | |||
254 | axn = LinkPathAction({}, None, self.pkgs_dir, source_short_path, self.prefix, |
||
255 | target_short_path, LinkType.hardlink, source_path_data) |
||
256 | |||
257 | assert axn.target_full_path == join(self.prefix, target_short_path) |
||
258 | axn.verify() |
||
259 | axn.execute() |
||
260 | assert isfile(axn.target_full_path) |
||
261 | assert not islink(axn.target_full_path) |
||
262 | assert stat_nlink(axn.target_full_path) == 2 |
||
263 | |||
264 | axn.reverse() |
||
265 | assert not lexists(axn.target_full_path) |
||
266 | |||
267 | View Code Duplication | def test_simple_LinkPathAction_softlink(self): |
|
268 | if not softlink_supported(__file__, self.prefix) and on_win: |
||
269 | pytest.skip("softlink not supported") |
||
270 | |||
271 | source_full_path = make_test_file(self.pkgs_dir) |
||
272 | target_short_path = source_short_path = basename(source_full_path) |
||
273 | |||
274 | correct_sha256 = compute_sha256sum(source_full_path) |
||
275 | correct_size_in_bytes = getsize(source_full_path) |
||
276 | path_type = PathType.hardlink |
||
277 | |||
278 | source_path_data = PathDataV1( |
||
279 | _path = source_short_path, |
||
280 | path_type=path_type, |
||
281 | sha256=correct_sha256, |
||
282 | size_in_bytes=correct_size_in_bytes, |
||
283 | ) |
||
284 | |||
285 | axn = LinkPathAction({}, None, self.pkgs_dir, source_short_path, self.prefix, |
||
286 | target_short_path, LinkType.softlink, source_path_data) |
||
287 | |||
288 | assert axn.target_full_path == join(self.prefix, target_short_path) |
||
289 | axn.verify() |
||
290 | axn.execute() |
||
291 | assert isfile(axn.target_full_path) |
||
292 | assert islink(axn.target_full_path) |
||
293 | assert stat_nlink(axn.target_full_path) == 1 |
||
294 | |||
295 | axn.reverse() |
||
296 | assert not lexists(axn.target_full_path) |
||
297 | assert lexists(source_full_path) |
||
298 | |||
299 | def test_simple_LinkPathAction_directory(self): |
||
300 | target_short_path = join('a', 'nested', 'directory') |
||
301 | axn = LinkPathAction({}, None, None, None, self.prefix, |
||
302 | target_short_path, LinkType.directory, None) |
||
303 | axn.verify() |
||
304 | axn.execute() |
||
305 | |||
306 | assert isdir(join(self.prefix, target_short_path)) |
||
307 | |||
308 | axn.reverse() |
||
309 | assert not lexists(axn.target_full_path) |
||
310 | assert not lexists(dirname(axn.target_full_path)) |
||
311 | assert not lexists(dirname(dirname(axn.target_full_path))) |
||
312 | |||
313 | View Code Duplication | def test_simple_LinkPathAction_copy(self): |
|
314 | source_full_path = make_test_file(self.pkgs_dir) |
||
315 | target_short_path = source_short_path = basename(source_full_path) |
||
316 | |||
317 | correct_sha256 = compute_sha256sum(source_full_path) |
||
318 | correct_size_in_bytes = getsize(source_full_path) |
||
319 | path_type = PathType.hardlink |
||
320 | |||
321 | source_path_data = PathDataV1( |
||
322 | _path = source_short_path, |
||
323 | path_type=path_type, |
||
324 | sha256=correct_sha256, |
||
325 | size_in_bytes=correct_size_in_bytes, |
||
326 | ) |
||
327 | |||
328 | axn = LinkPathAction({}, None, self.pkgs_dir, source_short_path, self.prefix, |
||
329 | target_short_path, LinkType.copy, source_path_data) |
||
330 | |||
331 | assert axn.target_full_path == join(self.prefix, target_short_path) |
||
332 | axn.verify() |
||
333 | axn.execute() |
||
334 | assert isfile(axn.target_full_path) |
||
335 | assert not islink(axn.target_full_path) |
||
336 | assert stat_nlink(axn.target_full_path) == 1 |
||
337 | |||
338 | axn.reverse() |
||
339 | assert not lexists(axn.target_full_path) |
||
340 | |||
526 |