Differential D5858 Diff 21020 swh/web/tests/common/test_django_command.py

Changeset View

Standalone View

swh/web/tests/common/test_django_command.py

	# Copyright (C) 2021 The Software Heritage developers			# Copyright (C) 2021 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 Affero General Public License version 3, or any later version			# License: GNU Affero 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

				from datetime import datetime, timedelta, timezone
	from io import StringIO			from io import StringIO

	import pytest			import pytest

	from django.core.management import call_command			from django.core.management import call_command

				from swh.core.api.classes import stream_results
				from swh.web.common.models import (
				SAVE_REQUEST_ACCEPTED,
				SAVE_TASK_FAILED,
				SAVE_TASK_SCHEDULED,
				SAVE_TASK_SUCCEEDED,
				VISIT_STATUS_FAILED,
				VISIT_STATUS_FULL,
				VISIT_STATUS_PARTIAL,
				)
				from swh.web.common.typing import SaveOriginRequestInfo

	@pytest.mark.parametrize("nb_results", [0, 10, 20])			MODULE_FQDN = "swh.web.common.management.commands"
	def test_command_refresh__with_statuses_refreshed(mocker, nb_results):			COMMAND_NAME = "refresh_savecodenow_statuses"
	"""Refresh status command reported updated non-terminal statuses.

	"""			AUTHORIZED_ORIGIN_URL = "https://scm.ourproject.org/anonscm/%s"
	command_name = "refresh_savecodenow_statuses"
	module_fqdn = "swh.web.common.management.commands"
				@pytest.fixture
				def mock_scheduler_and_refresh(mocker, swh_scheduler):
	mock_refresh = mocker.patch(			mock_refresh = mocker.patch(
	f"{module_fqdn}.{command_name}.refresh_save_origin_request_statuses"			f"{MODULE_FQDN}.{COMMAND_NAME}.refresh_save_origin_request_statuses"
	)			)
	# fake returned refreshed status			mock_scheduler = mocker.patch(f"{MODULE_FQDN}.{COMMAND_NAME}.get_scheduler")
	mock_refresh.return_value = [{"": ""}] * nb_results			mock_scheduler.return_value = swh_scheduler

				return mock_scheduler, mock_refresh
				anlambertUnsubmitted Done Inline Actions I would have split that fixture in two separate ones but not a big deal. anlambert: I would have split that fixture in two separate ones but not a big deal.
				ardumontAuthorUnsubmitted Done Inline Actions I agree. I did not spend much time in trying to refactor the tests after making it green ;) ardumont: I agree. I did not spend much time in trying to refactor the tests after making it green ;)


				@pytest.mark.parametrize("nb_results", [0, 10, 20])
				def test_command_refresh__with_statuses_refreshed(
				mock_scheduler_and_refresh, nb_results
				):
				"""Refresh status command reports non-terminal statuses updates.

				"""
				mock_scheduler, mock_refresh = mock_scheduler_and_refresh
				# fake returned refreshed status for 'archives' visit type
				mock_refresh.return_value = [{"visit_type": "archives",}] * nb_results

	out = StringIO()			out = StringIO()
	call_command(command_name, stdout=out)			call_command(COMMAND_NAME, stdout=out)

	assert mock_refresh.called			assert mock_refresh.called

	actual_output = out.getvalue()			actual_output = out.getvalue()
	if nb_results > 0:			if nb_results > 0:
	assert f"updated {nb_results}" in actual_output			assert f"updated {nb_results}" in actual_output
	else:			else:
	assert "Nothing" in actual_output			assert "Nothing" in actual_output


				@pytest.fixture
				def fake_refreshed_data():
				"""Prepare test data within the scheduler and the swh-web model db

				"""
				entries = [
				{
				"visit_type": "archives", # ignored from recurring task scheduling
				"visit_status": VISIT_STATUS_FULL,
				"task_status": SAVE_TASK_SUCCEEDED,
				},
				{
				"visit_type": "git", # scheduled as recurring task
				"visit_status": VISIT_STATUS_FULL,
				"task_status": SAVE_TASK_SUCCEEDED,
				},
				{
				"visit_type": "svn", # scheduled as recurring task
				"visit_status": VISIT_STATUS_FULL,
				"task_status": SAVE_TASK_SUCCEEDED,
				},
				{
				"visit_type": "hg", # scheduled as recurring task
				"visit_status": VISIT_STATUS_PARTIAL,
				"task_status": SAVE_TASK_FAILED,
				},
				{
				"visit_type": "svn", # ignored from recurring task scheduling
				"visit_status": VISIT_STATUS_FAILED,
				"task_status": SAVE_TASK_FAILED,
				},
				{
				"visit_type": "hg",
				"visit_status": "created", # ignored from recurring task scheduling
				"task_status": SAVE_TASK_SCHEDULED,
				},
				]
				time_now = datetime.now(tz=timezone.utc) - timedelta(days=len(entries))
				return [
				SaveOriginRequestInfo(
				visit_type=meta["visit_type"],
				visit_status=meta["visit_status"],
				origin_url=AUTHORIZED_ORIGIN_URL % i,
				save_request_date=time_now + timedelta(days=i - 1),
				save_request_status=SAVE_REQUEST_ACCEPTED,
				visit_date=time_now + timedelta(days=i),
				save_task_status=meta["task_status"],
				id=i,
				loading_task_id=i,
				)
				for i, meta in enumerate(entries)
				]


				def test_command_refresh__with_recurrent_tasks_scheduling(
				mock_scheduler_and_refresh, fake_refreshed_data, swh_scheduler
				):
				"""Refresh status command report updates of statuses. The successful ones without the
				type 'archived' are also scheduled recurringly.

				"""
				mock_scheduler, mock_refresh = mock_scheduler_and_refresh
				mock_refresh.return_value = fake_refreshed_data

				# only visit types (git, hg, svn) types with status (full, partial) are taken into
				# account for scheduling, so only 3 of those matches in the fake data set.
				expected_nb_scheduled = 0
				for entry in fake_refreshed_data:
				if entry["visit_type"] == "archives": # only deal with git, svn, hg
				continue
				if entry["visit_status"] not in ("partial", "full"):
				continue
				expected_nb_scheduled += 1

				assert expected_nb_scheduled == 3

				out = StringIO()
				call_command(COMMAND_NAME, stdout=out)

				actual_output = out.getvalue()
				assert f"Successfully updated {len(fake_refreshed_data)}" in actual_output

				lister = swh_scheduler.get_or_create_lister(
				name="save-code-now", instance_name="host"
				)

				result = list(stream_results(swh_scheduler.get_listed_origins, lister.id))
				assert len(result) == expected_nb_scheduled

				assert mock_scheduler.called