diff --git a/swh/icinga_plugins/tests/test_deposit.py b/swh/icinga_plugins/tests/test_deposit.py
index 60658c4..f0f5ca3 100644
--- a/swh/icinga_plugins/tests/test_deposit.py
+++ b/swh/icinga_plugins/tests/test_deposit.py
@@ -1,467 +1,506 @@
-# 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
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
import io
import os
import tarfile
import time
+from typing import Optional
from click.testing import CliRunner
import pytest
from swh.icinga_plugins.cli import icinga_cli_group
from .web_scenario import WebScenario
BASE_URL = "http://swh-deposit.example.org/1"
COMMON_OPTIONS = [
"--server",
BASE_URL,
"--username",
"test",
"--password",
"test",
"--collection",
"testcol",
]
SAMPLE_METADATA = """
Test Software
swh
test-software
No One
"""
ENTRY_TEMPLATE = """
42
2019-12-19 18:11:00
foo.tar.gz
{status}
http://purl.org/net/sword/package/SimpleZip
"""
STATUS_TEMPLATE = """
42
{status}
- {status_detail}
+ {status_detail}%s
"""
+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 {swhid}"
+ 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
+ == """
+
+ 42
+ deposited
+
+
+"""
+ )
+
+ actual_status = status_template(status="verified", status_detail="detail")
+ assert (
+ actual_status
+ == """
+
+ 42
+ verified
+ detail
+
+"""
+ )
+
+ actual_status = status_template(status="done", swhid="10")
+ assert (
+ actual_status
+ == """
+
+ 42
+ done
+
+ 10
+
+"""
+ )
+
+
@pytest.fixture(scope="session")
def tmp_path(tmp_path_factory):
return tmp_path_factory.mktemp(__name__)
@pytest.fixture(scope="session")
def sample_metadata(tmp_path):
"""Returns a sample metadata file's path
"""
path = os.path.join(tmp_path, "metadata.xml")
with open(path, "w") as fd:
fd.write(SAMPLE_METADATA)
return path
@pytest.fixture(scope="session")
def sample_archive(tmp_path):
"""Returns a sample archive's path
"""
path = os.path.join(tmp_path, "archive.tar.gz")
with tarfile.open(path, "w:gz") as tf:
tf.addfile(tarfile.TarInfo("hello.py"), io.BytesIO(b'print("Hello world")'))
return path
def invoke(args, catch_exceptions=False):
runner = CliRunner()
result = runner.invoke(icinga_cli_group, args)
if not catch_exceptions and result.exception:
print(result.output)
raise result.exception
return result
def test_deposit_immediate_success(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):
scenario = WebScenario()
scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="done")
)
scenario.install_mock(requests_mock)
result = invoke(
[
"check-deposit",
*COMMON_OPTIONS,
"single",
"--archive",
sample_archive,
"--metadata",
sample_metadata,
]
)
assert result.output == (
"DEPOSIT OK - Deposit took 0.00s and succeeded.\n"
"| 'load_time' = 0.00s\n"
"| 'total_time' = 0.00s\n"
"| 'upload_time' = 0.00s\n"
"| 'validation_time' = 0.00s\n"
)
assert result.exit_code == 0, result.output
def test_deposit_delays(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):
scenario = WebScenario()
scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="verified", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="loading", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="loading"),
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="done", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="done"),
)
scenario.install_mock(requests_mock)
result = invoke(
[
"check-deposit",
*COMMON_OPTIONS,
"single",
"--archive",
sample_archive,
"--metadata",
sample_metadata,
]
)
assert result.output == (
"DEPOSIT OK - Deposit took 30.00s and succeeded.\n"
"| 'load_time' = 20.00s\n"
"| 'total_time' = 30.00s\n"
"| 'upload_time' = 0.00s\n"
"| 'validation_time' = 10.00s\n"
)
assert result.exit_code == 0, result.output
def test_deposit_delay_warning(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):
scenario = WebScenario()
scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="verified", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="done", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="done"),
)
scenario.install_mock(requests_mock)
result = invoke(
[
"--warning",
"15",
"check-deposit",
*COMMON_OPTIONS,
"single",
"--archive",
sample_archive,
"--metadata",
sample_metadata,
],
catch_exceptions=True,
)
assert result.output == (
"DEPOSIT WARNING - Deposit took 20.00s and succeeded.\n"
"| 'load_time' = 10.00s\n"
"| 'total_time' = 20.00s\n"
"| 'upload_time' = 0.00s\n"
"| 'validation_time' = 10.00s\n"
)
assert result.exit_code == 1, result.output
def test_deposit_delay_critical(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):
scenario = WebScenario()
scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="verified", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
)
scenario.add_step(
"get",
f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="done", status_detail=""),
+ status_template(status="done"),
callback=lambda: time.sleep(60),
)
scenario.install_mock(requests_mock)
result = invoke(
[
"--critical",
"50",
"check-deposit",
*COMMON_OPTIONS,
"single",
"--archive",
sample_archive,
"--metadata",
sample_metadata,
],
catch_exceptions=True,
)
assert result.output == (
"DEPOSIT CRITICAL - Deposit took 80.00s and succeeded.\n"
"| 'load_time' = 70.00s\n"
"| 'total_time' = 80.00s\n"
"| 'upload_time' = 0.00s\n"
"| 'validation_time' = 10.00s\n"
)
assert result.exit_code == 2, result.output
def test_deposit_timeout(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):
scenario = WebScenario()
scenario.add_step(
"post",
f"{BASE_URL}/testcol/",
ENTRY_TEMPLATE.format(status="deposited"),
callback=lambda: time.sleep(1500),
)
scenario.add_step(
"get",
f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="verified", status_detail=""),
+ status_template(status="verified"),
callback=lambda: time.sleep(1500),
)
scenario.add_step(
"get",
f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="loading", status_detail=""),
+ status_template(status="loading"),
callback=lambda: time.sleep(1500),
)
scenario.install_mock(requests_mock)
result = invoke(
[
"check-deposit",
*COMMON_OPTIONS,
"single",
"--archive",
sample_archive,
"--metadata",
sample_metadata,
],
catch_exceptions=True,
)
assert result.output == (
"DEPOSIT CRITICAL - Timed out while in status loading "
"(4520.0s seconds since deposit started)\n"
"| 'total_time' = 4520.00s\n"
"| 'upload_time' = 1500.00s\n"
"| 'validation_time' = 1510.00s\n"
)
assert result.exit_code == 2, result.output
def test_deposit_rejected(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):
scenario = WebScenario()
scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)
scenario.add_step(
"get",
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)
result = invoke(
[
"check-deposit",
*COMMON_OPTIONS,
"single",
"--archive",
sample_archive,
"--metadata",
sample_metadata,
],
catch_exceptions=True,
)
assert result.output == (
"DEPOSIT CRITICAL - Deposit was rejected: booo\n"
"| 'total_time' = 10.00s\n"
"| 'upload_time' = 0.00s\n"
"| 'validation_time' = 10.00s\n"
)
assert result.exit_code == 2, result.output
def test_deposit_failed(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):
scenario = WebScenario()
scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="verified", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="loading", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="loading"),
)
scenario.add_step(
"get",
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)
result = invoke(
[
"check-deposit",
*COMMON_OPTIONS,
"single",
"--archive",
sample_archive,
"--metadata",
sample_metadata,
],
catch_exceptions=True,
)
assert result.output == (
"DEPOSIT CRITICAL - Deposit loading failed: booo\n"
"| 'load_time' = 20.00s\n"
"| 'total_time' = 30.00s\n"
"| 'upload_time' = 0.00s\n"
"| 'validation_time' = 10.00s\n"
)
assert result.exit_code == 2, result.output
def test_deposit_unexpected_status(
requests_mock, mocker, sample_archive, sample_metadata, mocked_time
):
scenario = WebScenario()
scenario.add_step(
"post", f"{BASE_URL}/testcol/", ENTRY_TEMPLATE.format(status="deposited")
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="verified", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="verified"),
)
scenario.add_step(
- "get",
- f"{BASE_URL}/testcol/42/status/",
- STATUS_TEMPLATE.format(status="loading", status_detail=""),
+ "get", f"{BASE_URL}/testcol/42/status/", status_template(status="loading"),
)
scenario.add_step(
"get",
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)
result = invoke(
[
"check-deposit",
*COMMON_OPTIONS,
"single",
"--archive",
sample_archive,
"--metadata",
sample_metadata,
],
catch_exceptions=True,
)
assert result.output == (
"DEPOSIT CRITICAL - Deposit got unexpected status: what (booo)\n"
"| 'load_time' = 20.00s\n"
"| 'total_time' = 30.00s\n"
"| 'upload_time' = 0.00s\n"
"| 'validation_time' = 10.00s\n"
)
assert result.exit_code == 2, result.output