Differential D8004 Diff 28852 swh/loader/cvs/tests/test_loader.py

Changeset View

Standalone View

swh/loader/cvs/tests/test_loader.py

# Copyright (C) 2016-2021 The Software Heritage developers		# Copyright (C) 2016-2022 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

import os		import os
import subprocess		import subprocess
import tempfile		import tempfile
from typing import Any, Dict		from typing import Any, Dict
		from urllib.parse import urlparse

import pytest		import pytest

		from swh.loader.cvs.cvsclient import CVSClient
from swh.loader.cvs.loader import BadPathException, CvsLoader		from swh.loader.cvs.loader import BadPathException, CvsLoader
from swh.loader.tests import (		from swh.loader.tests import (
assert_last_visit_matches,		assert_last_visit_matches,
check_snapshot,		check_snapshot,
get_stats,		get_stats,
prepare_repository_from_archive,		prepare_repository_from_archive,
)		)
from swh.model.hashutil import hash_to_bytes		from swh.model.hashutil import hash_to_bytes
▲ Show 20 Lines • Show All 1,210 Lines • ▼ Show 20 Lines	mock_subprocess.run.side_effect = [
),		),
subprocess.CompletedProcess(args=rsync_third_call, returncode=0),		subprocess.CompletedProcess(args=rsync_third_call, returncode=0),
]		]

loader = CvsLoader(swh_storage, repo_url)		loader = CvsLoader(swh_storage, repo_url)
loader.cvs_module_name = module_name		loader.cvs_module_name = module_name
loader.cvsroot_path = tmp_path		loader.cvsroot_path = tmp_path
loader.fetch_cvs_repo_with_rsync(host, path)		loader.fetch_cvs_repo_with_rsync(host, path)


		@pytest.mark.parametrize(
		"pserver_url",
		[
		"pserver://anonymous:anonymous@cvs.example.org/cvsroot/project/module",
		"pserver://anonymous@cvs.example.org/cvsroot/project/module",
		],
		)
		def test_cvs_client_connect_pserver(mocker, pserver_url):
		from swh.loader.cvs.cvsclient import socket

		conn = mocker.MagicMock()
		conn.recv.side_effect = [b"I LOVE YOU\n", b"Valid-requests \n", b"ok\n"]
		mocker.patch.object(socket, "create_connection").return_value = conn
		parsed_url = urlparse(pserver_url)

		# check cvs client can be instantiated without errors
		CVSClient(parsed_url)