1# -*- coding: utf-8 -*- 2# Copyright: (c) 2019, Ansible Project 3# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) 4 5# Make coding more python3-ish 6from __future__ import (absolute_import, division, print_function) 7__metaclass__ = type 8 9import json 10import os 11import re 12import pytest 13import stat 14import tarfile 15import tempfile 16import time 17 18from io import BytesIO, StringIO 19from units.compat.mock import MagicMock 20 21import ansible.constants as C 22from ansible import context 23from ansible.errors import AnsibleError 24from ansible.galaxy import api as galaxy_api 25from ansible.galaxy.api import CollectionVersionMetadata, GalaxyAPI, GalaxyError 26from ansible.galaxy.token import BasicAuthToken, GalaxyToken, KeycloakToken 27from ansible.module_utils._text import to_native, to_text 28from ansible.module_utils.six.moves.urllib import error as urllib_error 29from ansible.utils import context_objects as co 30from ansible.utils.display import Display 31 32 33@pytest.fixture(autouse='function') 34def reset_cli_args(): 35 co.GlobalCLIArgs._Singleton__instance = None 36 # Required to initialise the GalaxyAPI object 37 context.CLIARGS._store = {'ignore_certs': False} 38 yield 39 co.GlobalCLIArgs._Singleton__instance = None 40 41 42@pytest.fixture() 43def collection_artifact(tmp_path_factory): 44 ''' Creates a collection artifact tarball that is ready to be published ''' 45 output_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Output')) 46 47 tar_path = os.path.join(output_dir, 'namespace-collection-v1.0.0.tar.gz') 48 with tarfile.open(tar_path, 'w:gz') as tfile: 49 b_io = BytesIO(b"\x00\x01\x02\x03") 50 tar_info = tarfile.TarInfo('test') 51 tar_info.size = 4 52 tar_info.mode = 0o0644 53 tfile.addfile(tarinfo=tar_info, fileobj=b_io) 54 55 yield tar_path 56 57 58@pytest.fixture() 59def cache_dir(tmp_path_factory, monkeypatch): 60 cache_dir = to_text(tmp_path_factory.mktemp('Test ÅÑŚÌβŁÈ Galaxy Cache')) 61 monkeypatch.setitem(C.config._base_defs, 'GALAXY_CACHE_DIR', {'default': cache_dir}) 62 63 yield cache_dir 64 65 66def get_test_galaxy_api(url, version, token_ins=None, token_value=None, no_cache=True): 67 token_value = token_value or "my token" 68 token_ins = token_ins or GalaxyToken(token_value) 69 api = GalaxyAPI(None, "test", url, no_cache=no_cache) 70 # Warning, this doesn't test g_connect() because _availabe_api_versions is set here. That means 71 # that urls for v2 servers have to append '/api/' themselves in the input data. 72 api._available_api_versions = {version: '%s' % version} 73 api.token = token_ins 74 75 return api 76 77 78def get_collection_versions(namespace='namespace', name='collection'): 79 base_url = 'https://galaxy.server.com/api/v2/collections/{0}/{1}/'.format(namespace, name) 80 versions_url = base_url + 'versions/' 81 82 # Response for collection info 83 responses = [ 84 { 85 "id": 1000, 86 "href": base_url, 87 "name": name, 88 "namespace": { 89 "id": 30000, 90 "href": "https://galaxy.ansible.com/api/v1/namespaces/30000/", 91 "name": namespace, 92 }, 93 "versions_url": versions_url, 94 "latest_version": { 95 "version": "1.0.5", 96 "href": versions_url + "1.0.5/" 97 }, 98 "deprecated": False, 99 "created": "2021-02-09T16:55:42.749915-05:00", 100 "modified": "2021-02-09T16:55:42.749915-05:00", 101 } 102 ] 103 104 # Paginated responses for versions 105 page_versions = (('1.0.0', '1.0.1',), ('1.0.2', '1.0.3',), ('1.0.4', '1.0.5'),) 106 last_page = None 107 for page in range(1, len(page_versions) + 1): 108 if page < len(page_versions): 109 next_page = versions_url + '?page={0}'.format(page + 1) 110 else: 111 next_page = None 112 113 version_results = [] 114 for version in page_versions[int(page - 1)]: 115 version_results.append( 116 {'version': version, 'href': versions_url + '{0}/'.format(version)} 117 ) 118 119 responses.append( 120 { 121 'count': 6, 122 'next': next_page, 123 'previous': last_page, 124 'results': version_results, 125 } 126 ) 127 last_page = page 128 129 return responses 130 131 132def test_api_no_auth(): 133 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/") 134 actual = {} 135 api._add_auth_token(actual, "") 136 assert actual == {} 137 138 139def test_api_no_auth_but_required(): 140 expected = "No access token or username set. A token can be set with --api-key or at " 141 with pytest.raises(AnsibleError, match=expected): 142 GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/")._add_auth_token({}, "", required=True) 143 144 145def test_api_token_auth(): 146 token = GalaxyToken(token=u"my_token") 147 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) 148 actual = {} 149 api._add_auth_token(actual, "", required=True) 150 assert actual == {'Authorization': 'Token my_token'} 151 152 153def test_api_token_auth_with_token_type(monkeypatch): 154 token = KeycloakToken(auth_url='https://api.test/') 155 mock_token_get = MagicMock() 156 mock_token_get.return_value = 'my_token' 157 monkeypatch.setattr(token, 'get', mock_token_get) 158 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) 159 actual = {} 160 api._add_auth_token(actual, "", token_type="Bearer", required=True) 161 assert actual == {'Authorization': 'Bearer my_token'} 162 163 164def test_api_token_auth_with_v3_url(monkeypatch): 165 token = KeycloakToken(auth_url='https://api.test/') 166 mock_token_get = MagicMock() 167 mock_token_get.return_value = 'my_token' 168 monkeypatch.setattr(token, 'get', mock_token_get) 169 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) 170 actual = {} 171 api._add_auth_token(actual, "https://galaxy.ansible.com/api/v3/resource/name", required=True) 172 assert actual == {'Authorization': 'Bearer my_token'} 173 174 175def test_api_token_auth_with_v2_url(): 176 token = GalaxyToken(token=u"my_token") 177 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) 178 actual = {} 179 # Add v3 to random part of URL but response should only see the v2 as the full URI path segment. 180 api._add_auth_token(actual, "https://galaxy.ansible.com/api/v2/resourcev3/name", required=True) 181 assert actual == {'Authorization': 'Token my_token'} 182 183 184def test_api_basic_auth_password(): 185 token = BasicAuthToken(username=u"user", password=u"pass") 186 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) 187 actual = {} 188 api._add_auth_token(actual, "", required=True) 189 assert actual == {'Authorization': 'Basic dXNlcjpwYXNz'} 190 191 192def test_api_basic_auth_no_password(): 193 token = BasicAuthToken(username=u"user") 194 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) 195 actual = {} 196 api._add_auth_token(actual, "", required=True) 197 assert actual == {'Authorization': 'Basic dXNlcjo='} 198 199 200def test_api_dont_override_auth_header(): 201 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/") 202 actual = {'Authorization': 'Custom token'} 203 api._add_auth_token(actual, "", required=True) 204 assert actual == {'Authorization': 'Custom token'} 205 206 207def test_initialise_galaxy(monkeypatch): 208 mock_open = MagicMock() 209 mock_open.side_effect = [ 210 StringIO(u'{"available_versions":{"v1":"v1/"}}'), 211 StringIO(u'{"token":"my token"}'), 212 ] 213 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 214 215 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/") 216 actual = api.authenticate("github_token") 217 218 assert len(api.available_api_versions) == 2 219 assert api.available_api_versions['v1'] == u'v1/' 220 assert api.available_api_versions['v2'] == u'v2/' 221 assert actual == {u'token': u'my token'} 222 assert mock_open.call_count == 2 223 assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/' 224 assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent'] 225 assert mock_open.mock_calls[1][1][0] == 'https://galaxy.ansible.com/api/v1/tokens/' 226 assert 'ansible-galaxy' in mock_open.mock_calls[1][2]['http_agent'] 227 assert mock_open.mock_calls[1][2]['data'] == 'github_token=github_token' 228 229 230def test_initialise_galaxy_with_auth(monkeypatch): 231 mock_open = MagicMock() 232 mock_open.side_effect = [ 233 StringIO(u'{"available_versions":{"v1":"v1/"}}'), 234 StringIO(u'{"token":"my token"}'), 235 ] 236 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 237 238 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=GalaxyToken(token='my_token')) 239 actual = api.authenticate("github_token") 240 241 assert len(api.available_api_versions) == 2 242 assert api.available_api_versions['v1'] == u'v1/' 243 assert api.available_api_versions['v2'] == u'v2/' 244 assert actual == {u'token': u'my token'} 245 assert mock_open.call_count == 2 246 assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/' 247 assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent'] 248 assert mock_open.mock_calls[1][1][0] == 'https://galaxy.ansible.com/api/v1/tokens/' 249 assert 'ansible-galaxy' in mock_open.mock_calls[1][2]['http_agent'] 250 assert mock_open.mock_calls[1][2]['data'] == 'github_token=github_token' 251 252 253def test_initialise_automation_hub(monkeypatch): 254 mock_open = MagicMock() 255 mock_open.side_effect = [ 256 StringIO(u'{"available_versions":{"v2": "v2/", "v3":"v3/"}}'), 257 ] 258 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 259 token = KeycloakToken(auth_url='https://api.test/') 260 mock_token_get = MagicMock() 261 mock_token_get.return_value = 'my_token' 262 monkeypatch.setattr(token, 'get', mock_token_get) 263 264 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) 265 266 assert len(api.available_api_versions) == 2 267 assert api.available_api_versions['v2'] == u'v2/' 268 assert api.available_api_versions['v3'] == u'v3/' 269 270 assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/' 271 assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent'] 272 assert mock_open.mock_calls[0][2]['headers'] == {'Authorization': 'Bearer my_token'} 273 274 275def test_initialise_unknown(monkeypatch): 276 mock_open = MagicMock() 277 mock_open.side_effect = [ 278 urllib_error.HTTPError('https://galaxy.ansible.com/api/', 500, 'msg', {}, StringIO(u'{"msg":"raw error"}')), 279 urllib_error.HTTPError('https://galaxy.ansible.com/api/api/', 500, 'msg', {}, StringIO(u'{"msg":"raw error"}')), 280 ] 281 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 282 283 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=GalaxyToken(token='my_token')) 284 285 expected = "Error when finding available api versions from test (%s) (HTTP Code: 500, Message: msg)" \ 286 % api.api_server 287 with pytest.raises(AnsibleError, match=re.escape(expected)): 288 api.authenticate("github_token") 289 290 291def test_get_available_api_versions(monkeypatch): 292 mock_open = MagicMock() 293 mock_open.side_effect = [ 294 StringIO(u'{"available_versions":{"v1":"v1/","v2":"v2/"}}'), 295 ] 296 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 297 298 api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/") 299 actual = api.available_api_versions 300 assert len(actual) == 2 301 assert actual['v1'] == u'v1/' 302 assert actual['v2'] == u'v2/' 303 304 assert mock_open.call_count == 1 305 assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/' 306 assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent'] 307 308 309def test_publish_collection_missing_file(): 310 fake_path = u'/fake/ÅÑŚÌβŁÈ/path' 311 expected = to_native("The collection path specified '%s' does not exist." % fake_path) 312 313 api = get_test_galaxy_api("https://galaxy.ansible.com/api/", "v2") 314 with pytest.raises(AnsibleError, match=expected): 315 api.publish_collection(fake_path) 316 317 318def test_publish_collection_not_a_tarball(): 319 expected = "The collection path specified '{0}' is not a tarball, use 'ansible-galaxy collection build' to " \ 320 "create a proper release artifact." 321 322 api = get_test_galaxy_api("https://galaxy.ansible.com/api/", "v2") 323 with tempfile.NamedTemporaryFile(prefix=u'ÅÑŚÌβŁÈ') as temp_file: 324 temp_file.write(b"\x00") 325 temp_file.flush() 326 with pytest.raises(AnsibleError, match=expected.format(to_native(temp_file.name))): 327 api.publish_collection(temp_file.name) 328 329 330def test_publish_collection_unsupported_version(): 331 expected = "Galaxy action publish_collection requires API versions 'v2, v3' but only 'v1' are available on test " \ 332 "https://galaxy.ansible.com/api/" 333 334 api = get_test_galaxy_api("https://galaxy.ansible.com/api/", "v1") 335 with pytest.raises(AnsibleError, match=expected): 336 api.publish_collection("path") 337 338 339@pytest.mark.parametrize('api_version, collection_url', [ 340 ('v2', 'collections'), 341 ('v3', 'artifacts/collections'), 342]) 343def test_publish_collection(api_version, collection_url, collection_artifact, monkeypatch): 344 api = get_test_galaxy_api("https://galaxy.ansible.com/api/", api_version) 345 346 mock_call = MagicMock() 347 mock_call.return_value = {'task': 'http://task.url/'} 348 monkeypatch.setattr(api, '_call_galaxy', mock_call) 349 350 actual = api.publish_collection(collection_artifact) 351 assert actual == 'http://task.url/' 352 assert mock_call.call_count == 1 353 assert mock_call.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/%s/%s/' % (api_version, collection_url) 354 assert mock_call.mock_calls[0][2]['headers']['Content-length'] == len(mock_call.mock_calls[0][2]['args']) 355 assert mock_call.mock_calls[0][2]['headers']['Content-type'].startswith( 356 'multipart/form-data; boundary=') 357 assert mock_call.mock_calls[0][2]['args'].startswith(b'--') 358 assert mock_call.mock_calls[0][2]['method'] == 'POST' 359 assert mock_call.mock_calls[0][2]['auth_required'] is True 360 361 362@pytest.mark.parametrize('api_version, collection_url, response, expected', [ 363 ('v2', 'collections', {}, 364 'Error when publishing collection to test (%s) (HTTP Code: 500, Message: msg Code: Unknown)'), 365 ('v2', 'collections', { 366 'message': u'Galaxy error messäge', 367 'code': 'GWE002', 368 }, u'Error when publishing collection to test (%s) (HTTP Code: 500, Message: Galaxy error messäge Code: GWE002)'), 369 ('v3', 'artifact/collections', {}, 370 'Error when publishing collection to test (%s) (HTTP Code: 500, Message: msg Code: Unknown)'), 371 ('v3', 'artifact/collections', { 372 'errors': [ 373 { 374 'code': 'conflict.collection_exists', 375 'detail': 'Collection "mynamespace-mycollection-4.1.1" already exists.', 376 'title': 'Conflict.', 377 'status': '400', 378 }, 379 { 380 'code': 'quantum_improbability', 381 'title': u'Rändom(?) quantum improbability.', 382 'source': {'parameter': 'the_arrow_of_time'}, 383 'meta': {'remediation': 'Try again before'}, 384 }, 385 ], 386 }, u'Error when publishing collection to test (%s) (HTTP Code: 500, Message: Collection ' 387 u'"mynamespace-mycollection-4.1.1" already exists. Code: conflict.collection_exists), (HTTP Code: 500, ' 388 u'Message: Rändom(?) quantum improbability. Code: quantum_improbability)') 389]) 390def test_publish_failure(api_version, collection_url, response, expected, collection_artifact, monkeypatch): 391 api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version) 392 393 expected_url = '%s/api/%s/%s' % (api.api_server, api_version, collection_url) 394 395 mock_open = MagicMock() 396 mock_open.side_effect = urllib_error.HTTPError(expected_url, 500, 'msg', {}, 397 StringIO(to_text(json.dumps(response)))) 398 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 399 400 with pytest.raises(GalaxyError, match=re.escape(to_native(expected % api.api_server))): 401 api.publish_collection(collection_artifact) 402 403 404@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [ 405 ('https://galaxy.server.com/api', 'v2', 'Token', GalaxyToken('my token'), 406 '1234', 407 'https://galaxy.server.com/api/v2/collection-imports/1234/'), 408 ('https://galaxy.server.com/api/automation-hub/', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), 409 '1234', 410 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), 411]) 412def test_wait_import_task(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): 413 api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) 414 415 if token_ins: 416 mock_token_get = MagicMock() 417 mock_token_get.return_value = 'my token' 418 monkeypatch.setattr(token_ins, 'get', mock_token_get) 419 420 mock_open = MagicMock() 421 mock_open.return_value = StringIO(u'{"state":"success","finished_at":"time"}') 422 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 423 424 mock_display = MagicMock() 425 monkeypatch.setattr(Display, 'display', mock_display) 426 427 api.wait_import_task(import_uri) 428 429 assert mock_open.call_count == 1 430 assert mock_open.mock_calls[0][1][0] == full_import_uri 431 assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type 432 433 assert mock_display.call_count == 1 434 assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri 435 436 437@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [ 438 ('https://galaxy.server.com/api/', 'v2', 'Token', GalaxyToken('my token'), 439 '1234', 440 'https://galaxy.server.com/api/v2/collection-imports/1234/'), 441 ('https://galaxy.server.com/api/automation-hub', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), 442 '1234', 443 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), 444]) 445def test_wait_import_task_multiple_requests(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): 446 api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) 447 448 if token_ins: 449 mock_token_get = MagicMock() 450 mock_token_get.return_value = 'my token' 451 monkeypatch.setattr(token_ins, 'get', mock_token_get) 452 453 mock_open = MagicMock() 454 mock_open.side_effect = [ 455 StringIO(u'{"state":"test"}'), 456 StringIO(u'{"state":"success","finished_at":"time"}'), 457 ] 458 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 459 460 mock_display = MagicMock() 461 monkeypatch.setattr(Display, 'display', mock_display) 462 463 mock_vvv = MagicMock() 464 monkeypatch.setattr(Display, 'vvv', mock_vvv) 465 466 monkeypatch.setattr(time, 'sleep', MagicMock()) 467 468 api.wait_import_task(import_uri) 469 470 assert mock_open.call_count == 2 471 assert mock_open.mock_calls[0][1][0] == full_import_uri 472 assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type 473 assert mock_open.mock_calls[1][1][0] == full_import_uri 474 assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s my token' % token_type 475 476 assert mock_display.call_count == 1 477 assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri 478 479 assert mock_vvv.call_count == 1 480 assert mock_vvv.mock_calls[0][1][0] == \ 481 'Galaxy import process has a status of test, wait 2 seconds before trying again' 482 483 484@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri,', [ 485 ('https://galaxy.server.com/api/', 'v2', 'Token', GalaxyToken('my token'), 486 '1234', 487 'https://galaxy.server.com/api/v2/collection-imports/1234/'), 488 ('https://galaxy.server.com/api/automation-hub/', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), 489 '1234', 490 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), 491]) 492def test_wait_import_task_with_failure(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): 493 api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) 494 495 if token_ins: 496 mock_token_get = MagicMock() 497 mock_token_get.return_value = 'my token' 498 monkeypatch.setattr(token_ins, 'get', mock_token_get) 499 500 mock_open = MagicMock() 501 mock_open.side_effect = [ 502 StringIO(to_text(json.dumps({ 503 'finished_at': 'some_time', 504 'state': 'failed', 505 'error': { 506 'code': 'GW001', 507 'description': u'Becäuse I said so!', 508 509 }, 510 'messages': [ 511 { 512 'level': 'error', 513 'message': u'Somé error', 514 }, 515 { 516 'level': 'warning', 517 'message': u'Some wärning', 518 }, 519 { 520 'level': 'info', 521 'message': u'Somé info', 522 }, 523 ], 524 }))), 525 ] 526 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 527 528 mock_display = MagicMock() 529 monkeypatch.setattr(Display, 'display', mock_display) 530 531 mock_vvv = MagicMock() 532 monkeypatch.setattr(Display, 'vvv', mock_vvv) 533 534 mock_warn = MagicMock() 535 monkeypatch.setattr(Display, 'warning', mock_warn) 536 537 mock_err = MagicMock() 538 monkeypatch.setattr(Display, 'error', mock_err) 539 540 expected = to_native(u'Galaxy import process failed: Becäuse I said so! (Code: GW001)') 541 with pytest.raises(AnsibleError, match=re.escape(expected)): 542 api.wait_import_task(import_uri) 543 544 assert mock_open.call_count == 1 545 assert mock_open.mock_calls[0][1][0] == full_import_uri 546 assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type 547 548 assert mock_display.call_count == 1 549 assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri 550 551 assert mock_vvv.call_count == 1 552 assert mock_vvv.mock_calls[0][1][0] == u'Galaxy import message: info - Somé info' 553 554 assert mock_warn.call_count == 1 555 assert mock_warn.mock_calls[0][1][0] == u'Galaxy import warning message: Some wärning' 556 557 assert mock_err.call_count == 1 558 assert mock_err.mock_calls[0][1][0] == u'Galaxy import error message: Somé error' 559 560 561@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [ 562 ('https://galaxy.server.com/api/', 'v2', 'Token', GalaxyToken('my_token'), 563 '1234', 564 'https://galaxy.server.com/api/v2/collection-imports/1234/'), 565 ('https://galaxy.server.com/api/automation-hub/', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), 566 '1234', 567 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), 568]) 569def test_wait_import_task_with_failure_no_error(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): 570 api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) 571 572 if token_ins: 573 mock_token_get = MagicMock() 574 mock_token_get.return_value = 'my token' 575 monkeypatch.setattr(token_ins, 'get', mock_token_get) 576 577 mock_open = MagicMock() 578 mock_open.side_effect = [ 579 StringIO(to_text(json.dumps({ 580 'finished_at': 'some_time', 581 'state': 'failed', 582 'error': {}, 583 'messages': [ 584 { 585 'level': 'error', 586 'message': u'Somé error', 587 }, 588 { 589 'level': 'warning', 590 'message': u'Some wärning', 591 }, 592 { 593 'level': 'info', 594 'message': u'Somé info', 595 }, 596 ], 597 }))), 598 ] 599 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 600 601 mock_display = MagicMock() 602 monkeypatch.setattr(Display, 'display', mock_display) 603 604 mock_vvv = MagicMock() 605 monkeypatch.setattr(Display, 'vvv', mock_vvv) 606 607 mock_warn = MagicMock() 608 monkeypatch.setattr(Display, 'warning', mock_warn) 609 610 mock_err = MagicMock() 611 monkeypatch.setattr(Display, 'error', mock_err) 612 613 expected = 'Galaxy import process failed: Unknown error, see %s for more details \\(Code: UNKNOWN\\)' % full_import_uri 614 with pytest.raises(AnsibleError, match=expected): 615 api.wait_import_task(import_uri) 616 617 assert mock_open.call_count == 1 618 assert mock_open.mock_calls[0][1][0] == full_import_uri 619 assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type 620 621 assert mock_display.call_count == 1 622 assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri 623 624 assert mock_vvv.call_count == 1 625 assert mock_vvv.mock_calls[0][1][0] == u'Galaxy import message: info - Somé info' 626 627 assert mock_warn.call_count == 1 628 assert mock_warn.mock_calls[0][1][0] == u'Galaxy import warning message: Some wärning' 629 630 assert mock_err.call_count == 1 631 assert mock_err.mock_calls[0][1][0] == u'Galaxy import error message: Somé error' 632 633 634@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [ 635 ('https://galaxy.server.com/api', 'v2', 'Token', GalaxyToken('my token'), 636 '1234', 637 'https://galaxy.server.com/api/v2/collection-imports/1234/'), 638 ('https://galaxy.server.com/api/automation-hub', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), 639 '1234', 640 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), 641]) 642def test_wait_import_task_timeout(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): 643 api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) 644 645 if token_ins: 646 mock_token_get = MagicMock() 647 mock_token_get.return_value = 'my token' 648 monkeypatch.setattr(token_ins, 'get', mock_token_get) 649 650 def return_response(*args, **kwargs): 651 return StringIO(u'{"state":"waiting"}') 652 653 mock_open = MagicMock() 654 mock_open.side_effect = return_response 655 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 656 657 mock_display = MagicMock() 658 monkeypatch.setattr(Display, 'display', mock_display) 659 660 mock_vvv = MagicMock() 661 monkeypatch.setattr(Display, 'vvv', mock_vvv) 662 663 monkeypatch.setattr(time, 'sleep', MagicMock()) 664 665 expected = "Timeout while waiting for the Galaxy import process to finish, check progress at '%s'" % full_import_uri 666 with pytest.raises(AnsibleError, match=expected): 667 api.wait_import_task(import_uri, 1) 668 669 assert mock_open.call_count > 1 670 assert mock_open.mock_calls[0][1][0] == full_import_uri 671 assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type 672 assert mock_open.mock_calls[1][1][0] == full_import_uri 673 assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s my token' % token_type 674 675 assert mock_display.call_count == 1 676 assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri 677 678 # expected_wait_msg = 'Galaxy import process has a status of waiting, wait {0} seconds before trying again' 679 assert mock_vvv.call_count > 9 # 1st is opening Galaxy token file. 680 681 # FIXME: 682 # assert mock_vvv.mock_calls[1][1][0] == expected_wait_msg.format(2) 683 # assert mock_vvv.mock_calls[2][1][0] == expected_wait_msg.format(3) 684 # assert mock_vvv.mock_calls[3][1][0] == expected_wait_msg.format(4) 685 # assert mock_vvv.mock_calls[4][1][0] == expected_wait_msg.format(6) 686 # assert mock_vvv.mock_calls[5][1][0] == expected_wait_msg.format(10) 687 # assert mock_vvv.mock_calls[6][1][0] == expected_wait_msg.format(15) 688 # assert mock_vvv.mock_calls[7][1][0] == expected_wait_msg.format(22) 689 # assert mock_vvv.mock_calls[8][1][0] == expected_wait_msg.format(30) 690 691 692@pytest.mark.parametrize('api_version, token_type, version, token_ins', [ 693 ('v2', None, 'v2.1.13', None), 694 ('v3', 'Bearer', 'v1.0.0', KeycloakToken(auth_url='https://api.test/api/automation-hub/')), 695]) 696def test_get_collection_version_metadata_no_version(api_version, token_type, version, token_ins, monkeypatch): 697 api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins) 698 699 if token_ins: 700 mock_token_get = MagicMock() 701 mock_token_get.return_value = 'my token' 702 monkeypatch.setattr(token_ins, 'get', mock_token_get) 703 704 mock_open = MagicMock() 705 mock_open.side_effect = [ 706 StringIO(to_text(json.dumps({ 707 'download_url': 'https://downloadme.com', 708 'artifact': { 709 'sha256': 'ac47b6fac117d7c171812750dacda655b04533cf56b31080b82d1c0db3c9d80f', 710 }, 711 'namespace': { 712 'name': 'namespace', 713 }, 714 'collection': { 715 'name': 'collection', 716 }, 717 'version': version, 718 'metadata': { 719 'dependencies': {}, 720 } 721 }))), 722 ] 723 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 724 725 actual = api.get_collection_version_metadata('namespace', 'collection', version) 726 727 assert isinstance(actual, CollectionVersionMetadata) 728 assert actual.namespace == u'namespace' 729 assert actual.name == u'collection' 730 assert actual.download_url == u'https://downloadme.com' 731 assert actual.artifact_sha256 == u'ac47b6fac117d7c171812750dacda655b04533cf56b31080b82d1c0db3c9d80f' 732 assert actual.version == version 733 assert actual.dependencies == {} 734 735 assert mock_open.call_count == 1 736 assert mock_open.mock_calls[0][1][0] == '%s%s/collections/namespace/collection/versions/%s/' \ 737 % (api.api_server, api_version, version) 738 739 # v2 calls dont need auth, so no authz header or token_type 740 if token_type: 741 assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type 742 743 744@pytest.mark.parametrize('api_version, token_type, token_ins, response', [ 745 ('v2', None, None, { 746 'count': 2, 747 'next': None, 748 'previous': None, 749 'results': [ 750 { 751 'version': '1.0.0', 752 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.0', 753 }, 754 { 755 'version': '1.0.1', 756 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.1', 757 }, 758 ], 759 }), 760 # TODO: Verify this once Automation Hub is actually out 761 ('v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), { 762 'count': 2, 763 'next': None, 764 'previous': None, 765 'data': [ 766 { 767 'version': '1.0.0', 768 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.0', 769 }, 770 { 771 'version': '1.0.1', 772 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.1', 773 }, 774 ], 775 }), 776]) 777def test_get_collection_versions(api_version, token_type, token_ins, response, monkeypatch): 778 api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins) 779 780 if token_ins: 781 mock_token_get = MagicMock() 782 mock_token_get.return_value = 'my token' 783 monkeypatch.setattr(token_ins, 'get', mock_token_get) 784 785 mock_open = MagicMock() 786 mock_open.side_effect = [ 787 StringIO(to_text(json.dumps(response))), 788 ] 789 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 790 791 actual = api.get_collection_versions('namespace', 'collection') 792 assert actual == [u'1.0.0', u'1.0.1'] 793 794 page_query = '?limit=100' if api_version == 'v3' else '?page_size=100' 795 assert mock_open.call_count == 1 796 assert mock_open.mock_calls[0][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \ 797 'versions/%s' % (api_version, page_query) 798 if token_ins: 799 assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type 800 801 802@pytest.mark.parametrize('api_version, token_type, token_ins, responses', [ 803 ('v2', None, None, [ 804 { 805 'count': 6, 806 'next': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/?page=2&page_size=100', 807 'previous': None, 808 'results': [ # Pay no mind, using more manageable results than page_size would indicate 809 { 810 'version': '1.0.0', 811 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.0', 812 }, 813 { 814 'version': '1.0.1', 815 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.1', 816 }, 817 ], 818 }, 819 { 820 'count': 6, 821 'next': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/?page=3&page_size=100', 822 'previous': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions', 823 'results': [ 824 { 825 'version': '1.0.2', 826 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.2', 827 }, 828 { 829 'version': '1.0.3', 830 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.3', 831 }, 832 ], 833 }, 834 { 835 'count': 6, 836 'next': None, 837 'previous': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/?page=2&page_size=100', 838 'results': [ 839 { 840 'version': '1.0.4', 841 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.4', 842 }, 843 { 844 'version': '1.0.5', 845 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.5', 846 }, 847 ], 848 }, 849 ]), 850 ('v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), [ 851 { 852 'count': 6, 853 'links': { 854 # v3 links are relative and the limit is included during pagination 855 'next': '/api/v3/collections/namespace/collection/versions/?limit=100&offset=100', 856 'previous': None, 857 }, 858 'data': [ 859 { 860 'version': '1.0.0', 861 'href': '/api/v3/collections/namespace/collection/versions/1.0.0', 862 }, 863 { 864 'version': '1.0.1', 865 'href': '/api/v3/collections/namespace/collection/versions/1.0.1', 866 }, 867 ], 868 }, 869 { 870 'count': 6, 871 'links': { 872 'next': '/api/v3/collections/namespace/collection/versions/?limit=100&offset=200', 873 'previous': '/api/v3/collections/namespace/collection/versions', 874 }, 875 'data': [ 876 { 877 'version': '1.0.2', 878 'href': '/api/v3/collections/namespace/collection/versions/1.0.2', 879 }, 880 { 881 'version': '1.0.3', 882 'href': '/api/v3/collections/namespace/collection/versions/1.0.3', 883 }, 884 ], 885 }, 886 { 887 'count': 6, 888 'links': { 889 'next': None, 890 'previous': '/api/v3/collections/namespace/collection/versions/?limit=100&offset=100', 891 }, 892 'data': [ 893 { 894 'version': '1.0.4', 895 'href': '/api/v3/collections/namespace/collection/versions/1.0.4', 896 }, 897 { 898 'version': '1.0.5', 899 'href': '/api/v3/collections/namespace/collection/versions/1.0.5', 900 }, 901 ], 902 }, 903 ]), 904]) 905def test_get_collection_versions_pagination(api_version, token_type, token_ins, responses, monkeypatch): 906 api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins) 907 908 if token_ins: 909 mock_token_get = MagicMock() 910 mock_token_get.return_value = 'my token' 911 monkeypatch.setattr(token_ins, 'get', mock_token_get) 912 913 mock_open = MagicMock() 914 mock_open.side_effect = [StringIO(to_text(json.dumps(r))) for r in responses] 915 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 916 917 actual = api.get_collection_versions('namespace', 'collection') 918 assert actual == [u'1.0.0', u'1.0.1', u'1.0.2', u'1.0.3', u'1.0.4', u'1.0.5'] 919 920 assert mock_open.call_count == 3 921 922 if api_version == 'v3': 923 query_1 = 'limit=100' 924 query_2 = 'limit=100&offset=100' 925 query_3 = 'limit=100&offset=200' 926 else: 927 query_1 = 'page_size=100' 928 query_2 = 'page=2&page_size=100' 929 query_3 = 'page=3&page_size=100' 930 931 assert mock_open.mock_calls[0][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \ 932 'versions/?%s' % (api_version, query_1) 933 assert mock_open.mock_calls[1][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \ 934 'versions/?%s' % (api_version, query_2) 935 assert mock_open.mock_calls[2][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \ 936 'versions/?%s' % (api_version, query_3) 937 938 if token_type: 939 assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type 940 assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s my token' % token_type 941 assert mock_open.mock_calls[2][2]['headers']['Authorization'] == '%s my token' % token_type 942 943 944@pytest.mark.parametrize('responses', [ 945 [ 946 { 947 'count': 2, 948 'results': [{'name': '3.5.1', }, {'name': '3.5.2'}], 949 'next_link': None, 950 'next': None, 951 'previous_link': None, 952 'previous': None 953 }, 954 ], 955 [ 956 { 957 'count': 2, 958 'results': [{'name': '3.5.1'}], 959 'next_link': '/api/v1/roles/432/versions/?page=2&page_size=50', 960 'next': '/roles/432/versions/?page=2&page_size=50', 961 'previous_link': None, 962 'previous': None 963 }, 964 { 965 'count': 2, 966 'results': [{'name': '3.5.2'}], 967 'next_link': None, 968 'next': None, 969 'previous_link': '/api/v1/roles/432/versions/?&page_size=50', 970 'previous': '/roles/432/versions/?page_size=50', 971 }, 972 ] 973]) 974def test_get_role_versions_pagination(monkeypatch, responses): 975 api = get_test_galaxy_api('https://galaxy.com/api/', 'v1') 976 977 mock_open = MagicMock() 978 mock_open.side_effect = [StringIO(to_text(json.dumps(r))) for r in responses] 979 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 980 981 actual = api.fetch_role_related('versions', 432) 982 assert actual == [{'name': '3.5.1'}, {'name': '3.5.2'}] 983 984 assert mock_open.call_count == len(responses) 985 986 assert mock_open.mock_calls[0][1][0] == 'https://galaxy.com/api/v1/roles/432/versions/?page_size=50' 987 if len(responses) == 2: 988 assert mock_open.mock_calls[1][1][0] == 'https://galaxy.com/api/v1/roles/432/versions/?page=2&page_size=50' 989 990 991def test_missing_cache_dir(cache_dir): 992 os.rmdir(cache_dir) 993 GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False) 994 995 assert os.path.isdir(cache_dir) 996 assert stat.S_IMODE(os.stat(cache_dir).st_mode) == 0o700 997 998 cache_file = os.path.join(cache_dir, 'api.json') 999 with open(cache_file) as fd: 1000 actual_cache = fd.read() 1001 assert actual_cache == '{"version": 1}' 1002 assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o600 1003 1004 1005def test_existing_cache(cache_dir): 1006 cache_file = os.path.join(cache_dir, 'api.json') 1007 cache_file_contents = '{"version": 1, "test": "json"}' 1008 with open(cache_file, mode='w') as fd: 1009 fd.write(cache_file_contents) 1010 os.chmod(cache_file, 0o655) 1011 1012 GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False) 1013 1014 assert os.path.isdir(cache_dir) 1015 with open(cache_file) as fd: 1016 actual_cache = fd.read() 1017 assert actual_cache == cache_file_contents 1018 assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o655 1019 1020 1021@pytest.mark.parametrize('content', [ 1022 '', 1023 'value', 1024 '{"de" "finit" "ely" [\'invalid"]}', 1025 '[]', 1026 '{"version": 2, "test": "json"}', 1027 '{"version": 2, "key": "ÅÑŚÌβŁÈ"}', 1028]) 1029def test_cache_invalid_cache_content(content, cache_dir): 1030 cache_file = os.path.join(cache_dir, 'api.json') 1031 with open(cache_file, mode='w') as fd: 1032 fd.write(content) 1033 os.chmod(cache_file, 0o664) 1034 1035 GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False) 1036 1037 with open(cache_file) as fd: 1038 actual_cache = fd.read() 1039 assert actual_cache == '{"version": 1}' 1040 assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o664 1041 1042 1043def test_cache_complete_pagination(cache_dir, monkeypatch): 1044 1045 responses = get_collection_versions() 1046 cache_file = os.path.join(cache_dir, 'api.json') 1047 1048 api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v2', no_cache=False) 1049 1050 mock_open = MagicMock( 1051 side_effect=[ 1052 StringIO(to_text(json.dumps(r))) 1053 for r in responses 1054 ] 1055 ) 1056 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 1057 1058 actual_versions = api.get_collection_versions('namespace', 'collection') 1059 assert actual_versions == [u'1.0.0', u'1.0.1', u'1.0.2', u'1.0.3', u'1.0.4', u'1.0.5'] 1060 1061 with open(cache_file) as fd: 1062 final_cache = json.loads(fd.read()) 1063 1064 cached_server = final_cache['galaxy.server.com:'] 1065 cached_collection = cached_server['/api/v2/collections/namespace/collection/versions/'] 1066 cached_versions = [r['version'] for r in cached_collection['results']] 1067 1068 assert final_cache == api._cache 1069 assert cached_versions == actual_versions 1070 1071 1072def test_cache_flaky_pagination(cache_dir, monkeypatch): 1073 1074 responses = get_collection_versions() 1075 cache_file = os.path.join(cache_dir, 'api.json') 1076 1077 api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v2', no_cache=False) 1078 1079 # First attempt, fail midway through 1080 mock_open = MagicMock( 1081 side_effect=[ 1082 StringIO(to_text(json.dumps(responses[0]))), 1083 StringIO(to_text(json.dumps(responses[1]))), 1084 urllib_error.HTTPError(responses[1]['next'], 500, 'Error', {}, StringIO()), 1085 StringIO(to_text(json.dumps(responses[3]))), 1086 ] 1087 ) 1088 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 1089 1090 expected = ( 1091 r'Error when getting available collection versions for namespace\.collection ' 1092 r'from test \(https://galaxy\.server\.com/api/\) ' 1093 r'\(HTTP Code: 500, Message: Error Code: Unknown\)' 1094 ) 1095 with pytest.raises(GalaxyError, match=expected): 1096 api.get_collection_versions('namespace', 'collection') 1097 1098 with open(cache_file) as fd: 1099 final_cache = json.loads(fd.read()) 1100 1101 assert final_cache == { 1102 'version': 1, 1103 'galaxy.server.com:': { 1104 'modified': { 1105 'namespace.collection': responses[0]['modified'] 1106 } 1107 } 1108 } 1109 1110 # Reset API 1111 api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v2', no_cache=False) 1112 1113 # Second attempt is successful so cache should be populated 1114 mock_open = MagicMock( 1115 side_effect=[ 1116 StringIO(to_text(json.dumps(r))) 1117 for r in responses 1118 ] 1119 ) 1120 monkeypatch.setattr(galaxy_api, 'open_url', mock_open) 1121 1122 actual_versions = api.get_collection_versions('namespace', 'collection') 1123 assert actual_versions == [u'1.0.0', u'1.0.1', u'1.0.2', u'1.0.3', u'1.0.4', u'1.0.5'] 1124 1125 with open(cache_file) as fd: 1126 final_cache = json.loads(fd.read()) 1127 1128 cached_server = final_cache['galaxy.server.com:'] 1129 cached_collection = cached_server['/api/v2/collections/namespace/collection/versions/'] 1130 cached_versions = [r['version'] for r in cached_collection['results']] 1131 1132 assert cached_versions == actual_versions 1133 1134 1135def test_world_writable_cache(cache_dir, monkeypatch): 1136 mock_warning = MagicMock() 1137 monkeypatch.setattr(Display, 'warning', mock_warning) 1138 1139 cache_file = os.path.join(cache_dir, 'api.json') 1140 with open(cache_file, mode='w') as fd: 1141 fd.write('{"version": 2}') 1142 os.chmod(cache_file, 0o666) 1143 1144 api = GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False) 1145 assert api._cache is None 1146 1147 with open(cache_file) as fd: 1148 actual_cache = fd.read() 1149 assert actual_cache == '{"version": 2}' 1150 assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o666 1151 1152 assert mock_warning.call_count == 1 1153 assert mock_warning.call_args[0][0] == \ 1154 'Galaxy cache has world writable access (%s), ignoring it as a cache source.' % cache_file 1155 1156 1157def test_no_cache(cache_dir): 1158 cache_file = os.path.join(cache_dir, 'api.json') 1159 with open(cache_file, mode='w') as fd: 1160 fd.write('random') 1161 1162 api = GalaxyAPI(None, "test", 'https://galaxy.ansible.com/') 1163 assert api._cache is None 1164 1165 with open(cache_file) as fd: 1166 actual_cache = fd.read() 1167 assert actual_cache == 'random' 1168 1169 1170def test_clear_cache_with_no_cache(cache_dir): 1171 cache_file = os.path.join(cache_dir, 'api.json') 1172 with open(cache_file, mode='w') as fd: 1173 fd.write('{"version": 1, "key": "value"}') 1174 1175 GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', clear_response_cache=True) 1176 assert not os.path.exists(cache_file) 1177 1178 1179def test_clear_cache(cache_dir): 1180 cache_file = os.path.join(cache_dir, 'api.json') 1181 with open(cache_file, mode='w') as fd: 1182 fd.write('{"version": 1, "key": "value"}') 1183 1184 GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', clear_response_cache=True, no_cache=False) 1185 1186 with open(cache_file) as fd: 1187 actual_cache = fd.read() 1188 assert actual_cache == '{"version": 1}' 1189 assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o600 1190 1191 1192@pytest.mark.parametrize(['url', 'expected'], [ 1193 ('http://hostname/path', 'hostname:'), 1194 ('http://hostname:80/path', 'hostname:80'), 1195 ('https://testing.com:invalid', 'testing.com:'), 1196 ('https://testing.com:1234', 'testing.com:1234'), 1197 ('https://username:password@testing.com/path', 'testing.com:'), 1198 ('https://username:password@testing.com:443/path', 'testing.com:443'), 1199]) 1200def test_cache_id(url, expected): 1201 actual = galaxy_api.get_cache_id(url) 1202 assert actual == expected 1203