security-scanner/tests/test_cli_sync.py at main · source-security-dev/security-scanner · 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
import pytest

from security_scanner.catalog.scan_target import ScanTarget
from security_scanner.cli.commands import targets
from security_scanner.cli.app import main
from security_scanner.targets.repo_lister import DiscoveredRepo


@pytest.fixture
def fake_store(monkeypatch):
    class _Store:
        def __init__(self):
            self._targets: dict[str, ScanTarget] = {}

        def put_scan_target(self, target: ScanTarget) -> None:
            self._targets[target.url] = target

        def list_scan_targets(self):
            return list(self._targets.values())

        def delete_scan_target(self, url: str) -> None:
            self._targets.pop(url, None)

    store = _Store()
    monkeypatch.setattr(targets, "store_from_args", lambda args: store)
    return store


def _fake_lister(monkeypatch, repos):
    monkeypatch.setattr(targets, "list_owner_repos", lambda owner, **kw: repos)


def test_sync_inserts_new_repos_disabled_by_default(fake_store, monkeypatch, capsys):
    _fake_lister(monkeypatch, [
        DiscoveredRepo("https://github.com/o/a", "o/a", False),
        DiscoveredRepo("https://github.com/o/b", "o/b", False),
    ])
    code = main(["sync", "o", "--storage-backend", "dynamodb"])
    assert code == 0
    targets = {t.url: t for t in fake_store.list_scan_targets()}
    assert targets["https://github.com/o/a"].enabled is False
    out = capsys.readouterr().out
    assert "discovered: 2, inserted: 2, existing: 0" in out
    assert "enable-target" in out


def test_sync_preserves_existing_enabled_state(fake_store, monkeypatch, capsys):
    fake_store.put_scan_target(ScanTarget("https://github.com/o/a", "o/a", enabled=True))
    _fake_lister(monkeypatch, [
        DiscoveredRepo("https://github.com/o/a", "o/a", False),
        DiscoveredRepo("https://github.com/o/b", "o/b", False),
    ])
    code = main(["sync", "o", "--storage-backend", "dynamodb"])
    assert code == 0
    targets = {t.url: t for t in fake_store.list_scan_targets()}
    assert targets["https://github.com/o/a"].enabled is True
    assert targets["https://github.com/o/b"].enabled is False
    assert "discovered: 2, inserted: 1, existing: 1" in capsys.readouterr().out


def test_sync_enable_all_inserts_enabled(fake_store, monkeypatch):
    _fake_lister(monkeypatch, [DiscoveredRepo("https://github.com/o/a", "o/a", False)])
    main(["sync", "o", "--enable-all", "--storage-backend", "dynamodb"])
    assert fake_store.list_scan_targets()[0].enabled is True


def test_sync_returns_1_on_sync_error(fake_store, monkeypatch, capsys):
    from security_scanner.targets.repo_lister import SyncError

    def boom(owner, **kw):
        raise SyncError("gh auth login required")

    monkeypatch.setattr(targets, "list_owner_repos", boom)
    code = main(["sync", "o", "--storage-backend", "dynamodb"])
    assert code == 1
    assert "gh auth login required" in capsys.readouterr().err


def test_sync_rejects_non_github_owner(fake_store, capsys):
    code = main(["sync", "gitlab.com/group", "--storage-backend", "dynamodb"])
    assert code == 1
    assert "bare GitHub owner" in capsys.readouterr().err