Differential D7300 Diff 26419 swh/icinga_plugins/tests/test_vault.py

Changeset View

Standalone View

swh/icinga_plugins/tests/test_vault.py

Show First 20 Lines • Show All 59 Lines • ▼ Show 20 Lines


def test_vault_immediate_success(requests_mock, mocker, mocked_time):		def test_vault_immediate_success(requests_mock, mocker, mocked_time):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step("get", url_api, {}, status_code=404)		scenario.add_step("get", url_api, {}, status_code=404)
scenario.add_step("post", url_api, response_pending)		scenario.add_step("post", url_api, response_pending)
scenario.add_step("get", url_api, response_done)		scenario.add_step("get", url_api, response_done)
scenario.add_step("get", url_fetch, "xx" * 40)		scenario.add_step(
		"get", url_fetch, "xx" * 40, headers={"Content-Type": "application/gzip"}
		)

scenario.install_mock(requests_mock)		scenario.install_mock(requests_mock)

get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")		get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")
get_storage_mock.side_effect = FakeStorage		get_storage_mock.side_effect = FakeStorage

result = invoke(		result = invoke(
[		[
Show All 16 Lines

def test_vault_delayed_success(requests_mock, mocker, mocked_time):		def test_vault_delayed_success(requests_mock, mocker, mocked_time):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step("get", url_api, {}, status_code=404)		scenario.add_step("get", url_api, {}, status_code=404)
scenario.add_step("post", url_api, response_pending)		scenario.add_step("post", url_api, response_pending)
scenario.add_step("get", url_api, response_pending)		scenario.add_step("get", url_api, response_pending)
scenario.add_step("get", url_api, response_done)		scenario.add_step("get", url_api, response_done)
scenario.add_step("get", url_fetch, "xx" * 40)		scenario.add_step(
		"get", url_fetch, "xx" * 40, headers={"Content-Type": "application/gzip"}
		)

scenario.install_mock(requests_mock)		scenario.install_mock(requests_mock)

get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")		get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")
get_storage_mock.side_effect = FakeStorage		get_storage_mock.side_effect = FakeStorage

result = invoke(		result = invoke(
[		[
▲ Show 20 Lines • Show All 117 Lines • ▼ Show 20 Lines	def test_vault_cached_directory(requests_mock, mocker, mocked_time):
"""First serves a directory that's already in the cache, to		"""First serves a directory that's already in the cache, to
test that vault_check requests another one."""		test that vault_check requests another one."""
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step("get", url_api, {}, status_code=200)		scenario.add_step("get", url_api, {}, status_code=200)
scenario.add_step("get", url_api, {}, status_code=404)		scenario.add_step("get", url_api, {}, status_code=404)
scenario.add_step("post", url_api, response_pending)		scenario.add_step("post", url_api, response_pending)
scenario.add_step("get", url_api, response_done)		scenario.add_step("get", url_api, response_done)
scenario.add_step("get", url_fetch, "xx" * 40)		scenario.add_step(
		"get", url_fetch, "xx" * 40, headers={"Content-Type": "application/gzip"}
		)

scenario.install_mock(requests_mock)		scenario.install_mock(requests_mock)

get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")		get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")
get_storage_mock.side_effect = FakeStorage		get_storage_mock.side_effect = FakeStorage

result = invoke(		result = invoke(
[		[
Show All 34 Lines	result = invoke(
],		],
catch_exceptions=True,		catch_exceptions=True,
)		)

assert result.output == ("VAULT CRITICAL - No directory exists in the archive.\n")		assert result.output == ("VAULT CRITICAL - No directory exists in the archive.\n")
assert result.exit_code == 2, result.output		assert result.exit_code == 2, result.output


def test_vault_immediate_success_but_fetch_failed(requests_mock, mocker, mocked_time):		def test_vault_fetch_failed(requests_mock, mocker, mocked_time):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step("get", url_api, {}, status_code=404)		scenario.add_step("get", url_api, {}, status_code=404)
scenario.add_step("post", url_api, response_pending)		scenario.add_step("post", url_api, response_pending)
scenario.add_step("get", url_api, response_done)		scenario.add_step("get", url_api, response_done)
scenario.add_step("get", url_fetch, "", status_code=500)		scenario.add_step(
		"get",
		url_fetch,
		"",
		status_code=500,
		olasdUnsubmitted Not Done Inline Actions hmm, wouldn't a 500 error be plaintext or html rather than gzip? olasd: hmm, wouldn't a 500 error be plaintext or html rather than gzip?
		vlorentzAuthorUnsubmitted Done Inline Actions indeed, that wasn't intentional. But it doesn't matter much vlorentz: indeed, that wasn't intentional. But it doesn't matter much
		headers={"Content-Type": "application/gzip"},
		)

scenario.install_mock(requests_mock)		scenario.install_mock(requests_mock)

get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")		get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")
get_storage_mock.side_effect = FakeStorage		get_storage_mock.side_effect = FakeStorage

result = invoke(		result = invoke(
[		[
Show All 10 Lines	def test_vault_fetch_failed(requests_mock, mocker, mocked_time):
assert result.output == (		assert result.output == (
f"VAULT CRITICAL - cooking directory {DIR_ID} took "		f"VAULT CRITICAL - cooking directory {DIR_ID} took "
f"10.00s and succeeded, but fetch failed with status code 500.\n"		f"10.00s and succeeded, but fetch failed with status code 500.\n"
f"\| 'total_time' = 10.00s\n"		f"\| 'total_time' = 10.00s\n"
)		)
assert result.exit_code == 2, result.output		assert result.exit_code == 2, result.output


def test_vault_immediate_success_but_fetch_empty(requests_mock, mocker, mocked_time):		def test_vault_fetch_empty(requests_mock, mocker, mocked_time):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step("get", url_api, {}, status_code=404)		scenario.add_step("get", url_api, {}, status_code=404)
scenario.add_step("post", url_api, response_pending)		scenario.add_step("post", url_api, response_pending)
scenario.add_step("get", url_api, response_done)		scenario.add_step("get", url_api, response_done)
scenario.add_step("get", url_fetch, "")		scenario.add_step(
		"get", url_fetch, "", headers={"Content-Type": "application/gzip"}
		)

scenario.install_mock(requests_mock)		scenario.install_mock(requests_mock)

get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")		get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")
get_storage_mock.side_effect = FakeStorage		get_storage_mock.side_effect = FakeStorage

result = invoke(		result = invoke(
[		[
Show All 10 Lines	def test_vault_fetch_empty(requests_mock, mocker, mocked_time):
assert result.output == (		assert result.output == (
f"VAULT CRITICAL - cooking directory {DIR_ID} took "		f"VAULT CRITICAL - cooking directory {DIR_ID} took "
f"10.00s and succeeded, but fetch was empty.\n"		f"10.00s and succeeded, but fetch was empty.\n"
f"\| 'total_time' = 10.00s\n"		f"\| 'total_time' = 10.00s\n"
)		)
assert result.exit_code == 2, result.output		assert result.exit_code == 2, result.output


def test_vault_immediate_success_but_no_fetch_url(requests_mock, mocker, mocked_time):		def test_vault_fetch_missing_content_type(requests_mock, mocker, mocked_time):
		scenario = WebScenario()

		scenario.add_step("get", url_api, {}, status_code=404)
		scenario.add_step("post", url_api, response_pending)
		scenario.add_step("get", url_api, response_done)
		scenario.add_step("get", url_fetch, "")

		scenario.install_mock(requests_mock)

		get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")
		get_storage_mock.side_effect = FakeStorage

		result = invoke(
		[
		"check-vault",
		"--swh-web-url",
		"mock://swh-web.example.org",
		"--swh-storage-url",
		"foo://example.org",
		"directory",
		],
		catch_exceptions=True,
		)

		assert result.output == (
		"VAULT CRITICAL - Unexpected Content-Type when downloading bundle: None\n"
		"\| 'total_time' = 10.00s\n"
		)
		assert result.exit_code == 2, result.output


		def test_vault_no_fetch_url(requests_mock, mocker, mocked_time):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step("get", url_api, {}, status_code=404)		scenario.add_step("get", url_api, {}, status_code=404)
scenario.add_step("post", url_api, response_pending)		scenario.add_step("post", url_api, response_pending)
scenario.add_step("get", url_api, response_done_no_fetch)		scenario.add_step("get", url_api, response_done_no_fetch)

scenario.install_mock(requests_mock)		scenario.install_mock(requests_mock)

Show All 21 Lines