Differential D8787 Diff 31702 swh/loader/svn/tests/test_utils.py

Changeset View

Standalone View

swh/loader/svn/tests/test_utils.py

# Copyright (C) 2016-2022 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 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 logging		import logging
import os		import os
from pathlib import Path		from pathlib import Path
import pty		import pty
		import re
import shutil		import shutil
from subprocess import Popen		from subprocess import Popen, run

import pytest		import pytest

from swh.loader.svn import utils		from swh.loader.svn import utils
		from swh.loader.tests import prepare_repository_from_archive


def test_outputstream():		def test_outputstream():
stdout_r, stdout_w = pty.openpty()		stdout_r, stdout_w = pty.openpty()
echo = Popen(["echo", "-e", "foo\nbar\nbaz"], stdout=stdout_w)		echo = Popen(["echo", "-e", "foo\nbar\nbaz"], stdout=stdout_w)
os.close(stdout_w)		os.close(stdout_w)
stdout_stream = utils.OutputStream(stdout_r)		stdout_stream = utils.OutputStream(stdout_r)
lines = []		lines = []
▲ Show 20 Lines • Show All 77 Lines • ▼ Show 20 Lines	):
assert os.path.exists(repo_path), "Repository should exists"		assert os.path.exists(repo_path), "Repository should exists"
assert os.path.exists(dump_ori_path), "Original dump path should still exists"		assert os.path.exists(dump_ori_path), "Original dump path should still exists"

assert len(caplog.record_tuples) == 1		assert len(caplog.record_tuples) == 1
assert "Failure to remove" in caplog.record_tuples[0][2]		assert "Failure to remove" in caplog.record_tuples[0][2]
assert mock_remove.called		assert mock_remove.called


		def test_init_svn_repo_from_truncated_dump(datadir, tmp_path):
		"""Mounting partial svn repository from a truncated dump should work"""

		# prepare a repository
		archive_name = "pkg-gourmet"
		archive_path = os.path.join(datadir, f"{archive_name}.tgz")
		repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)

		# dump it to file
		dump_path = str(tmp_path / f"{archive_name}.dump")
		truncated_dump_path = str(tmp_path / f"{archive_name}_truncated.dump")
		svnrdump_cmd = ["svnrdump", "dump", repo_url]
		with open(dump_path, "wb") as dump:
		run(svnrdump_cmd, stdout=dump)

		# create a truncated dump file that will generate a "svnadmin load" error
		with open(dump_path, "rb") as dump, open(
		truncated_dump_path, "wb"
		) as truncated_dump:
		dump_lines = dump.readlines()
		vlorentzUnsubmitted Not Done Inline Actions `assert dump_lines > 150` to make sure the test is useful vlorentz: `assert dump_lines > 150` to make sure the test is useful
		assert len(dump_lines) > 150
		truncated_dump_content = b"".join(dump_lines[:150])
		truncated_dump.write(truncated_dump_content)

		# compute max revision number with non truncated data
		revs = re.findall(rb"Revision-number: ([0-9]+)", truncated_dump_content)
		max_rev = int(revs[-1]) - 1

		# prepare repository from truncated dump
		_, repo_path = utils.init_svn_repo_from_dump(
		truncated_dump_path, gzip=False, root_dir=tmp_path, max_rev=max_rev
		)

		# check expected number of revisions have been loaded
		svnadmin_info = run(["svnadmin", "info", repo_path], capture_output=True, text=True)
		assert f"Revisions: {max_rev}\n" in svnadmin_info.stdout


def test_init_svn_repo_from_archive_dump(datadir, tmp_path):		def test_init_svn_repo_from_archive_dump(datadir, tmp_path):
"""Mounting svn repository out of an archive dump is ok"""		"""Mounting svn repository out of an archive dump is ok"""
dump_name = "penguinsdbtools2018.dump.gz"		dump_name = "penguinsdbtools2018.dump.gz"
dump_path = os.path.join(datadir, dump_name)		dump_path = os.path.join(datadir, dump_name)

tmp_repo, repo_path = utils.init_svn_repo_from_archive_dump(		tmp_repo, repo_path = utils.init_svn_repo_from_archive_dump(
dump_path, cleanup_dump=False, root_dir=tmp_path		dump_path, cleanup_dump=False, root_dir=tmp_path
)		)
▲ Show 20 Lines • Show All 325 Lines • Show Last 20 Lines