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