libvcs/tests/sync/test_hg.py at 95dda539a0b0ae2a8d3a4a514fa57d987491f98c · vcs-python/libvcs · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
"""Tests for libvcs hg repos."""

from __future__ import annotations

import pathlib
import shutil
import typing as t

import pytest

from libvcs import exc
from libvcs._internal.run import run
from libvcs._internal.shortcuts import create_project
from libvcs.pytest_plugin import hg_remote_repo_single_commit_post_init
from libvcs.sync.base import SyncResult
from libvcs.sync.hg import HgSync

if t.TYPE_CHECKING:
    from libvcs.pytest_plugin import CreateRepoFn

if not shutil.which("hg"):
    pytestmark = pytest.mark.skip(reason="hg is not available")


def test_hg_sync(
    tmp_path: pathlib.Path,
    projects_path: pathlib.Path,
    hg_remote_repo: pathlib.Path,
) -> None:
    """Test HgSync."""
    repo_name = "my_mercurial_project"

    mercurial_repo = HgSync(
        url=f"file://{hg_remote_repo}",
        path=projects_path / repo_name,
    )

    run(["hg", "init", mercurial_repo.repo_name], cwd=tmp_path)

    mercurial_repo.update_repo()

    test_repo_revision = run(
        ["hg", "parents", "--template={rev}"],
        cwd=projects_path / repo_name,
    )

    assert mercurial_repo.get_revision() == test_repo_revision


def test_repo_mercurial_via_create_project(
    tmp_path: pathlib.Path,
    projects_path: pathlib.Path,
    hg_remote_repo: pathlib.Path,
) -> None:
    """Test HgSync via create_project()."""
    repo_name = "my_mercurial_project"

    mercurial_repo = create_project(
        url=f"file://{hg_remote_repo}",
        path=projects_path / repo_name,
        vcs="hg",
    )

    run(["hg", "init", mercurial_repo.repo_name], cwd=tmp_path)

    mercurial_repo.update_repo()

    test_repo_revision = run(
        ["hg", "parents", "--template={rev}"],
        cwd=projects_path / repo_name,
    )

    assert mercurial_repo.get_revision() == test_repo_revision


def test_vulnerability_2022_03_12_command_injection(
    monkeypatch: pytest.MonkeyPatch,
    user_path: pathlib.Path,
    tmp_path: pathlib.Path,
    hg_remote_repo: pathlib.Path,
) -> None:
    """Prevent hg aliases from executed arbitrary commands via URLs.

    As of 0.11 this code path is/was only executed via .obtain(), so this only would
    effect explicit invocation of .object() or update_repo() of uncloned destination.
    """
    random_dir = tmp_path / "random"
    random_dir.mkdir()
    monkeypatch.chdir(str(random_dir))
    mercurial_repo = create_project(
        url="--config=alias.clone=!touch ./HELLO",
        vcs="hg",
        path="./",
    )
    result = mercurial_repo.update_repo()

    assert not result.ok, "update_repo() should report failure for malicious URL"
    assert any(e.step == "obtain" for e in result.errors), (
        "Error should be recorded under 'obtain' step"
    )
    assert not pathlib.Path(
        random_dir / "HELLO",
    ).exists(), "Prevent command injection in hg aliases"


def test_update_repo_pull_failure_returns_sync_result(
    projects_path: pathlib.Path,
    create_hg_remote_repo: CreateRepoFn,
) -> None:
    """Test that a deleted remote in update_repo() returns SyncResult with error."""
    repo_name = "my_hg_error_project"
    hg_remote = create_hg_remote_repo(
        remote_repo_post_init=hg_remote_repo_single_commit_post_init,
    )

    hg_repo = HgSync(
        url=f"file://{hg_remote}",
        path=projects_path / repo_name,
    )

    hg_repo.update_repo()

    shutil.rmtree(hg_remote)

    result = hg_repo.update_repo()

    assert isinstance(result, SyncResult)
    assert result.ok is False
    assert len(result.errors) > 0
    assert result.errors[0].step == "pull"
    assert isinstance(result.errors[0].exception, exc.CommandError)