Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/cli/test_admin.py
# Copyright (C) 2019-2020 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 pytest | import pytest | ||||
from swh.deposit.cli.admin import admin as cli | from swh.deposit.cli.admin import admin as cli | ||||
from swh.deposit.config import DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_PARTIAL | |||||
from swh.deposit.models import DepositClient, DepositCollection | from swh.deposit.models import DepositClient, DepositCollection | ||||
@pytest.fixture(autouse=True) | @pytest.fixture(autouse=True) | ||||
def enable_db_access_for_all_tests(db): | def enable_db_access_for_all_tests(db): | ||||
pass | pass | ||||
▲ Show 20 Lines • Show All 165 Lines • ▼ Show 20 Lines | """ | ||||
assert ( | assert ( | ||||
result2.output | result2.output | ||||
== f"""Collection '{collection_name}' exists, skipping. | == f"""Collection '{collection_name}' exists, skipping. | ||||
Update user '{user_name}'. | Update user '{user_name}'. | ||||
User '{user_name}' updated. | User '{user_name}' updated. | ||||
""" | """ | ||||
) | ) | ||||
def test_cli_admin_reschedule_unknown_deposit(cli_runner): | |||||
"""Rescheduling unknown deposit should report failure | |||||
""" | |||||
unknown_deposit_id = 666 | |||||
from swh.deposit.models import Deposit | |||||
try: | |||||
Deposit.objects.get(id=unknown_deposit_id) | |||||
except Deposit.DoesNotExist: | |||||
pass | |||||
result = cli_runner.invoke( | |||||
cli, ["deposit", "reschedule", "--deposit-id", unknown_deposit_id] | |||||
) | |||||
assert result.output == f"Deposit {unknown_deposit_id} does not exist.\n" | |||||
assert result.exit_code == 1 | |||||
def test_cli_admin_reschedule_verified_deposit(cli_runner, complete_deposit): | |||||
"""Rescheduling verified deposit should do nothing but report | |||||
""" | |||||
deposit = complete_deposit | |||||
deposit.status = "verified" | |||||
deposit.save() | |||||
result = cli_runner.invoke( | |||||
cli, ["deposit", "reschedule", "--deposit-id", deposit.id] | |||||
) | |||||
assert result.output == f"Deposit {deposit.id} already set for rescheduling.\n" | |||||
assert result.exit_code == 0 | |||||
@pytest.mark.parametrize( | |||||
"status_to_check", [DEPOSIT_STATUS_PARTIAL, DEPOSIT_STATUS_DEPOSITED] | |||||
) | |||||
def test_cli_admin_reschedule_unaccepted_deposit_status( | |||||
status_to_check, cli_runner, complete_deposit | |||||
): | |||||
"""Rescheduling verified deposit should do nothing but report | |||||
""" | |||||
deposit = complete_deposit | |||||
deposit.status = status_to_check # not accepted status will fail the check | |||||
deposit.save() | |||||
result = cli_runner.invoke( | |||||
cli, ["deposit", "reschedule", "--deposit-id", deposit.id] | |||||
) | |||||
assert result.output == ( | |||||
f"Deposit {deposit.id} cannot be rescheduled (status: {deposit.status}).\n" | |||||
"Rescheduling deposit is only accepted for deposit with status: done, failed.\n" | |||||
) | |||||
assert result.exit_code == 1 | |||||
def test_cli_admin_reschedule_missing_task_id(cli_runner, complete_deposit): | |||||
"""Rescheduling deposit with no load_task_id cannot work. | |||||
""" | |||||
deposit = complete_deposit | |||||
deposit.load_task_id = "" # drop the load-task-id so it fails the check | |||||
deposit.save() | |||||
result = cli_runner.invoke( | |||||
cli, ["deposit", "reschedule", "--deposit-id", deposit.id] | |||||
) | |||||
assert result.output == ( | |||||
f"Deposit {deposit.id} cannot be rescheduled. It misses the " | |||||
"associated scheduler task id (field load_task_id).\n" | |||||
) | |||||
assert result.exit_code == 1 | |||||
# def test_cli_admin_reschedule_nominal(cli_runner, complete_deposit, swh_scheduler): | |||||
# """Rescheduling deposit with no load_task_id cannot work. | |||||
# """ | |||||
# from swh.deposit.models import Deposit | |||||
# deposit = complete_deposit | |||||
# import pdb; pdb.set_trace() | |||||
# deposit.load_task_id = 10 # drop the load-task-id so it fails the check | |||||
# deposit.save() | |||||
# result = cli_runner.invoke( | |||||
# cli, ["deposit", "reschedule", "--deposit-id", deposit.id] | |||||
# ) | |||||
# assert result.exit_code == 0 | |||||
# # Ensure the | |||||
# deposit = Deposit.objects.get(id=deposit.id) | |||||
# # got reset to a state which allows rescheduling | |||||
# assert deposit.id | |||||
# assert deposit.swhid is None | |||||
# assert deposit.swhid_context is None | |||||
# assert deposit.status == DEPOSIT_STATUS_VERIFIED |