3
3
"""
4
4
5
5
from pathlib import Path
6
+ from unittest .mock import patch
6
7
7
- from pytest import fixture
8
+ from pytest import fixture , mark
8
9
9
10
from uwtools .api .config import get_yaml_config
10
11
from uwtools .drivers .stager import FileStager
11
-
12
- # Need one file on a non-/tmp disk for testing hardlink fallback
13
- B_FILE = Path ("./b.foo" ).resolve ()
12
+ from uwtools .utils import tasks
14
13
15
14
16
15
class NullDriver (FileStager ):
@@ -24,14 +23,6 @@ def taskname(self, s):
24
23
return f"Null { s } "
25
24
26
25
27
- def setup_module (module ): # noqa: ARG001
28
- B_FILE .touch ()
29
-
30
-
31
- def teardown_module (module ): # noqa: ARG001
32
- B_FILE .unlink ()
33
-
34
-
35
26
@fixture
36
27
def driver (tmp_path ):
37
28
i = tmp_path / "input"
@@ -42,15 +33,14 @@ def driver(tmp_path):
42
33
tmp: %s
43
34
files_to_copy:
44
35
a.foo: '{{ tmp }}/input/a.foo'
45
- b.foo: %s
36
+ b.foo: '{{ tmp }}/input/b.foo'
46
37
files_to_link:
47
38
c.foo: '{{ tmp }}/input/a.foo'
48
39
d.foo: '{{ tmp }}/input/a.foo'
49
40
output/<files>: !glob '{{ tmp }}/input/*.foo'
50
41
files_to_hardlink:
51
- e.foo: %s
52
- f.foo: '{{ tmp }}/input/a.foo'
53
- """ % (tmp_path , B_FILE , B_FILE )
42
+ e.foo: '{{ tmp }}/input/a.foo'
43
+ """ % (tmp_path )
54
44
cfg = tmp_path / "config.yaml"
55
45
cfg .write_text (config )
56
46
return NullDriver (config_file = cfg , rundir = tmp_path )
@@ -62,14 +52,18 @@ def test_files_to_copy(driver, tmp_path):
62
52
assert (tmp_path / "b.foo" ).is_file ()
63
53
64
54
65
- def test_files_to_hardlink (driver , tmp_path ):
66
- driver .files_hardlinked ()
67
- a = tmp_path / "input" / "a.foo"
55
+ @mark .parametrize ("success" , [True , False ])
56
+ def test_files_to_hardlink (driver , success , tmp_path ):
68
57
e = tmp_path / "e.foo"
69
- f = tmp_path / "f.foo"
70
- assert e .is_file ()
71
- assert not e .samefile (B_FILE )
72
- assert a .samefile (f )
58
+ if success :
59
+ driver .files_hardlinked ()
60
+ # It's a hardlink
61
+ assert e .stat ().st_nlink == 2
62
+ else :
63
+ with patch .object (tasks .os , "link" , side_effect = OSError ()):
64
+ driver .files_hardlinked ()
65
+ # It's a copy
66
+ assert e .stat ().st_nlink == 1
73
67
74
68
75
69
def test_files_to_link (driver , tmp_path ):
0 commit comments