openhound-github/tests/test_source_org_canonicalization.py at 8390738cde7e185de5db6958a2456251c0c0c3f2 · SpecterOps/openhound-github · 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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
import importlib
import inspect

import pytest

from openhound_github.models import Organization
from openhound_github.resources.organization import organizations
from openhound_github.source import (
    GithubOrgAppCredentials,
    GithubTokenCredentials,
    OrgContext,
    SourceContext,
    _canonicalize_org_names,
)


class _FakeResponse:
    def __init__(self, payload: dict):
        self._payload = payload

    def json(self) -> dict:
        return self._payload


class _FakeClient:
    def __init__(self, payloads: dict[str, dict]):
        self.payloads = payloads
        self.get_calls: list[str] = []

    def get(self, path: str) -> _FakeResponse:
        self.get_calls.append(path)
        return _FakeResponse(self.payloads[path])


def test_canonicalize_org_names_rewrites_org_context_and_caches_response() -> None:
    client = _FakeClient(
        {
            "/orgs/spectertst": {
                "login": "SpecterTst",
                "node_id": "O_kgDOCoV2OQ",
            }
        }
    )
    ctx = SourceContext(
        organizations=[OrgContext(client=client, org_name="spectertst")]
    )

    _canonicalize_org_names(ctx)

    assert ctx.organizations[0].org_name == "SpecterTst"
    assert ctx.organizations_cache == {
        "SpecterTst": {
            "login": "SpecterTst",
            "node_id": "O_kgDOCoV2OQ",
        }
    }


def test_organizations_reuses_preflight_org_response() -> None:
    client = _FakeClient(
        {
            "/orgs/spectertst": {
                "id": 123456,
                "login": "SpecterTst",
                "node_id": "O_kgDOCoV2OQ",
            },
            "/orgs/SpecterTst/actions/permissions": {},
            "/orgs/SpecterTst/actions/permissions/self-hosted-runners": {},
            "/orgs/SpecterTst/actions/permissions/workflow": {},
        }
    )
    ctx = SourceContext(
        organizations=[OrgContext(client=client, org_name="spectertst")],
        deployment_type="ghes",
        ghes_version="3.22.1",
    )

    _canonicalize_org_names(ctx)
    rows = list(inspect.unwrap(organizations._pipe.gen)(ctx))

    assert rows[0]["login"] == "SpecterTst"
    assert rows[0]["database_id"] == 123456
    assert rows[0]["github_deployment_type"] == "ghes"
    assert rows[0]["ghes_version"] == "3.22.1"
    node = Organization(**rows[0]).as_node
    assert node.properties.database_id == 123456
    assert node.properties.node_id == "O_kgDOCoV2OQ"
    assert node.properties.github_deployment_type == "ghes"
    assert node.properties.ghes_version == "3.22.1"
    assert client.get_calls == [
        "/orgs/spectertst",
        "/orgs/SpecterTst/actions/permissions",
        "/orgs/SpecterTst/actions/permissions/self-hosted-runners",
        "/orgs/SpecterTst/actions/permissions/workflow",
    ]


@pytest.mark.parametrize(
    "credentials",
    (
        GithubTokenCredentials(token="token", org_name="spectertst"),
        GithubOrgAppCredentials(
            client_id="Iv1.client-id",
            install_id="12345",
            key_path="/tmp/github-app.pem",
            org_name="spectertst",
        ),
    ),
)
def test_org_only_sources_canonicalize_before_resource_fanout(
    monkeypatch: pytest.MonkeyPatch,
    credentials,
) -> None:
    source_module = importlib.import_module("openhound_github.source")
    captured_ctx: dict[str, SourceContext] = {}

    class FakeRESTClient:
        def __init__(self, **kwargs) -> None:
            pass

        def get(self, path: str) -> _FakeResponse:
            assert path == "/orgs/spectertst"
            return _FakeResponse({"login": "SpecterTst", "node_id": "O_kgDOCoV2OQ"})

    monkeypatch.setattr(source_module, "RESTClient", FakeRESTClient)
    monkeypatch.setattr(source_module, "GithubInstallation", lambda **_: object())
    monkeypatch.setattr(
        source_module, "GitHubAppInstallationAuth", lambda **_: object()
    )

    def fake_organization_resources(ctx: SourceContext):
        captured_ctx["ctx"] = ctx
        return ()

    monkeypatch.setattr(
        source_module, "organization_resources", fake_organization_resources
    )

    resources = source_module.source.__wrapped__(
        credentials=credentials,
        host="https://api.github.com",
        emit_legacy_scim_correlations=False,
    )

    assert resources == ()
    assert captured_ctx["ctx"].organizations[0].org_name == "SpecterTst"
    assert (
        captured_ctx["ctx"].organizations_cache["SpecterTst"]["node_id"]
        == "O_kgDOCoV2OQ"
    )