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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
# pylint: disable=W0212,W0621
from io import BytesIO
from pytest import raises, fixture
import ansible_runner.loader
from ansible_runner.exceptions import ConfigurationError
@fixture
def loader(tmp_path):
return ansible_runner.loader.ArtifactLoader(str(tmp_path))
def test__load_json_success(loader):
res = loader._load_json('{"test": "string"}')
assert isinstance(res, dict)
assert res['test'] == 'string'
def test__load_json_failure(loader):
res = loader._load_json('---\ntest: string')
assert res is None
res = loader._load_json('test string')
assert res is None
def test__load_yaml_success(loader):
res = loader._load_yaml('---\ntest: string')
assert isinstance(res, dict)
assert res['test'] == 'string'
res = loader._load_yaml('{"test": "string"}')
assert isinstance(res, dict)
assert res['test'] == 'string'
def test__load_yaml_failure(loader):
res = loader._load_yaml('---\ntest: string:')
assert res is None
def test_abspath(loader, tmp_path):
res = loader.abspath('/test')
assert res == '/test'
res = loader.abspath('test')
assert res == tmp_path.joinpath('test').as_posix()
res = loader.abspath('~/test')
assert res.startswith('/')
def test_load_file_text_cache_hit(loader, mocker, tmp_path):
mock_get_contents = mocker.patch.object(ansible_runner.loader.ArtifactLoader, '_get_contents')
mock_get_contents.return_value = 'test\nstring'
assert not loader._cache
testfile = tmp_path.joinpath('test').as_posix()
# cache miss
res = loader.load_file(testfile, str)
assert mock_get_contents.called
assert mock_get_contents.called_with_args(testfile)
assert res == b'test\nstring'
assert testfile in loader._cache
mock_get_contents.reset_mock()
# cache hit
res = loader.load_file(testfile, str)
assert not mock_get_contents.called
assert res == b'test\nstring'
assert testfile in loader._cache
def test_load_file_json(loader, mocker, tmp_path):
mock_get_contents = mocker.patch.object(ansible_runner.loader.ArtifactLoader, '_get_contents')
mock_get_contents.return_value = '---\ntest: string'
assert not loader._cache
testfile = tmp_path.joinpath('test').as_posix()
res = loader.load_file(testfile)
assert mock_get_contents.called
assert mock_get_contents.called_with_args(testfile)
assert testfile in loader._cache
assert res['test'] == 'string'
def test_load_file_type_check(loader, mocker, tmp_path):
mock_get_contents = mocker.patch.object(ansible_runner.loader.ArtifactLoader, '_get_contents')
mock_get_contents.return_value = '---\ntest: string'
assert not loader._cache
testfile = tmp_path.joinpath('test').as_posix()
# type check passes
res = loader.load_file(testfile, dict)
assert res is not None
mock_get_contents.reset_mock()
mock_get_contents.return_value = 'test string'
loader._cache = {}
# type check fails
with raises(ConfigurationError):
res = loader.load_file(testfile, dict)
assert res is not None
def test_get_contents_ok(loader, mocker):
mock_open = mocker.patch('codecs.open')
handler = BytesIO()
handler.write(b"test string")
handler.seek(0)
mock_open.return_value.__enter__.return_value = handler
res = loader._get_contents('/tmp')
assert res == b'test string'
def test_get_contents_invalid_path(loader, tmp_path):
with raises(ConfigurationError):
loader._get_contents(tmp_path.joinpath('invalid').as_posix())
def test_get_contents_exception(loader, tmp_path):
with raises(ConfigurationError):
loader._get_contents(tmp_path.as_posix())
|
