Changeset View
Changeset View
Standalone 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=bytes.fromhex("b6f40292c4e94a8f7e7b4aff50e6c7429ab98e2a"), | |||||
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 |