build_manifest.py 8.91 KB
Newer Older
1
2
3
4
5
6
#!/usr/bin/env python3

import os
import requests
import sys

7
from typing import Dict, Tuple, List
8
from urllib.parse import urlparse
9
# from pprint import pprint
10

11
12
GSTREAMER_MODULES: List[str] = [
    # 'orc',
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
    'gst-build',
    'gstreamer',
    'gst-plugins-base',
    'gst-plugins-good',
    'gst-plugins-bad',
    'gst-plugins-ugly',
    'gst-libav',
    'gst-devtools',
    'gst-docs',
    'gst-editing-services',
    'gst-omx',
    'gst-python',
    'gst-rtsp-server'
]

28
MANIFEST_TEMPLATE: str = """<?xml version="1.0" encoding="UTF-8"?>
29
<manifest>
30
  <remote fetch="{}" name="user"/>
31
32
  <remote fetch="https://gitlab.freedesktop.org/gstreamer/" name="gstreamer"/>
  <remote fetch="git://anongit.freedesktop.org/gstreamer/" name="origin"/>
33
{}
34
35
36
</manifest>"""


37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# Basically, pytest will happily let a test mutate a variable, and then run
# the next tests one the same environment without reset the vars.
def preserve_ci_vars(func):
    """Preserve the original CI Variable values"""
    def wrapper():
        try:
            token = os.environ["CI_JOB_TOKEN"]
            url = os.environ["CI_PROJECT_URL"]
            user = os.environ["GITLAB_USER_LOGIN"]
        except KeyError:
            token = "invalid"
            url = "invalid"
            user = ""

        func()

        os.environ["CI_JOB_TOKEN"] = token
        os.environ["CI_PROJECT_URL"] = url
        os.environ["GITLAB_USER_LOGIN"] = user

    return wrapper


60
def request_raw(path: str, token: str, project_url: str) -> List[Dict[str, str]]:
61
62
    gitlab_header: Dict[str, str] = {'JOB_TOKEN': token }
    base_url: str = get_hostname(project_url)
63
    url: str = f"https://{base_url}/api/v4/{path}"
64
    print(f"GET {url}")
65
    resp = requests.get(url, headers=gitlab_header)
66

67
    print(f"Request returned: {resp.status_code}")
68
69
70
71
    if not resp.ok:
        return None

    return resp.json()
72

73

74
def request(path: str) -> List[Dict[str, str]]:
75
76
77
78
    # mock: "xxxxxxxxxxxxxxxxxxxx"
    token: str = os.environ["CI_JOB_TOKEN"]
    # mock: "https://gitlab.freedesktop.org/gstreamer/gstreamer"
    project_url: str = os.environ['CI_PROJECT_URL']
79
80
81
    return request_raw(path, token, project_url)


82
def request_wrap(path: str) -> List[Dict[str, str]]:
83
84
    resp: List[Dict[str, str]] = request(path)

85
86
87
    if not resp:
        return None
    if not resp[0]:
88
89
        return None

90
91
92
    # Not sure if there will be any edge cases where it returns more than one
    # so lets see if anyone complains
    assert len(resp) == 1
93
94
95
    return resp[0]


96
def get_project_branch(project_id: int, name: str) -> Dict[str, str]:
97
    print(f"Searching for {name} branch in project {project_id}")
98
99
100
101
    path = f"projects/{project_id}/repository/branches?search={name}"
    return request_wrap(path)


102
@preserve_ci_vars
103
104
105
106
107
108
109
110
111
112
113
114
115
def test_get_project_branch():
    id = 1353
    os.environ["CI_JOB_TOKEN"] = "xxxxxxxxxxxxxxxxxxxx"
    os.environ["CI_PROJECT_URL"] = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-good"

    twelve = get_project_branch(id, '1.12')
    assert twelve is not None
    assert twelve['name'] == '1.12'

    fourteen = get_project_branch(id, '1.14')
    assert fourteen is not None
    assert fourteen['name'] == '1.14'

116
117
118
    failure = get_project_branch(id, 'why-would-anyone-chose-this-branch-name')
    assert failure is None

119
120
121
    failure2 = get_project_branch("invalid-id", '1.12')
    assert failure2 is None

122

123
124
# Documentation: https://docs.gitlab.com/ce/api/projects.html#list-user-projects
def search_user_namespace(user: str, project: str) -> Dict[str, str]:
125
    print(f"Searching for {project} project in @{user} user's namespace")
126
    path = f"/users/{user}/projects?search={project}"
127
    return request_wrap(path)
128
129


130
@preserve_ci_vars
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def test_search_user_namespace():
    os.environ["CI_JOB_TOKEN"] = "xxxxxxxxxxxxxxxxxxxx"
    os.environ["CI_PROJECT_URL"] = "https://gitlab.freedesktop.org/alatiera/gst-plugins-good"
    user = "alatiera"

    gst = search_user_namespace("alatiera", "gstreamer")
    assert gst is not None
    assert gst['path'] == 'gstreamer'

    gst_good = search_user_namespace("alatiera", "gst-plugins-good")
    assert gst_good is not None
    assert gst_good['path'] == 'gst-plugins-good'

    res = search_user_namespace("alatiera", "404-project-not-found")
    assert res is None

147
148
# Documentation: https://docs.gitlab.com/ee/api/search.html#group-search-api
def search_group_namespace(group_id: str, project: str) -> Dict[str, str]:
149
    print(f"Searching for {project} project in @{group_id} group namespace")
150
    path = f"groups/{group_id}/search?scope=projects&search={project}"
151
    return request_wrap(path)
152
153


154
@preserve_ci_vars
155
156
157
def test_search_group_namespace():
    import pytest
    try:
158
159
160
161
162
163
164
165
        instance = get_hostname(os.environ['CI_PROJECT_URL'])
        # This tests need to authenticate with the gitlab instace,
        # and its hardcoded to check for the gitlab-org which exists
        # on gitlab.com
        # This can be changed to a gstreamer once its migrated to gitlab.fd.o
        if instance != "gitlab.freedesktop.org":
            # too lazy to use a different error
            raise KeyError
166
167
168
169
170
171
172
173
174
175
176
177
178
    except KeyError:
        pytest.skip("Need to be authenticated with gitlab to succed")

    os.environ["CI_PROJECT_URL"] = "https://gitlab.freedesktop.org/gstreamer/gstreamer"
    group = "gstreamer"

    lab = search_group_namespace(group, "gstreamer")
    assert lab is not None
    assert lab['path'] == 'gstreamer'

    res = search_user_namespace(group, "404-project-not-found")
    assert res is None

179

180
181
182
183
184
185
186
187
188
189
190
191
def get_hostname(url: str) -> str:
    return urlparse(url).hostname


def test_get_hostname():
    gitlab = 'https://gitlab.com/example/a_project'
    assert get_hostname(gitlab) == 'gitlab.com'

    fdo = 'https://gitlab.freedesktop.org/example/a_project'
    assert get_hostname(fdo) == 'gitlab.freedesktop.org'


192
def find_repository_sha(module: str, branchname: str) -> Tuple[str, str]:
193
194
    user_login: str = os.environ["GITLAB_USER_LOGIN"]
    project = search_user_namespace(user_login, module)
195

196
    # Find a fork in the User's namespace
197
    if project:
198
        id = project['id']
199
        print(f"User project found, id: {id}")
200
201
202
        # If we have a branch with same name, use it.
        branch = get_project_branch(id, branchname)
        if branch is not None:
203
204
205
            path = project['namespace']['path']
            print("Found mathcing branch in user's namespace")
            print(f"{path}/{branchname}")
206
207

            return 'user', branch['commit']['id']
208
        print(f"Did not found user branch named {branchname}")
209

210
    # This won't work until gstreamer migrates to gitlab
211
212
213
    # Else check the upstream gstreamer repository
    project = search_group_namespace('gstreamer', module)
    if project:
214
        print(f"Project found in Gstreamer upstream, id: {id}")
215
216
        id = project['id']
        # If we have a branch with same name, use it.
217
218
        branch = get_project_branch(id, branchname)
        if branch is not None:
219
            print("Found matching branch in upstream gst repo")
220
221
222
223
224
            print(f"gstreamer/{branchname}")
            return 'gstreamer', branch['commit']['id']

        branch = get_project_branch(id, 'master')
        if branch is not None:
225
            # print("Falling back to master branch in upstream repo")
226
227
            print('gstreamer/master')
            return 'gstreamer', branch.attributes['commit']['id']
228

229
    print('Falling back to origin/master')
230
231
232
    return 'origin', 'master'


233
@preserve_ci_vars
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def test_find_repository_sha():
    os.environ["CI_JOB_TOKEN"] = "xxxxxxxxxxxxxxxxxxxx"
    os.environ["CI_PROJECT_URL"] = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-good"
    os.environ["GITLAB_USER_LOGIN"] = "alatiera"

    # This should find the repository in the user namespace
    remote, git_ref = find_repository_sha("gst-plugins-good", "1.2")
    assert remote == "user"
    assert git_ref == "08ab260b8a39791e7e62c95f4b64fd5b69959325"

    # This should fallback to upstream master branch since no matching branch was found
    remote, git_ref = find_repository_sha("gst-plugins-good", "totally-valid-branch-name")
    assert remote == "origin"
    assert git_ref == "master"

    # This should fallback to upstream master branch since no repository was found
    remote, git_ref = find_repository_sha("totally-valid-project-name", "1.2")
    assert remote == "origin"
    assert git_ref == "master"


255
if __name__ == "__main__":
256
    projects: str = ''
257
    project_template: str = "  <project name=\"{}\" remote=\"{}\" revision=\"{}\" />\n"
258
    user_remote_url: str = os.path.dirname(os.environ['CI_PROJECT_URL'])
259
260
    if not user_remote_url.endswith('/'):
        user_remote_url += '/'
261

262
    for module in GSTREAMER_MODULES:
263
        print(f"Checking {module}:", end=' ')
264
265
        current_branch: str = os.environ['CI_COMMIT_REF_NAME']
        remote, revision = find_repository_sha(module, current_branch)
266
        projects += project_template.format(module, remote, revision)
267
268

    with open('manifest.xml', mode='w') as manifest:
269
        print(MANIFEST_TEMPLATE.format(user_remote_url, projects), file=manifest)