build_manifest.py 8.86 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 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29
# Each item is a Tuple of (project-path, project-id)
# ex. https://gitlab.freedesktop.org/gstreamer/gst-build
# has project path 'gst-build' and project-id '1342'
# TODO: Named tuples are awesome
GSTREAMER_MODULES: List[Tuple[str, int]] = [
    # ('orc', 1360),
    ('gst-build', 1342),
    ('gstreamer', 1357),
    ('gst-plugins-base', 1352),
    ('gst-plugins-good', 1353),
    ('gst-plugins-bad', 1351),
    ('gst-plugins-ugly', 1354),
    ('gst-libav', 1349),
    ('gst-devtools', 1344),
    ('gst-docs', 1345),
    ('gst-editing-services', 1346),
    ('gst-omx', 1350),
    ('gst-python', 1355),
    ('gst-rtsp-server', 1362),
30
    ('gstreamer-vaapi', 1359),
31 32
]

33
MANIFEST_TEMPLATE: str = """<?xml version="1.0" encoding="UTF-8"?>
34
<manifest>
35
  <remote fetch="{}" name="user"/>
36
  <remote fetch="https://gitlab.freedesktop.org/gstreamer/" name="origin"/>
37
{}
38 39 40
</manifest>"""


41 42 43 44 45 46 47
# 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:
            url = os.environ["CI_PROJECT_URL"]
48
            user = os.environ["CI_PROJECT_NAMESPACE"]
49 50 51 52
        except KeyError:
            url = "invalid"
            user = ""

53 54 55 56
        private = os.getenv("READ_PROJECTS_TOKEN", default=None)
        if not private:
            os.environ["READ_PROJECTS_TOKEN"] = "FOO"

57 58 59
        func()

        os.environ["CI_PROJECT_URL"] = url
60
        os.environ["CI_PROJECT_NAMESPACE"] = user
61

62 63 64 65 66 67
        if private:
            os.environ["READ_PROJECTS_TOKEN"] = private
        # if it was set after
        elif os.getenv("READ_PROJECTS_TOKEN", default=None):
            del os.environ["READ_PROJECTS_TOKEN"]

68 69 70
    return wrapper


71
def request_raw(path: str, headers: Dict[str, str], project_url: str) -> List[Dict[str, str]]:
72 73
    # ex. base_url = "gitlab.freedesktop.org"
    base_url: str = urlparse(project_url).hostname
74
    url: str = f"https://{base_url}/api/v4/{path}"
75
    print(f"GET {url}")
76
    # print(f"Headers: {headers}")
77
    resp = requests.get(url, headers=headers)
78

79
    print(f"Request returned: {resp.status_code}")
80 81 82 83
    if not resp.ok:
        return None

    return resp.json()
84

85

86
def request(path: str) -> List[Dict[str, str]]:
87 88 89 90 91 92 93 94 95 96 97
    # Check if there is a custom token set
    # API calls to Group namespaces need to be authenticated
    # regardless if the group/projects are public or not.
    # CI_JOB_TOKEN has an actuall value only for private jobs
    # and that's also an Gitlab EE feature.
    # Which means no matter what we need to give the runner
    # an actuall token if we want to query even the public
    # gitlab.fd.o/gstreamer group
    try:
        headers: Dict[str, str] = {'Private-Token': os.environ["READ_PROJECTS_TOKEN"] }
    except KeyError:
98
        # print("Custom token was not set, group api querries will fail")
99 100 101
        # JOB_TOKEN is the default placeholder of CI_JOB_TOKEN
        headers: Dict[str, str] = {'JOB_TOKEN': "xxxxxxxxxxxxxxxxxxxx" }

102 103
    # mock: "https://gitlab.freedesktop.org/gstreamer/gstreamer"
    project_url: str = os.environ['CI_PROJECT_URL']
104
    return request_raw(path, headers, project_url)
105 106


107 108 109 110
def get_project_branch(project_id: int, name: str) -> Dict[str, str]:
    print(f"Searching for {name} branch in project {project_id}")
    path = f"projects/{project_id}/repository/branches?search={name}"
    results = request(path)
111

112
    if not results:
113
        return None
114

115 116 117 118 119
    # The api returns a list of projects that match the search
    # we want the exact match, which might not be the first on the list
    for project_res in results:
        if project_res["name"] == name:
            return project_res
120

121
    return None
122 123


124
@preserve_ci_vars
125 126 127
def test_get_project_branch():
    id = 1353
    os.environ["CI_PROJECT_URL"] = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-good"
128
    del os.environ["READ_PROJECTS_TOKEN"]
129 130 131 132 133 134 135 136 137

    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'

138 139 140
    failure = get_project_branch(id, 'why-would-anyone-chose-this-branch-name')
    assert failure is None

141 142 143
    failure2 = get_project_branch("invalid-id", '1.12')
    assert failure2 is None

144

145 146
# Documentation: https://docs.gitlab.com/ce/api/projects.html#list-user-projects
def search_user_namespace(user: str, project: str) -> Dict[str, str]:
147
    print(f"Searching for {project} project in @{user} user's namespace")
148
    path = f"users/{user}/projects?search={project}"
149 150 151 152 153 154 155 156 157 158 159 160
    results = request(path)

    if not results:
        return None

    # The api returns a list of projects that match the search
    # we want the exact match, which might not be the first on the list
    for project_res in results:
        if project_res["path"] == project:
            return project_res

    return None
161 162


163
@preserve_ci_vars
164 165
def test_search_user_namespace():
    os.environ["CI_PROJECT_URL"] = "https://gitlab.freedesktop.org/alatiera/gst-plugins-good"
166
    del os.environ["READ_PROJECTS_TOKEN"]
167 168 169 170 171 172 173 174 175 176 177 178 179
    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

180 181 182 183
    # Passing a group namespace instead of user should return None
    res = search_user_namespace("gstreamer", "gst-plugins-good")
    assert res is None

184

185
def find_repository_sha(module: Tuple[str, int], branchname: str) -> Tuple[str, str]:
186 187
    namespace: str = os.environ["CI_PROJECT_NAMESPACE"]
    project = search_user_namespace(namespace, module[0])
188

189
    # Find a fork in the User's namespace
190
    if project:
191
        id = project['id']
192
        print(f"User project found, id: {id}")
193 194 195
        # If we have a branch with same name, use it.
        branch = get_project_branch(id, branchname)
        if branch is not None:
196 197 198
            path = project['namespace']['path']
            print("Found mathcing branch in user's namespace")
            print(f"{path}/{branchname}")
199 200

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

203 204 205 206 207
    # Check upstream project for a branch
    branch = get_project_branch(module[1], branchname)
    if branch is not None:
        print("Found mathcing branch in upstream project")
        print(f"gstreamer/{branchname}")
208
        return 'origin', branch['commit']['id']
209 210 211 212 213 214

    # Fallback to using upstream master branch
    branch = get_project_branch(module[1], 'master')
    if branch is not None:
        print("Falling back to master branch on upstream project")
        print(f"gstreamer/master")
215
        return 'origin', branch['commit']['id']
216 217 218 219 220

    # This should never occur given the upstream fallback above
    print("If something reaches that point, please file a bug")
    print("https://gitlab.freedesktop.org/gstreamer/gst-ci/issues")
    assert False
221 222


223
@preserve_ci_vars
224 225
def test_find_repository_sha():
    os.environ["CI_PROJECT_URL"] = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-good"
226
    os.environ["CI_PROJECT_NAMESPACE"] = "alatiera"
227
    del os.environ["READ_PROJECTS_TOKEN"]
228 229

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

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

    # This should fallback to upstream master branch since no repository was found
239
    remote, git_ref = find_repository_sha(("totally-valid-project-name", 42), "1.2")
240
    assert remote == "origin"
241 242
    # This is now the sha of the last commit
    # assert git_ref == "master"
243 244


245
if __name__ == "__main__":
246
    projects: str = ''
247
    project_template: str = "  <project name=\"{}\" remote=\"{}\" revision=\"{}\" />\n"
248
    user_remote_url: str = os.path.dirname(os.environ['CI_PROJECT_URL'])
249 250
    if not user_remote_url.endswith('/'):
        user_remote_url += '/'
251

252
    for module in GSTREAMER_MODULES:
253
        print(f"Checking {module}:", end=' ')
254 255
        current_branch: str = os.environ['CI_COMMIT_REF_NAME']
        remote, revision = find_repository_sha(module, current_branch)
256
        projects += project_template.format(module[0], remote, revision)
257 258

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