1from io import BytesIO 2 3from pytest import raises, fixture 4from unittest.mock import patch 5from six import string_types 6 7import ansible_runner.loader 8 9from ansible_runner.exceptions import ConfigurationError 10 11 12@fixture 13def loader(): 14 return ansible_runner.loader.ArtifactLoader('/tmp') 15 16 17def test__load_json_success(loader): 18 res = loader._load_json('{"test": "string"}') 19 assert isinstance(res, dict) 20 assert res['test'] == 'string' 21 22 23def test__load_json_failure(loader): 24 res = loader._load_json('---\ntest: string') 25 assert res is None 26 27 res = loader._load_json('test string') 28 assert res is None 29 30 31def test__load_yaml_success(loader): 32 res = loader._load_yaml('---\ntest: string') 33 assert isinstance(res, dict) 34 assert res['test'] == 'string' 35 36 res = loader._load_yaml('{"test": "string"}') 37 assert isinstance(res, dict) 38 assert res['test'] == 'string' 39 40 41def test__load_yaml_failure(loader): 42 res = loader._load_yaml('---\ntest: string:') 43 assert res is None 44 45 46def test_abspath(loader): 47 res = loader.abspath('/test') 48 assert res == '/test' 49 50 res = loader.abspath('test') 51 assert res == '/tmp/test' 52 53 res = loader.abspath('~/test') 54 assert res.startswith('/') 55 56 57def test_load_file_text(loader): 58 with patch.object(ansible_runner.loader.ArtifactLoader, 'get_contents') as mock_get_contents: 59 mock_get_contents.return_value = 'test\nstring' 60 61 assert not loader._cache 62 63 # cache miss 64 res = loader.load_file('/tmp/test', string_types) 65 assert mock_get_contents.called 66 assert mock_get_contents.called_with_args('/tmp/test') 67 assert res == b'test\nstring' 68 assert '/tmp/test' in loader._cache 69 70 mock_get_contents.reset_mock() 71 72 # cache hit 73 res = loader.load_file('/tmp/test', string_types) 74 assert not mock_get_contents.called 75 assert res == b'test\nstring' 76 assert '/tmp/test' in loader._cache 77 78 79def test_load_file_json(loader): 80 with patch.object(ansible_runner.loader.ArtifactLoader, 'get_contents') as mock_get_contents: 81 mock_get_contents.return_value = '---\ntest: string' 82 83 assert not loader._cache 84 85 res = loader.load_file('/tmp/test') 86 87 assert mock_get_contents.called 88 assert mock_get_contents.called_with_args('/tmp/test') 89 assert '/tmp/test' in loader._cache 90 assert res['test'] == 'string' 91 92 93def test_load_file_type_check(loader): 94 with patch.object(ansible_runner.loader.ArtifactLoader, 'get_contents') as mock_get_contents: 95 mock_get_contents.return_value = '---\ntest: string' 96 97 assert not loader._cache 98 99 # type check passes 100 res = loader.load_file('/tmp/test', dict) 101 assert res is not None 102 103 mock_get_contents.reset_mock() 104 mock_get_contents.return_value = 'test string' 105 106 loader._cache = {} 107 108 # type check fails 109 with raises(ConfigurationError): 110 res = loader.load_file('/tmp/test', dict) 111 assert res is not None 112 113 114def test_get_contents_ok(loader): 115 with patch('codecs.open') as mock_open: 116 handler = BytesIO() 117 handler.write(b"test string") 118 handler.seek(0) 119 120 mock_open.return_value.__enter__.return_value = handler 121 122 res = loader.get_contents('/tmp') 123 assert res == b'test string' 124 125 126def test_get_contents_invalid_path(loader): 127 with raises(ConfigurationError): 128 loader.get_contents('/tmp/invalid') 129 130 131def test_get_contents_exception(loader): 132 with raises(ConfigurationError): 133 loader.get_contents('/tmp') 134