diff --git a/requirements-swh.txt b/requirements-swh.txt
index 4abaae7..8ec47d2 100644
--- a/requirements-swh.txt
+++ b/requirements-swh.txt
@@ -1,3 +1,3 @@
-swh.core[http] # required by swh.storage
+swh.core[http] >= 0.3
swh.deposit
swh.storage >= v0.0.162
diff --git a/setup.py b/setup.py
index 062cfed..7d0e2bd 100755
--- a/setup.py
+++ b/setup.py
@@ -1,78 +1,78 @@
#!/usr/bin/env python3
# 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
from io import open
from os import path, walk
from setuptools import find_packages, setup
here = path.abspath(path.dirname(__file__))
# Get the long description from the README file
with open(path.join(here, "README.md"), encoding="utf-8") as f:
long_description = f.read()
def parse_requirements(name=None):
if name:
reqf = "requirements-%s.txt" % name
else:
reqf = "requirements.txt"
requirements = []
if not path.exists(reqf):
return requirements
with open(reqf) as f:
for line in f.readlines():
line = line.strip()
if not line or line.startswith("#"):
continue
requirements.append(line)
return requirements
# package generated static assets as module data files
data_files = []
for root, _, files in walk("data/"):
root_files = [path.join(root, i) for i in files]
data_files.append((path.join("share/swh/icinga-plugins", root), root_files))
setup(
name="swh.icinga_plugins",
description="Icinga plugins for Software Heritage infrastructure " "monitoring",
long_description=long_description,
long_description_content_type="text/markdown",
python_requires=">=3.7",
author="Software Heritage developers",
author_email="swh-devel@inria.fr",
url="https://forge.softwareheritage.org/diffusion/swh-icinga-plugins",
packages=find_packages(), # packages's modules
install_requires=parse_requirements() + parse_requirements("swh"),
tests_require=parse_requirements("test"),
setup_requires=["setuptools-scm"],
use_scm_version=True,
extras_require={"testing": parse_requirements("test")},
include_package_data=True,
entry_points="""
[swh.cli.subcommands]
- icinga_plugins=swh.icinga_plugins.cli:cli
+ icinga_plugins=swh.icinga_plugins.cli
""",
classifiers=[
"Programming Language :: Python :: 3",
"Intended Audience :: Developers",
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
"Operating System :: OS Independent",
"Development Status :: 3 - Alpha",
],
project_urls={
"Bug Reports": "https://forge.softwareheritage.org/maniphest",
"Funding": "https://www.softwareheritage.org/donate",
"Source": "https://forge.softwareheritage.org/source/swh-icinga-plugins",
},
data_files=data_files,
)
diff --git a/swh/icinga_plugins/cli.py b/swh/icinga_plugins/cli.py
index b37adc6..26ce1fa 100644
--- a/swh/icinga_plugins/cli.py
+++ b/swh/icinga_plugins/cli.py
@@ -1,103 +1,103 @@
# 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
# WARNING: do not import unnecessary things here to keep cli startup time under
# control
import sys
import click
-from swh.core.cli import CONTEXT_SETTINGS
+from swh.core.cli import CONTEXT_SETTINGS, swh as swh_cli_group
-@click.group(name="icinga_plugins", context_settings=CONTEXT_SETTINGS)
+@swh_cli_group.group(name="icinga_plugins", context_settings=CONTEXT_SETTINGS)
@click.option("-w", "--warning", type=int, help="Warning threshold.")
@click.option("-c", "--critical", type=int, help="Critical threshold.")
@click.pass_context
-def cli(ctx, warning, critical):
+def icinga_cli_group(ctx, warning, critical):
"""Main command for Icinga plugins
"""
ctx.ensure_object(dict)
if warning:
ctx.obj["warning_threshold"] = int(warning)
if critical:
ctx.obj["critical_threshold"] = int(critical)
-@cli.group(name="check-vault")
+@icinga_cli_group.group(name="check-vault")
@click.option(
"--swh-storage-url", type=str, required=True, help="URL to an swh-storage HTTP API"
)
@click.option(
"--swh-web-url", type=str, required=True, help="URL to an swh-web instance"
)
@click.option(
"--poll-interval",
type=int,
default=10,
help="Interval (in seconds) between two polls to the API, "
"to check for cooking status.",
)
@click.pass_context
def check_vault(ctx, **kwargs):
ctx.obj.update(kwargs)
@check_vault.command(name="directory")
@click.pass_context
def check_vault_directory(ctx):
"""Picks a random directory, requests its cooking via swh-web,
and waits for completion."""
from .vault import VaultCheck
sys.exit(VaultCheck(ctx.obj).main())
-@cli.group(name="check-deposit")
+@icinga_cli_group.group(name="check-deposit")
@click.option(
"--server",
type=str,
default="https://deposit.softwareheritage.org/1",
help="URL to the SWORD server to test",
)
@click.option("--username", type=str, required=True, help="Login for the SWORD server")
@click.option(
"--password", type=str, required=True, help="Password for the SWORD server"
)
@click.option(
"--collection",
type=str,
required=True,
help="Software collection to use on the SWORD server",
)
@click.option(
"--poll-interval",
type=int,
default=10,
help="Interval (in seconds) between two polls to the API, "
"to check for ingestion status.",
)
@click.pass_context
def check_deposit(ctx, **kwargs):
ctx.obj.update(kwargs)
@check_deposit.command(name="single")
@click.option(
"--archive", type=click.Path(), required=True, help="Software artefact to upload"
)
@click.option(
"--metadata",
type=click.Path(),
required=True,
help="Metadata file for the software artefact.",
)
@click.pass_context
def check_deposit_single(ctx, **kwargs):
"""Checks the provided archive and metadata file and be deposited."""
from .deposit import DepositCheck
ctx.obj.update(kwargs)
sys.exit(DepositCheck(ctx.obj).main())
diff --git a/swh/icinga_plugins/tests/test_deposit.py b/swh/icinga_plugins/tests/test_deposit.py
index 9aae6d8..de4327c 100644
--- a/swh/icinga_plugins/tests/test_deposit.py
+++ b/swh/icinga_plugins/tests/test_deposit.py
@@ -1,467 +1,467 @@
# 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 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}
"""
@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)
+ 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", 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
diff --git a/swh/icinga_plugins/tests/test_vault.py b/swh/icinga_plugins/tests/test_vault.py
index be5bb57..787612a 100644
--- a/swh/icinga_plugins/tests/test_vault.py
+++ b/swh/icinga_plugins/tests/test_vault.py
@@ -1,291 +1,291 @@
# 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 time
from click.testing import CliRunner
-from swh.icinga_plugins.cli import cli
+from swh.icinga_plugins.cli import icinga_cli_group
from .web_scenario import WebScenario
dir_id = "ab" * 20
response_pending = {
"obj_id": dir_id,
"obj_type": "directory",
"progress_message": "foo",
"status": "pending",
}
response_done = {
"fetch_url": f"/api/1/vault/directory/{dir_id}/raw/",
"id": 9,
"obj_id": dir_id,
"obj_type": "directory",
"status": "done",
}
response_failed = {
"obj_id": dir_id,
"obj_type": "directory",
"progress_message": "foobar",
"status": "failed",
}
response_unknown_status = {
"obj_id": dir_id,
"obj_type": "directory",
"progress_message": "what",
"status": "boo",
}
class FakeStorage:
def __init__(self, foo, **kwargs):
pass
def directory_get_random(self):
return bytes.fromhex(dir_id)
def invoke(args, catch_exceptions=False):
runner = CliRunner()
- result = runner.invoke(cli, args)
+ result = runner.invoke(icinga_cli_group, args)
if not catch_exceptions and result.exception:
print(result.output)
raise result.exception
return result
def test_vault_immediate_success(requests_mock, mocker, mocked_time):
scenario = WebScenario()
url = f"mock://swh-web.example.org/api/1/vault/directory/{dir_id}/"
scenario.add_step("get", url, {}, status_code=404)
scenario.add_step("post", url, response_pending)
scenario.add_step("get", url, response_done)
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",
]
)
assert result.output == (
f"VAULT OK - cooking directory {dir_id} took "
f"10.00s and succeeded.\n"
f"| 'total_time' = 10.00s\n"
)
assert result.exit_code == 0, result.output
def test_vault_delayed_success(requests_mock, mocker, mocked_time):
scenario = WebScenario()
url = f"mock://swh-web.example.org/api/1/vault/directory/{dir_id}/"
scenario.add_step("get", url, {}, status_code=404)
scenario.add_step("post", url, response_pending)
scenario.add_step("get", url, response_pending)
scenario.add_step("get", url, response_done)
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",
]
)
assert result.output == (
f"VAULT OK - cooking directory {dir_id} took "
f"20.00s and succeeded.\n"
f"| 'total_time' = 20.00s\n"
)
assert result.exit_code == 0, result.output
def test_vault_failure(requests_mock, mocker, mocked_time):
scenario = WebScenario()
url = f"mock://swh-web.example.org/api/1/vault/directory/{dir_id}/"
scenario.add_step("get", url, {}, status_code=404)
scenario.add_step("post", url, response_pending)
scenario.add_step("get", url, response_failed)
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 == (
f"VAULT CRITICAL - cooking directory {dir_id} took "
f"10.00s and failed with: foobar\n"
f"| 'total_time' = 10.00s\n"
)
assert result.exit_code == 2, result.output
def test_vault_unknown_status(requests_mock, mocker, mocked_time):
scenario = WebScenario()
url = f"mock://swh-web.example.org/api/1/vault/directory/{dir_id}/"
scenario.add_step("get", url, {}, status_code=404)
scenario.add_step("post", url, response_pending)
scenario.add_step("get", url, response_unknown_status)
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 == (
f"VAULT CRITICAL - cooking directory {dir_id} took "
f"10.00s and resulted in unknown status: boo\n"
f"| 'total_time' = 10.00s\n"
)
assert result.exit_code == 2, result.output
def test_vault_timeout(requests_mock, mocker, mocked_time):
scenario = WebScenario()
url = f"mock://swh-web.example.org/api/1/vault/directory/{dir_id}/"
scenario.add_step("get", url, {}, status_code=404)
scenario.add_step("post", url, response_pending)
scenario.add_step("get", url, response_pending)
scenario.add_step("get", url, response_pending, callback=lambda: time.sleep(4000))
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 == (
f"VAULT CRITICAL - cooking directory {dir_id} took more than "
f"4020.00s and has status: foo\n"
f"| 'total_time' = 4020.00s\n"
)
assert result.exit_code == 2, result.output
def test_vault_cached_directory(requests_mock, mocker, mocked_time):
"""First serves a directory that's already in the cache, to
test that vault_check requests another one."""
scenario = WebScenario()
url = f"mock://swh-web.example.org/api/1/vault/directory/{dir_id}/"
scenario.add_step("get", url, {}, status_code=200)
scenario.add_step("get", url, {}, status_code=404)
scenario.add_step("post", url, response_pending)
scenario.add_step("get", url, response_done)
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",
]
)
assert result.output == (
f"VAULT OK - cooking directory {dir_id} took "
f"10.00s and succeeded.\n"
f"| 'total_time' = 10.00s\n"
)
assert result.exit_code == 0, result.output
def test_vault_no_directory(requests_mock, mocker, mocked_time):
"""Tests with an empty storage"""
scenario = WebScenario()
scenario.install_mock(requests_mock)
get_storage_mock = mocker.patch("swh.icinga_plugins.vault.get_storage")
get_storage_mock.side_effect = FakeStorage
mocker.patch(f"{__name__}.FakeStorage.directory_get_random", return_value=None)
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 - No directory exists in the archive.\n")
assert result.exit_code == 2, result.output