Differential D4269 Diff 15098 swh/icinga_plugins/tests/test_deposit.py

Changeset View

Standalone View

swh/icinga_plugins/tests/test_deposit.py

# Copyright (C) 2019 The Software Heritage developers		# Copyright (C) 2019-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

import io		import io
import os		import os
import tarfile		import tarfile
import time		import time
		from typing import Optional

from click.testing import CliRunner		from click.testing import CliRunner
import pytest		import pytest

from swh.icinga_plugins.cli import icinga_cli_group		from swh.icinga_plugins.cli import icinga_cli_group

from .web_scenario import WebScenario		from .web_scenario import WebScenario

Show All 40 Lines


STATUS_TEMPLATE = """		STATUS_TEMPLATE = """
<entry xmlns="http://www.w3.org/2005/Atom"		<entry xmlns="http://www.w3.org/2005/Atom"
xmlns:sword="http://purl.org/net/sword/"		xmlns:sword="http://purl.org/net/sword/"
xmlns:dcterms="http://purl.org/dc/terms/">		xmlns:dcterms="http://purl.org/dc/terms/">
<deposit_id>42</deposit_id>		<deposit_id>42</deposit_id>
<deposit_status>{status}</deposit_status>		<deposit_status>{status}</deposit_status>
<deposit_status_detail>{status_detail}</deposit_status_detail>		<deposit_status_detail>{status_detail}</deposit_status_detail>%s
</entry>		</entry>
"""		"""


		def status_template(
		status: str, status_detail: str = "", swhid: Optional[str] = None
		) -> str:
		"""Generate a proper status template out of status, status_detail and optional swhid

		"""
		if swhid is not None:
		template = STATUS_TEMPLATE % f"\n <deposit_swh_id>{swhid}</deposit_swh_id>"
		return template.format(status=status, status_detail=status_detail, swhid=swhid)
		template = STATUS_TEMPLATE % ""
		return template.format(status=status, status_detail=status_detail)


		def test_status_template():
		actual_status = status_template(status="deposited")
		assert (
		actual_status
		== """
		<entry xmlns="http://www.w3.org/2005/Atom"
		xmlns:sword="http://purl.org/net/sword/"
		xmlns:dcterms="http://purl.org/dc/terms/">
		<deposit_id>42</deposit_id>
		<deposit_status>deposited</deposit_status>
		<deposit_status_detail></deposit_status_detail>
		</entry>
		"""
		)

		actual_status = status_template(status="verified", status_detail="detail")
		assert (
		actual_status
		== """
		<entry xmlns="http://www.w3.org/2005/Atom"
		xmlns:sword="http://purl.org/net/sword/"
		xmlns:dcterms="http://purl.org/dc/terms/">
		<deposit_id>42</deposit_id>
		<deposit_status>verified</deposit_status>
		<deposit_status_detail>detail</deposit_status_detail>
		</entry>
		"""
		)

		actual_status = status_template(status="done", swhid="10")
		assert (
		actual_status
		== """
		<entry xmlns="http://www.w3.org/2005/Atom"
		xmlns:sword="http://purl.org/net/sword/"
		xmlns:dcterms="http://purl.org/dc/terms/">
		<deposit_id>42</deposit_id>
		<deposit_status>done</deposit_status>
		<deposit_status_detail></deposit_status_detail>
		<deposit_swh_id>10</deposit_swh_id>
		</entry>
		"""
		)


@pytest.fixture(scope="session")		@pytest.fixture(scope="session")
def tmp_path(tmp_path_factory):		def tmp_path(tmp_path_factory):
return tmp_path_factory.mktemp(__name__)		return tmp_path_factory.mktemp(__name__)


@pytest.fixture(scope="session")		@pytest.fixture(scope="session")
def sample_metadata(tmp_path):		def sample_metadata(tmp_path):
"""Returns a sample metadata file's path		"""Returns a sample metadata file's path
▲ Show 20 Lines • Show All 66 Lines • ▼ Show 20 Lines	def test_deposit_delays(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time		requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):		):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step(		scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")		"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="verified", status_detail=""),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="loading"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="loading", status_detail=""),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="done"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="done", status_detail=""),
)		)

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

result = invoke(		result = invoke(
[		[
"check-deposit",		"check-deposit",
*COMMON_OPTIONS,		*COMMON_OPTIONS,
Show All 19 Lines	def test_deposit_delay_warning(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time		requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):		):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step(		scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")		"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="verified", status_detail=""),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="done"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="done", status_detail=""),
)		)

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

result = invoke(		result = invoke(
[		[
"--warning",		"--warning",
"15",		"15",
Show All 22 Lines	def test_deposit_delay_critical(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time		requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):		):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step(		scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")		"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="verified", status_detail=""),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get",
f"{BASE_URL}/testcol/42/status/",		f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="done", status_detail=""),		status_template(status="done"),
callback=lambda: time.sleep(60),		callback=lambda: time.sleep(60),
)		)

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

result = invoke(		result = invoke(
[		[
"--critical",		"--critical",
Show All 28 Lines	scenario.add_step(
"post",		"post",
f"{BASE_URL}/testcol/",		f"{BASE_URL}/testcol/",
ENTRY_TEMPLATE.format(status="deposited"),		ENTRY_TEMPLATE.format(status="deposited"),
callback=lambda: time.sleep(1500),		callback=lambda: time.sleep(1500),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get",
f"{BASE_URL}/testcol/42/status/",		f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="verified", status_detail=""),		status_template(status="verified"),
callback=lambda: time.sleep(1500),		callback=lambda: time.sleep(1500),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get",
f"{BASE_URL}/testcol/42/status/",		f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="loading", status_detail=""),		status_template(status="loading"),
callback=lambda: time.sleep(1500),		callback=lambda: time.sleep(1500),
)		)

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

result = invoke(		result = invoke(
[		[
"check-deposit",		"check-deposit",
Show All 23 Lines	):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step(		scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")		"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)		)
scenario.add_step(		scenario.add_step(
"get",		"get",
f"{BASE_URL}/testcol/42/status/",		f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="rejected", status_detail="booo"),		status_template(status="rejected", status_detail="booo"),
)		)

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

result = invoke(		result = invoke(
[		[
"check-deposit",		"check-deposit",
*COMMON_OPTIONS,		*COMMON_OPTIONS,
Show All 19 Lines	def test_deposit_failed(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time		requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):		):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step(		scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")		"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="verified", status_detail=""),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="loading"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="loading", status_detail=""),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get",
f"{BASE_URL}/testcol/42/status/",		f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="failed", status_detail="booo"),		status_template(status="failed", status_detail="booo"),
)		)

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

result = invoke(		result = invoke(
[		[
"check-deposit",		"check-deposit",
*COMMON_OPTIONS,		*COMMON_OPTIONS,
Show All 20 Lines	def test_deposit_unexpected_status(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time		requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):		):
scenario = WebScenario()		scenario = WebScenario()

scenario.add_step(		scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")		"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="verified", status_detail=""),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get", f"{BASE_URL}/testcol/42/status/", status_template(status="loading"),
f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="loading", status_detail=""),
)		)
scenario.add_step(		scenario.add_step(
"get",		"get",
f"{BASE_URL}/testcol/42/status/",		f"{BASE_URL}/testcol/42/status/",
STATUS_TEMPLATE.format(status="what", status_detail="booo"),		status_template(status="what", status_detail="booo"),
)		)

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

result = invoke(		result = invoke(
[		[
"check-deposit",		"check-deposit",
*COMMON_OPTIONS,		*COMMON_OPTIONS,
Show All 17 Lines