diff --git a/swh/icinga_plugins/tests/test_deposit.py b/swh/icinga_plugins/tests/test_deposit.py
index eddfabd..7df3095 100644
--- a/swh/icinga_plugins/tests/test_deposit.py
+++ b/swh/icinga_plugins/tests/test_deposit.py
@@ -1,375 +1,375 @@
# Copyright (C) 2019 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 click.testing import CliRunner
import pytest
from swh.icinga_plugins.cli import cli
from .web_scenario import WebScenario
-BASE_URL = 'mock://swh-deposit.example.org/1'
+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}
'''
@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(cli, 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', 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', BASE_URL + '/testcol/',
ENTRY_TEMPLATE.format(status='deposited'))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='verified', status_detail=''))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='loading', status_detail=''))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='done', status_detail=''))
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', BASE_URL + '/testcol/',
ENTRY_TEMPLATE.format(status='deposited'))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='verified', status_detail=''))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='done', status_detail=''))
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', BASE_URL + '/testcol/',
ENTRY_TEMPLATE.format(status='deposited'))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='verified', status_detail=''))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='done', status_detail=''),
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', BASE_URL + '/testcol/',
ENTRY_TEMPLATE.format(status='deposited'),
callback=lambda: time.sleep(1500))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='verified', status_detail=''),
callback=lambda: time.sleep(1500))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='loading', status_detail=''),
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', BASE_URL + '/testcol/',
ENTRY_TEMPLATE.format(status='deposited'))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(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', BASE_URL + '/testcol/',
ENTRY_TEMPLATE.format(status='deposited'))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='verified', status_detail=''))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='loading', status_detail=''))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(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', BASE_URL + '/testcol/',
ENTRY_TEMPLATE.format(status='deposited'))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='verified', status_detail=''))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(status='loading', status_detail=''))
scenario.add_step(
'get', BASE_URL + '/testcol/42/status/',
STATUS_TEMPLATE.format(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