Differential D8808 Diff 31739 swh/loader/git/tests/test_loader.py

Changeset View

Standalone View

swh/loader/git/tests/test_loader.py

Show All 20 Lines
from swh.loader.git import converters, dumb		from swh.loader.git import converters, dumb
from swh.loader.git.loader import GitLoader		from swh.loader.git.loader import GitLoader
from swh.loader.git.tests.test_from_disk import SNAPSHOT1, FullGitLoaderTests		from swh.loader.git.tests.test_from_disk import SNAPSHOT1, FullGitLoaderTests
from swh.loader.tests import (		from swh.loader.tests import (
assert_last_visit_matches,		assert_last_visit_matches,
get_stats,		get_stats,
prepare_repository_from_archive,		prepare_repository_from_archive,
)		)
from swh.model.model import Origin, OriginVisit, OriginVisitStatus, Snapshot		from swh.model.model import (
		Origin,
		OriginVisit,
		OriginVisitStatus,
		Snapshot,
		SnapshotBranch,
		TargetType,
		)


class CommonGitLoaderNotFound:		class CommonGitLoaderNotFound:
@pytest.fixture(autouse=True)		@pytest.fixture(autouse=True)
def __inject_fixtures(self, mocker):		def __inject_fixtures(self, mocker):
"""Inject required fixtures in unittest.TestCase class"""		"""Inject required fixtures in unittest.TestCase class"""
self.mocker = mocker		self.mocker = mocker

▲ Show 20 Lines • Show All 514 Lines • ▼ Show 20 Lines	):
"has_parent_origins": True,		"has_parent_origins": True,
}		}
assert [c for c in statsd_report.mock_calls if c[1][0].startswith("git_")] == [		assert [c for c in statsd_report.mock_calls if c[1][0].startswith("git_")] == [
call("git_total", "c", 1, {}, 1),		call("git_total", "c", 1, {}, 1),
call("git_ignored_refs_percent", "h", 0.0, {}, 1),		call("git_ignored_refs_percent", "h", 0.0, {}, 1),
call("git_known_refs_percent", "h", expected_git_known_refs_percent, {}, 1),		call("git_known_refs_percent", "h", expected_git_known_refs_percent, {}, 1),
]		]

		def test_load_incremental_negotiation(self):
		"""Check that the packfile negotiated when running an incremental load only
		contains the "new" commits, and not all objects."""

		snapshot_id = b"\x01" * 20
		now = datetime.datetime.now(tz=datetime.timezone.utc)

		def ovgl(origin_url, allowed_statuses, require_snapshot, type):
		if origin_url == f"base://{self.repo_url}":
		return OriginVisit(origin=origin_url, visit=42, date=now, type="git")
		else:
		return None

		self.loader.storage.origin_visit_get_latest.side_effect = ovgl
		self.loader.storage.origin_visit_status_get_latest.return_value = (
		OriginVisitStatus(
		origin=f"base://{self.repo_url}",
		visit=42,
		snapshot=snapshot_id,
		date=now,
		status="full",
		)
		)
		self.loader.storage.snapshot_get_branches.return_value = {
		"id": snapshot_id,
		"branches": {
		b"refs/heads/master": SnapshotBranch(
		# id of the initial commit in the git repository fixture
		target=b"\xb6\xf4\x02\x92\xc4\xe9J\x8f~{J\xffP\xe6\xc7B\x9a\xb9\x8e*",
		target_type=TargetType.REVISION,
		),
		},
		"next_branch": None,
		}

		res = self.loader.load()
		assert res == {"status": "eventful"}

		stats = get_stats(self.loader.storage)
		assert stats == {
		"content": 3, # instead of 4 for the full repository
		"directory": 6, # instead of 7
		"origin": 1,
		"origin_visit": 1,
		"release": 0,
		"revision": 6, # instead of 7
		"skipped_content": 0,
		"snapshot": 1,
		}


class DumbGitLoaderTestBase(FullGitLoaderTests):		class DumbGitLoaderTestBase(FullGitLoaderTests):
"""Prepare a git repository to be loaded using the HTTP dumb transfer protocol."""		"""Prepare a git repository to be loaded using the HTTP dumb transfer protocol."""

@pytest.fixture(autouse=True)		@pytest.fixture(autouse=True)
def init(self, swh_storage, datadir, tmp_path):		def init(self, swh_storage, datadir, tmp_path):
# remove any proxy settings in order to successfully spawn a local HTTP server		# remove any proxy settings in order to successfully spawn a local HTTP server
http_proxy = os.environ.get("http_proxy")		http_proxy = os.environ.get("http_proxy")
▲ Show 20 Lines • Show All 189 Lines • Show Last 20 Lines