security-scanner/tests/test_cli_quickstart.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
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
"""CLI tests for quickstart."""

from __future__ import annotations

from pathlib import Path

from security_scanner.cli import main
from security_scanner.runtime.quickstart import QuickstartSummary
from security_scanner.runtime.incremental_discovery import IncrementalDiscoverySummary
from security_scanner.runtime.scan_worker import ScanWorkerSummary
from security_scanner.storage.base import QueueStatus


TARGET_URL = "https://github.com/example-org/example-repo"
CUSTOM_GITLAB_URL = "https://source.example.test/example-group/example-repo"


def test_quickstart_cli_runs_runtime_with_dynamodb_defaults(monkeypatch, capsys):
    calls = {}

    monkeypatch.setattr(
        "security_scanner.cli.commands.quickstart.run_doctor",
        lambda target_url, private, scm_provider: type(
            "Doctor",
            (),
            {"ok": True, "checks": []},
        )(),
    )
    monkeypatch.setattr("security_scanner.cli._store.store_from_args", lambda args: object())

    def fake_run_quickstart(request):
        calls["request"] = request
        return QuickstartSummary(
            target_url=request.target_url,
            initialized=IncrementalDiscoverySummary(targets=1, refs_observed=1),
            current_jobs_enqueued=1,
            enqueued=IncrementalDiscoverySummary(targets=1),
            worker=ScanWorkerSummary(leased=1, completed=1),
            status=QueueStatus(
                job_counts_by_status={"completed": 1},
                expired_job_leases=0,
                expired_repo_leases=0,
            ),
        )

    monkeypatch.setattr("security_scanner.cli.commands.quickstart.run_quickstart", fake_run_quickstart)

    exit_code = main(["quickstart", TARGET_URL])

    captured = capsys.readouterr()
    assert exit_code == 0
    assert calls["request"].target_url == TARGET_URL
    assert calls["request"].max_jobs == 10
    assert "current jobs enqueued: 1" in captured.out
    assert "worker completed: 1" in captured.out


def test_quickstart_cli_fails_on_private_auth_preflight(monkeypatch, capsys):
    doctor = type("Doctor", (), {"ok": False, "checks": []})()
    monkeypatch.setattr(
        "security_scanner.cli.commands.quickstart.run_doctor",
        lambda target_url, private, scm_provider: doctor,
    )
    monkeypatch.setattr(
        "security_scanner.cli.commands.quickstart.render_doctor_result",
        lambda result: "fail: github auth - set GH_TOKEN or run gh auth login\n",
    )

    exit_code = main(["quickstart", TARGET_URL, "--private"])

    captured = capsys.readouterr()
    assert exit_code == 1
    assert "set GH_TOKEN" in captured.err
    assert "ghp_" not in captured.err


def test_quickstart_cli_passes_scm_provider_to_fetcher(monkeypatch, tmp_path):
    calls = {}

    monkeypatch.setattr(
        "security_scanner.cli.commands.quickstart.run_doctor",
        lambda target_url, private, scm_provider: type(
            "Doctor",
            (),
            {"ok": True, "checks": []},
        )(),
    )
    monkeypatch.setattr("security_scanner.cli._store.store_from_args", lambda args: object())

    def fake_fetch_or_clone(url, **kwargs):
        calls["fetch"] = (url, kwargs)
        return tmp_path / "repo"

    def fake_run_quickstart(request):
        fetched = request.fetch_repo(request.target_url)
        assert isinstance(fetched, Path)
        return QuickstartSummary(
            target_url=request.target_url,
            initialized=IncrementalDiscoverySummary(targets=1),
            current_jobs_enqueued=0,
            enqueued=IncrementalDiscoverySummary(targets=1),
            worker=None,
            status=QueueStatus(
                job_counts_by_status={},
                expired_job_leases=0,
                expired_repo_leases=0,
            ),
        )

    monkeypatch.setattr("security_scanner.cli.commands.quickstart.fetch_or_clone", fake_fetch_or_clone)
    monkeypatch.setattr("security_scanner.cli.commands.quickstart.run_quickstart", fake_run_quickstart)

    exit_code = main(
        [
            "quickstart",
            CUSTOM_GITLAB_URL,
            "--scm-provider",
            "gitlab",
            "--private",
        ]
    )

    assert exit_code == 0
    assert calls["fetch"] == (
        CUSTOM_GITLAB_URL,
        {"allow_git_fallback": False, "scm_provider": "gitlab"},
    )