Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/mercurial/tests/test_loader.py
| Show First 20 Lines • Show All 51 Lines • ▼ Show 20 Lines | expected_snapshot = Snapshot( | ||||
| target=hash_to_bytes(tip_revision_default), | target=hash_to_bytes(tip_revision_default), | ||||
| target_type=TargetType.REVISION, | target_type=TargetType.REVISION, | ||||
| ), | ), | ||||
| b"HEAD": SnapshotBranch(target=b"develop", target_type=TargetType.ALIAS,), | b"HEAD": SnapshotBranch(target=b"develop", target_type=TargetType.ALIAS,), | ||||
| }, | }, | ||||
| ) | ) | ||||
| assert_last_visit_matches( | assert_last_visit_matches( | ||||
| swh_storage, | swh_storage, repo_url, status="full", type="hg", snapshot=expected_snapshot.id, | ||||
| repo_url, | |||||
| status="full", | |||||
| type="hg", | |||||
| snapshot=expected_snapshot.id, | |||||
| ) | ) | ||||
| check_snapshot(expected_snapshot, swh_storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
| stats = get_stats(swh_storage) | stats = get_stats(swh_storage) | ||||
| assert stats == { | assert stats == { | ||||
| "content": 2, | "content": 2, | ||||
| "directory": 3, | "directory": 3, | ||||
| "origin": 1, | "origin": 1, | ||||
| ▲ Show 20 Lines • Show All 247 Lines • ▼ Show 20 Lines | def test_visit_repository_with_transplant_operations(swh_storage, datadir, tmp_path): | ||||
| assert len(hg_changesets) > 0 | assert len(hg_changesets) > 0 | ||||
| assert len(transplant_sources) > 0 | assert len(transplant_sources) > 0 | ||||
| assert transplant_sources.issubset(hg_changesets) | assert transplant_sources.issubset(hg_changesets) | ||||
| def test_clone_with_timeout_timeout(caplog, tmp_path, monkeypatch): | def test_clone_with_timeout_timeout(caplog, tmp_path, monkeypatch): | ||||
| log = logging.getLogger("test_clone_with_timeout") | log = logging.getLogger("test_clone_with_timeout") | ||||
| def clone_timeout(source, dest): | def clone_timeout(source, dest, *args, **kwargs): | ||||
| time.sleep(60) | time.sleep(60) | ||||
| monkeypatch.setattr(hglib, "clone", clone_timeout) | monkeypatch.setattr(hglib, "clone", clone_timeout) | ||||
| with pytest.raises(CloneTimeoutError): | with pytest.raises(CloneTimeoutError): | ||||
| HgBundle20Loader.clone_with_timeout( | HgBundle20Loader.clone_with_timeout( | ||||
| log, "https://www.mercurial-scm.org/repo/hello", tmp_path, 1 | log, "https://www.mercurial-scm.org/repo/hello", tmp_path, 1 | ||||
| ) | ) | ||||
| for record in caplog.records: | for record in caplog.records: | ||||
| assert record.levelname == "WARNING" | assert record.levelname == "WARNING" | ||||
| assert "https://www.mercurial-scm.org/repo/hello" in record.getMessage() | assert "https://www.mercurial-scm.org/repo/hello" in record.getMessage() | ||||
| assert record.args == ("https://www.mercurial-scm.org/repo/hello", 1) | assert record.args == ("https://www.mercurial-scm.org/repo/hello", 1) | ||||
| def test_clone_with_timeout_returns(caplog, tmp_path, monkeypatch): | def test_clone_with_timeout_returns(caplog, tmp_path, monkeypatch): | ||||
| log = logging.getLogger("test_clone_with_timeout") | log = logging.getLogger("test_clone_with_timeout") | ||||
| def clone_return(source, dest): | def clone_return(source, dest, *args, **kwargs): | ||||
| return (source, dest) | return (source, dest) | ||||
| monkeypatch.setattr(hglib, "clone", clone_return) | monkeypatch.setattr(hglib, "clone", clone_return) | ||||
| assert HgBundle20Loader.clone_with_timeout( | assert HgBundle20Loader.clone_with_timeout( | ||||
| log, "https://www.mercurial-scm.org/repo/hello", tmp_path, 1 | log, "https://www.mercurial-scm.org/repo/hello", tmp_path, 1 | ||||
| ) == ("https://www.mercurial-scm.org/repo/hello", tmp_path) | ) == ("https://www.mercurial-scm.org/repo/hello", tmp_path) | ||||
| def test_clone_with_timeout_exception(caplog, tmp_path, monkeypatch): | def test_clone_with_timeout_exception(caplog, tmp_path, monkeypatch): | ||||
| log = logging.getLogger("test_clone_with_timeout") | log = logging.getLogger("test_clone_with_timeout") | ||||
| def clone_return(source, dest): | def clone_return(source, dest, *args, **kwargs): | ||||
| raise ValueError("Test exception") | raise ValueError("Test exception") | ||||
| monkeypatch.setattr(hglib, "clone", clone_return) | monkeypatch.setattr(hglib, "clone", clone_return) | ||||
| with pytest.raises(ValueError) as excinfo: | with pytest.raises(ValueError) as excinfo: | ||||
| HgBundle20Loader.clone_with_timeout( | HgBundle20Loader.clone_with_timeout( | ||||
| log, "https://www.mercurial-scm.org/repo/hello", tmp_path, 1 | log, "https://www.mercurial-scm.org/repo/hello", tmp_path, 1 | ||||
| ) | ) | ||||
| assert "Test exception" in excinfo.value.args[0] | assert "Test exception" in excinfo.value.args[0] | ||||