Skip to content

API Reference / Source Code

Below is a list of all classes and functions used for CLA Bot. Those are considered to be 'internal', with no need for users to access any of these. Please use the CLI tools instead.

CLA Bot relies on third party modules to interface with GitLab, git, SQLite, and csv. Please check pyproject.toml for more details on the required packages.

'Core' components

src.bot.cla_bot

run(argv=sys.argv)

Wrapper to run CLA Bot, adds command line interface.

This will read all options from the command line or environment.

Source code in src/bot/cla_bot.py 
 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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def run(argv=sys.argv):
    """Wrapper to run CLA Bot, adds command line interface.

    This will read all options from the command line or environment.
    """
    from importlib.metadata import PackageNotFoundError, version

    import configargparse

    try:
        __version__ = version("cla-bot")
    except PackageNotFoundError:
        # package is not installed
        __version__ = "unknown"

    # Parse command line options
    parser = configargparse.ArgumentParser(
        prog=argv[0],
        description="A bot to check compliance with your CLA.",
        formatter_class=configargparse.ArgumentDefaultsHelpFormatter,
    )
    #### General 'settings'
    parser.add_argument(
        "-n",
        "--dry-run",
        dest="dry",
        action="store_true",
        default=False,
        help="only perform dry run, do not alter project",
    )
    parser.add_argument(
        "-f",
        "--force",
        dest="force",
        action="store_true",
        default=False,
        help="force the bot, ignoring other pipelines or running bots "
        "(may corrupt data or issue multiple comments)",
    )
    parser.add_argument(
        "-m",
        "--merge-id",
        dest="mr_iid",
        type=int,
        default=None,
        env_var="CI_MERGE_REQUEST_IID",
        help="process merge request with this id (project (i)id, not instance wide id)",
    )
    parser.add_argument(
        "-g",
        "--gitlab",
        dest="gitlab",
        action="store_true",
        default=False,
        env_var="CLA_BOT_GITLAB_WORKFLOW",
        help="work completely within gitlab (no papertrail)",
    )
    parser.add_argument(
        "-sb",
        "--store-branch",
        dest="store_branch",
        type=str,
        default="main",
        env_var="CLA_BOT_STORAGE_BRANCH",
        help="branch inside CLA store to commit data to (string)",
    )
    parser.add_argument(
        "-sts",
        "--store-scope",
        dest="store_scope",
        type=str,
        default="private",
        env_var="CLA_BOT_STORAGE_SCOPE",
        help="scope of CLA Store, private (single project) or global (multiple projects) (string)",
    )
    parser.add_argument(
        "-sty",
        "--store-type",
        dest="store_type",
        type=str,
        default="csv",
        env_var="CLA_BOT_STORAGE_TYPE",
        help="type of CLA Store, csv or SQLite (string)",
    )
    parser.add_argument(
        "-a",
        "--allow-list",
        dest="allow_list",
        type=str,
        default="",
        env_var="CLA_BOT_ALLOW_LIST",
        help="comma separated list of always approved contributors (user names)",
    )
    parser.add_argument(
        "-aa",
        "--auto-allow",
        dest="auto_allow",
        type=str,
        default="",
        env_var="CLA_BOT_AUTO_ALLOW",
        help="minimum project roles ('Developer', 'Maintainer', 'Owner') "
        "where project members are automatically approved",
    )
    #### Playground project
    parser.add_argument(
        "-pp",
        "--playground",
        dest="playground_id",
        type=str,
        default=None,
        env_var="CLA_BOT_PLAYGROUND",
        help="PLAYGROUND project to check contributions in (numeric id)",
    )
    parser.add_argument(
        "-pt",
        "--playground-token",
        dest="playground_token",
        type=str,
        default=None,
        help="personal access token (API) to access PLAYGROUND (repo to inspect)",
    )
    #### Home project
    parser.add_argument(
        "-hp",
        "--home",
        dest="home_id",
        type=str,
        default=None,
        env_var="CI_PROJECT_ID",
        help="HOME project (numeric id)",
    )
    parser.add_argument(
        "-ht",
        "--home-token",
        dest="home_token",
        type=str,
        default=None,
        env_var="CLA_BOT_API_TOKEN",
        help="personal access token (API) to access HOME (CLA-Bot's repo)",
    )
    #### Store project
    parser.add_argument(
        "-sp",
        "--store",
        dest="store_url",
        type=str,
        default=None,
        env_var="CLA_BOT_STORAGE_PROJECT",
        required=True,
        help="project to store CLA signatures (URL)",
    )
    parser.add_argument(
        "-st",
        "--store-token",
        dest="store_token",
        type=str,
        default=None,
        env_var="CLA_BOT_STORAGE_TOKEN",
        required=True,
        help="access token (repository) to access STORE (CLA store project)",
    )
    parser.add_argument(
        "-v",
        "--version",
        action="version",
        version=f"%(prog)s version: {__version__}",
        help="show version",
    )

    args = parser.parse_args()

    # Before we continue, we check the environment to decide in which
    # configuration the bot is supposed to run. Options are:
    #   1) The bot was started by a trigger, then inspect the trigger payload
    #   2) The bot was started as part of a merge pipeline,
    #      The CI environment is read by default
    #   3) The bot was started manually and given command line parameters
    payload = Payload()
    if payload.triggered:
        args.playground_id = payload.project_id
        if payload.trigger_event == "issue note":
            raise SystemExit(f"{tc.YLW}Irrlevant trigger event, exiting.{tc.CLR}")
        else:
            args.mr_iid = payload.trigger_mr_iid

    # Access tokens are expected either as command line argument or defined as
    # CI/CD variables - with a possible indirection:
    # - access token for the project to check is to be stored as
    #   CLA_BOT_API_TOKEN_{args.project}
    # - storage access token for the storage project is to be stored as
    #   CLA_BOT_STORAGE_TOKEN
    if not args.playground_token:
        try:
            args.playground_token = os.environ["CLA_BOT_API_TOKEN_" + str(args.playground_id)]
        except KeyError:
            raise SystemExit(
                f"{tc.RED}Missing environment variable CLA_BOT_API_TOKEN_"
                f"{args.playground_id} to access the project!{tc.CLR}"
            ) from None

    # Fill our three projects
    playground = ProjectData(
        name="Playground",
        id=args.playground_id,
        url=None,
        token=args.playground_token,
    )
    home = ProjectData(
        name="Home",
        id=args.home_id,
        url=None,
        token=args.home_token,
    )
    store = ProjectData(
        name="Store",
        id=None,
        url=args.store_url,
        token=args.store_token,
    )

    # Run main script
    bot = Bot(
        force=args.force,
        dry_run=args.dry,
        gitlab_workflow=args.gitlab,
        allow_list=args.allow_list,
        auto_allow=args.auto_allow,
        store_branch=args.store_branch,
        store_scope=args.store_scope,
        store_type=args.store_type,
        mr_iid=args.mr_iid,
        playground=playground,
        home=home,
        store=store,
        payload=payload,
    )
    bot.exec()

src.bot.add_cla

run(argv=sys.argv)

Wrapper that adds a command line interface to CLA Bot adding signatures.

This will read all options from the command line. Input can also be taken from $CI_COMMIT_MESSAGE if run from within a pipeline.

Source code in src/bot/add_cla.py 
 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
151
152
153
154
155
156
157
158
159
160
def run(argv=sys.argv):
    """Wrapper that adds a command line interface to CLA Bot adding signatures.

    This will read all options from the command line.
    Input can also be taken from $CI_COMMIT_MESSAGE if run from within a pipeline.
    """
    from importlib.metadata import PackageNotFoundError, version

    import configargparse

    try:
        __version__ = version("cla-bot")
    except PackageNotFoundError:
        # package is not installed
        __version__ = "unknown"

    # Parse command line options
    parser = configargparse.ArgumentParser(
        prog=argv[0],
        description="Use CLA-bot to manually add signatures to the database. "
        + "This is really only useful when using a paper trail instead of a fully "
        + "gitlab integrated workflow.",
        formatter_class=configargparse.ArgumentDefaultsHelpFormatter,
    )
    #### General settings
    parser.add_argument(
        "-n",
        "--dry-run",
        dest="dry",
        action="store_true",
        default=False,
        help="only perform dry run, do not alter project",
    )
    parser.add_argument(
        "-sb",
        "--store-branch",
        dest="store_branch",
        type=str,
        default="main",
        env_var="CLA_BOT_STORAGE_BRANCH",
        help="branch inside CLA store to commit data to (string)",
    )
    parser.add_argument(
        "-sts",
        "--store-scope",
        dest="store_scope",
        type=str,
        default="private",
        env_var="CLA_BOT_STORAGE_SCOPE",
        help="scope of CLA Store, private (single project) or global (multiple projects) (string)",
    )
    parser.add_argument(
        "-sty",
        "--store-type",
        dest="store_type",
        type=str,
        default="csv",
        env_var="CLA_BOT_STORAGE_TYPE",
        help="type of CLA Store, csv or SQLite (string)",
    )
    #### Playground project
    parser.add_argument(
        "-pp",
        "--Playground",
        dest="playground_id",
        type=str,
        default=None,
        env_var="CLA_BOT_PLAYGROUND",
        required=True,
        help="PLAYGROUND project to record contributors for (numeric id)",
    )
    #### Store project
    parser.add_argument(
        "-sp",
        "--store",
        dest="store_url",
        type=str,
        default=None,
        env_var="CLA_BOT_STORAGE_PROJECT",
        required=True,
        help="project to store CLA signatures (URL)",
    )
    parser.add_argument(
        "-st",
        "--store-token",
        dest="store_token",
        type=str,
        default=None,
        env_var="CLA_BOT_STORAGE_TOKEN",
        required=True,
        help="access token (repository) to access the CLA store project",
    )
    #### Data to enter into database
    parser.add_argument(
        "--name",
        dest="name",
        type=str,
        default=None,
        help="name of contributor (may be extracted from $CI_COMMIT_MESSAGE)",
    )
    parser.add_argument(
        "--id",
        dest="uid",
        type=str,
        default=None,
        help="gitlab user_id of contributor (may be extracted from $CI_COMMIT_MESSAGE)",
    )
    parser.add_argument(
        "--email",
        dest="email",
        type=str,
        default=None,
        help="email of contributor (may be extracted from $CI_COMMIT_MESSAGE)",
    )
    parser.add_argument(
        "--unique",
        dest="is_individual",
        action="store_true",
        default=True,
        help="is the contributor an indiviual or signing on behalf of an institution? "
        + "(may be extracted from $CI_COMMIT_MESSAGE)",
    )
    parser.add_argument(
        "-v",
        "--version",
        action="version",
        version=f"%(prog)s version: {__version__}",
        help="show version",
    )

    args = parser.parse_args()

    # Run main script
    adder = Adder(
        project=args.playground_id,
        sig_store_token=args.store_token,
        sig_store=args.store_url,
        store_branch=args.store_branch,
        store_scope=args.store_scope,
        store_type=args.store_type,
        name=args.name,
        dry_run=args.dry,
        uid=args.uid,
        email=args.email,
        is_individual=args.is_individual,
    )
    adder.exec()

src.bot.core

Adder

Core component of CLA Bot when adding signatures

Source code in src/bot/core.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
class Adder:
    """Core component of **_CLA Bot_** when adding signatures"""

    def __init__(
        self,
        project,
        sig_store_token,
        sig_store,
        store_branch,
        store_scope,
        store_type,
        name,
        dry_run=False,
        uid=None,
        email="",
        is_individual=True,
    ):
        self._dry = dry_run
        self._sig_store_token = sig_store_token
        self._sig_store = sig_store
        self._store_branch = store_branch
        self._store_scope = store_scope
        self._store_type = store_type
        self._project = project

        # Contributor's data
        self._name = name
        self._id = uid
        self._email = email
        self._is_individual = is_individual

    def exec(self):
        import os
        import re

        # Check out store
        store = Repository(
            dry_run=self._dry,
            token=self._sig_store_token,
            store_url=self._sig_store,
            store_branch=self._store_branch,
        )

        if self._store_type.casefold() == "csv":
            db = CSVDataBase(
                dry_run=self._dry,
                path=store._sig_store_path,
                project_id=self._project if self._store_scope=="private" else 0,
            )
        elif self._store_type.casefold() == "sql":
            db = SQLDataBase(
                dry_run=self._dry,
                path=store._sig_store_path,
                project_id=self._project if self._store_scope=="private" else 0,
            )
        else:
            raise SystemExit("Did not understand the storage type. Exiting.")

        print("----")

        # Read from commit message
        commit_message = os.environ.get("CI_COMMIT_MESSAGE", "")
        if commit_message == "" and (self._name is None or self._id is None or self._email is None):
            raise SystemExit(
                f"{tc.RED}Please provide all of name, email, and user_id either "
                f"via the command line or in the commit message.{tc.CLR}\n"
                "The commit message should include 'key: value' pairs "
                "on separate lines, e.g.\n"
                "  <commit message>\n"
                "  name: <name>\n"
                "  uid: <UID>\n"
                "  email: <EMAIL>"
            )
        if self._name is None:
            self._name = re.match(
                r"name:\s*(.*)$",
                commit_message,
                re.IGNORECASE,  # fmt: skip
            ).group(1)
        if self._id is None:
            self._id = re.match(
                r"uid:\s*(.*)$",
                commit_message,
                re.IGNORECASE,  # fmt: skip
            ).group(1)
        if self._email is None:
            self._email = re.match(
                r"email:\s*(.*)$",
                commit_message,
                re.IGNORECASE,  # fmt: skip
            ).group(1)

        print("  Adding known contributor")
        print("    name:         ", self._name)
        print("    id:           ", self._id)
        print("    e-mail:       ", self._email)
        print("    is individual:", self._is_individual)

        db.add_signature(self._id, self._name, self._email, self._is_individual, 0)

        print("----")

Bot

Core component of CLA Bot

Source code in src/bot/core.py 
 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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
class Bot:
    """
    Core component of **_CLA Bot_**
    """

    bot_strings = {
        "signature": "I have read the CLA Document and I hereby sign the CLA",
        "ack": " has signed the CLA.",
        "ask": " please sign the CLA! You can find it [here](",
        "not_found": (
            " not found within the GitLab instance. "
            "**Please ensure compliance with the CLA in a separate channel!**"
        ),
        "red_shield": "<img src='https://img.shields.io/badge/CLA-not%20signed-red'>",
        "green_shield": "<img src='https://img.shields.io/badge/CLA-signed-green'>",
    }

    def __init__(
        self,
        force,
        dry_run=False,
        gitlab_workflow=False,
        allow_list="",
        auto_allow="",
        store_branch="",
        store_scope="",
        store_type="",
        mr_iid=None,
        playground=None,
        home=None,
        store=None,
        payload=None,
    ):
        """
        Constructor of the bot.
        """

        self._force = force
        self._dry = dry_run
        self._gitlab_workflow = gitlab_workflow
        self._allow_list = allow_list
        self._auto_allow = auto_allow
        self._store_branch = store_branch
        self._store_scope = store_scope
        self._store_type = store_type
        self._mr_iid = mr_iid
        self._playground = playground
        self._home = home
        self._store = store
        self._payload = payload

        if self._gitlab_workflow:
            self.bot_strings["ask"] += (
                os.environ.get("CLA_URL", "...URL...") + ")."
                "  \nYou can sign by adding a comment saying \n>" + self.bot_strings["signature"]
            )
        else:
            self.bot_strings["ask"] += (
                os.environ.get("CLA_URL", "...URL...") + ")."
                "  \nOnce signed, return it to the maintainers of the project."
            )

        self.bot_strings["ask"] += "\n\n " + self.bot_strings["red_shield"]
        self.bot_strings["ack"] += "\n\n " + self.bot_strings["green_shield"]

    def exec(self):
        """
        Main wrapper to run **_CLA Bot_**.

        It will in turn connect to the different locations ('playground',
        'store', 'home'), read the payload and then check if anything is to do.
        Then iterate (or pick one) over merge requests to check contributors,
        leave notes, close discussions, the lot.
        """
        # Contact GitLab instance - PLAYGROUND
        #  This is where we check for MRs and discussions, also comment
        playground = Project(
            token=self._playground.token,
            project=self._playground.id,
            dry_run=self._dry,
            allow_list=self._allow_list,
            auto_allow=self._auto_allow,
            instance=os.getenv("CI_SERVER_URL"),
            name=self._playground.name,
        )
        # Bail if we caused the triggering event ourselves
        #  If we would save the Bot's id, we could do this much earlier...
        if self._payload.triggered and playground._me == self._payload.trigger_uid:
            raise SystemExit(
                f"Event triggered by uid {self._payload.trigger_uid} "
                f"(that's me.... my uid is {playground._me}). Exiting."
            )
        home = Project(
            token=self._home.token,
            project=self._home.id,
            dry_run=self._dry,
            instance=os.getenv("CI_SERVER_URL"),
            name=self._home.name,
        )

        # Check out store
        store = Repository(
            dry_run=self._dry,
            token=self._store.token,
            store_url=self._store.url,
            store_branch=self._store_branch,
        )

        # Connect to signature store 'database'
        if self._store_type.casefold() == "csv":
            db = CSVDataBase(
                dry_run=self._dry,
                path=store._sig_store_path,
                project_id=playground._project.get_id() if self._store_scope=="private" else 0,
            )
        elif self._store_type.casefold() == "sql" or self._store_type.casefold() == "sqlite":
            db = SQLDataBase(
                dry_run=self._dry,
                path=store._sig_store_path,
                project_id=playground._project.get_id() if self._store_scope=="private" else 0,
            )
        else:
            raise SystemExit("Did not understand the storage type. Exiting.")

        if self._mr_iid is None:
            print("   Iterating over merge requests")
        else:
            print("   Looking for my merge request")

        # There may be multiple triggers, schedules, pushes, anything really,
        # starting a pipeline. To avoid racing for a MR and fixing/checking it
        # simulataneously, wait until we are in fact the pipeline with the
        # lowest id running. Anyone coming after us, will likely do the same as
        # ourselves (and thus, nothing).
        if not self._force:
            home.wait_until_its_our_turn(os.getenv("CI_PIPELINE_ID"))

        print("----")

        # Get list of all open merge requests
        mrs = playground.get_open_mrs()

        for mr in mrs:
            # Run on single MR if asked for it
            # TODO: perhaps change the loop to be for specific MR by searching
            # for it instead of all?
            if self._mr_iid is None or self._mr_iid == mr.iid:
                print(f"    Checking MR: {str(mr.iid)} ({mr.title})")
                print("     Checking for involved users")
                users, names, not_found_names = playground.obtain_involved_users(mr)

                for name in not_found_names:
                    print(f"      {tc.RED}{name} not found!{tc.CLR}")
                    # TODO: Well, technically we could add name+email in the
                    # SQLite file and confirm that, but a GitLab workflow won't work
                    if not playground.commented_before(
                        mr.notes.list(), self.bot_strings["not_found"], name
                    ):
                        playground.start_discussion(mr, self.bot_strings["not_found"], name)

                for name, user in zip(names, users, strict=True):
                    print(f"      {name} ({str(user)}) found")

                    if playground.user_is_in_allow_list(user):
                        print(f"       {tc.GRN}User has a hall pass{tc.CLR} (allow list)")
                        continue
                    if playground.role_is_in_auto_allow(user):
                        print(f"       {tc.GRN}User has a hall pass{tc.CLR} (auto_allow)")
                        continue

                    if db.signature_valid(user):
                        print(f"       {tc.GRN}User has signed CLA{tc.CLR}")
                        if not playground.commented_before(
                            mr.notes.list(), self.bot_strings["ack"], name
                        ):
                            if playground.commented_before(
                                mr.notes.list(), self.bot_strings["ask"], name
                            ):
                                playground.mark_thread_resolved(
                                    mr,
                                    self.bot_strings["ask"],
                                    self.bot_strings["ack"],
                                    name,
                                )
                            else:
                                playground.add_comment(mr, self.bot_strings["ack"], name)
                        continue

                    else:
                        if self._gitlab_workflow:
                            # GITLAB WORKFLOW ##########################################
                            print("     Checking comments")
                            CLA_signed_in_note = playground.CLA_is_signed(
                                mr.notes.list(), user, self.bot_strings["signature"]
                            )
                            if CLA_signed_in_note:
                                # Save copy of MR text and all notes
                                if not os.path.exists(
                                    store._sig_store_path + "/" + str(playground._project.get_id())
                                ):
                                    os.mkdir(
                                        store._sig_store_path
                                        + "/"
                                        + str(playground._project.get_id())
                                    )
                                snappy_path = (
                                    store._sig_store_path
                                    + "/"
                                    + str(playground._project.get_id())
                                    + "/"
                                    + "UID_"
                                    + str(user)
                                    + "_MR_"
                                    + str(mr.iid)
                                    + ".sig"
                                )
                                with open(snappy_path, "w") as snappy:
                                    snappy.write(mr.to_json())
                                    for note in mr.notes.list():
                                        snappy.write(note.to_json())

                                # Add comment acknowleding the signature
                                url = mr.web_url + "#note_" + str(CLA_signed_in_note)
                                if not playground.commented_before(
                                    mr.notes.list(), self.bot_strings["ack"], name
                                ):
                                    if playground.commented_before(
                                        mr.notes.list(), self.bot_strings["ask"], name
                                    ):
                                        playground.mark_thread_resolved(
                                            mr,
                                            self.bot_strings["ask"],
                                            self.bot_strings["ack"],
                                            name,
                                            url,
                                        )
                                    else:
                                        playground.add_comment(
                                            mr, self.bot_strings["ack"], name, url
                                        )

                                # Add signature to db
                                db.add_signature(user, name, "", True, mr.iid)
                            else:
                                if not playground.commented_before(
                                    mr.notes.list(), self.bot_strings["ask"], name
                                ):
                                    print(f"       {tc.YLW}Poking user{tc.CLR}")
                                    playground.start_discussion(mr, self.bot_strings["ask"], name)
                        else:
                            # PAPER TRAIL ##############################################
                            if not playground.commented_before(
                                mr.notes.list(), self.bot_strings["ask"], name
                            ):
                                print(f"       {tc.YLW}Poking user{tc.CLR}")
                                playground.start_discussion(mr, self.bot_strings["ask"], name)

        print("----")

__init__(force, dry_run=False, gitlab_workflow=False, allow_list='', auto_allow='', store_branch='', store_scope='', store_type='', mr_iid=None, playground=None, home=None, store=None, payload=None)

Constructor of the bot.

Source code in src/bot/core.py 
 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
def __init__(
    self,
    force,
    dry_run=False,
    gitlab_workflow=False,
    allow_list="",
    auto_allow="",
    store_branch="",
    store_scope="",
    store_type="",
    mr_iid=None,
    playground=None,
    home=None,
    store=None,
    payload=None,
):
    """
    Constructor of the bot.
    """

    self._force = force
    self._dry = dry_run
    self._gitlab_workflow = gitlab_workflow
    self._allow_list = allow_list
    self._auto_allow = auto_allow
    self._store_branch = store_branch
    self._store_scope = store_scope
    self._store_type = store_type
    self._mr_iid = mr_iid
    self._playground = playground
    self._home = home
    self._store = store
    self._payload = payload

    if self._gitlab_workflow:
        self.bot_strings["ask"] += (
            os.environ.get("CLA_URL", "...URL...") + ")."
            "  \nYou can sign by adding a comment saying \n>" + self.bot_strings["signature"]
        )
    else:
        self.bot_strings["ask"] += (
            os.environ.get("CLA_URL", "...URL...") + ")."
            "  \nOnce signed, return it to the maintainers of the project."
        )

    self.bot_strings["ask"] += "\n\n " + self.bot_strings["red_shield"]
    self.bot_strings["ack"] += "\n\n " + self.bot_strings["green_shield"]

exec()

Main wrapper to run CLA Bot.

It will in turn connect to the different locations ('playground', 'store', 'home'), read the payload and then check if anything is to do. Then iterate (or pick one) over merge requests to check contributors, leave notes, close discussions, the lot.

Source code in src/bot/core.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def exec(self):
    """
    Main wrapper to run **_CLA Bot_**.

    It will in turn connect to the different locations ('playground',
    'store', 'home'), read the payload and then check if anything is to do.
    Then iterate (or pick one) over merge requests to check contributors,
    leave notes, close discussions, the lot.
    """
    # Contact GitLab instance - PLAYGROUND
    #  This is where we check for MRs and discussions, also comment
    playground = Project(
        token=self._playground.token,
        project=self._playground.id,
        dry_run=self._dry,
        allow_list=self._allow_list,
        auto_allow=self._auto_allow,
        instance=os.getenv("CI_SERVER_URL"),
        name=self._playground.name,
    )
    # Bail if we caused the triggering event ourselves
    #  If we would save the Bot's id, we could do this much earlier...
    if self._payload.triggered and playground._me == self._payload.trigger_uid:
        raise SystemExit(
            f"Event triggered by uid {self._payload.trigger_uid} "
            f"(that's me.... my uid is {playground._me}). Exiting."
        )
    home = Project(
        token=self._home.token,
        project=self._home.id,
        dry_run=self._dry,
        instance=os.getenv("CI_SERVER_URL"),
        name=self._home.name,
    )

    # Check out store
    store = Repository(
        dry_run=self._dry,
        token=self._store.token,
        store_url=self._store.url,
        store_branch=self._store_branch,
    )

    # Connect to signature store 'database'
    if self._store_type.casefold() == "csv":
        db = CSVDataBase(
            dry_run=self._dry,
            path=store._sig_store_path,
            project_id=playground._project.get_id() if self._store_scope=="private" else 0,
        )
    elif self._store_type.casefold() == "sql" or self._store_type.casefold() == "sqlite":
        db = SQLDataBase(
            dry_run=self._dry,
            path=store._sig_store_path,
            project_id=playground._project.get_id() if self._store_scope=="private" else 0,
        )
    else:
        raise SystemExit("Did not understand the storage type. Exiting.")

    if self._mr_iid is None:
        print("   Iterating over merge requests")
    else:
        print("   Looking for my merge request")

    # There may be multiple triggers, schedules, pushes, anything really,
    # starting a pipeline. To avoid racing for a MR and fixing/checking it
    # simulataneously, wait until we are in fact the pipeline with the
    # lowest id running. Anyone coming after us, will likely do the same as
    # ourselves (and thus, nothing).
    if not self._force:
        home.wait_until_its_our_turn(os.getenv("CI_PIPELINE_ID"))

    print("----")

    # Get list of all open merge requests
    mrs = playground.get_open_mrs()

    for mr in mrs:
        # Run on single MR if asked for it
        # TODO: perhaps change the loop to be for specific MR by searching
        # for it instead of all?
        if self._mr_iid is None or self._mr_iid == mr.iid:
            print(f"    Checking MR: {str(mr.iid)} ({mr.title})")
            print("     Checking for involved users")
            users, names, not_found_names = playground.obtain_involved_users(mr)

            for name in not_found_names:
                print(f"      {tc.RED}{name} not found!{tc.CLR}")
                # TODO: Well, technically we could add name+email in the
                # SQLite file and confirm that, but a GitLab workflow won't work
                if not playground.commented_before(
                    mr.notes.list(), self.bot_strings["not_found"], name
                ):
                    playground.start_discussion(mr, self.bot_strings["not_found"], name)

            for name, user in zip(names, users, strict=True):
                print(f"      {name} ({str(user)}) found")

                if playground.user_is_in_allow_list(user):
                    print(f"       {tc.GRN}User has a hall pass{tc.CLR} (allow list)")
                    continue
                if playground.role_is_in_auto_allow(user):
                    print(f"       {tc.GRN}User has a hall pass{tc.CLR} (auto_allow)")
                    continue

                if db.signature_valid(user):
                    print(f"       {tc.GRN}User has signed CLA{tc.CLR}")
                    if not playground.commented_before(
                        mr.notes.list(), self.bot_strings["ack"], name
                    ):
                        if playground.commented_before(
                            mr.notes.list(), self.bot_strings["ask"], name
                        ):
                            playground.mark_thread_resolved(
                                mr,
                                self.bot_strings["ask"],
                                self.bot_strings["ack"],
                                name,
                            )
                        else:
                            playground.add_comment(mr, self.bot_strings["ack"], name)
                    continue

                else:
                    if self._gitlab_workflow:
                        # GITLAB WORKFLOW ##########################################
                        print("     Checking comments")
                        CLA_signed_in_note = playground.CLA_is_signed(
                            mr.notes.list(), user, self.bot_strings["signature"]
                        )
                        if CLA_signed_in_note:
                            # Save copy of MR text and all notes
                            if not os.path.exists(
                                store._sig_store_path + "/" + str(playground._project.get_id())
                            ):
                                os.mkdir(
                                    store._sig_store_path
                                    + "/"
                                    + str(playground._project.get_id())
                                )
                            snappy_path = (
                                store._sig_store_path
                                + "/"
                                + str(playground._project.get_id())
                                + "/"
                                + "UID_"
                                + str(user)
                                + "_MR_"
                                + str(mr.iid)
                                + ".sig"
                            )
                            with open(snappy_path, "w") as snappy:
                                snappy.write(mr.to_json())
                                for note in mr.notes.list():
                                    snappy.write(note.to_json())

                            # Add comment acknowleding the signature
                            url = mr.web_url + "#note_" + str(CLA_signed_in_note)
                            if not playground.commented_before(
                                mr.notes.list(), self.bot_strings["ack"], name
                            ):
                                if playground.commented_before(
                                    mr.notes.list(), self.bot_strings["ask"], name
                                ):
                                    playground.mark_thread_resolved(
                                        mr,
                                        self.bot_strings["ask"],
                                        self.bot_strings["ack"],
                                        name,
                                        url,
                                    )
                                else:
                                    playground.add_comment(
                                        mr, self.bot_strings["ack"], name, url
                                    )

                            # Add signature to db
                            db.add_signature(user, name, "", True, mr.iid)
                        else:
                            if not playground.commented_before(
                                mr.notes.list(), self.bot_strings["ask"], name
                            ):
                                print(f"       {tc.YLW}Poking user{tc.CLR}")
                                playground.start_discussion(mr, self.bot_strings["ask"], name)
                    else:
                        # PAPER TRAIL ##############################################
                        if not playground.commented_before(
                            mr.notes.list(), self.bot_strings["ask"], name
                        ):
                            print(f"       {tc.YLW}Poking user{tc.CLR}")
                            playground.start_discussion(mr, self.bot_strings["ask"], name)

    print("----")

Payload

A class to hold 'payload' data from webhooks/triggers. This is used as configuration/input for CLA Bot.

The parameters extracted from the payload and stored for later use include
  • a flag if the pipeline was actually triggered (we could be using a command line to force interaction)
  • the project id (for the 'playground')
  • the triggering event type (comments, MRs, commits, ...)
  • the MR id (actually the project internal id)
  • the user id of the user causing the event (might be the bot itself)
Source code in src/bot/core.py 
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
class Payload:
    """
    A class to hold 'payload' data from webhooks/triggers. This is used as
    configuration/input for **_CLA Bot_**.

    The parameters extracted from the payload and stored for later use include:
        - a flag if the pipeline was actually triggered (we could be using a
          command line to force interaction)
        - the project id (for the 'playground')
        - the triggering event type (comments, MRs, commits, ...)
        - the MR id (actually the project internal id)
        - the user id of the user causing the event (might be the bot itself)
    """

    def __init__(self):
        self.triggered = False
        self.project_id = None
        self.trigger_event = None
        self.trigger_mr_iid = None
        self.trigger_uid = None

        if os.environ.get("CI_PIPELINE_SOURCE", "").casefold() == "trigger":
            self.triggered = True
            # Get payload file
            try:
                trigger_payload_file = os.environ["TRIGGER_PAYLOAD"]
            except KeyError:
                raise SystemExit(f"{tc.RED}ENV[TRIGGER_PAYLOAD] not found!{tc.CLR}") from None

            # Open file
            with open(trigger_payload_file) as trigger_payload_filep:
                # Read json blob from file
                trigger_payload = json.load(trigger_payload_filep)

            # Get project id event originated from
            self.project_id = trigger_payload["project"]["id"]

            # The trigger can be from two event types:
            # comments (issues/MRs) or merge requests

            # Merge request event, find MR that triggered the bot
            if trigger_payload["event_type"].casefold() == "merge_request":
                self.trigger_event = "MR"
                self.trigger_mr_iid = trigger_payload["object_attributes"]["iid"]
                self.trigger_uid = trigger_payload["object_attributes"]["updated_by_id"]

            # Comment event, find comment type
            if trigger_payload["event_type"].casefold() == "note":
                if (
                    trigger_payload["object_attributes"]["noteable_type"].casefold()
                    == "MergeRequest".casefold()
                ):
                    self.trigger_event = "MR note"
                    self.trigger_mr_iid = trigger_payload["merge_request"]["iid"]
                else:
                    self.trigger_event = "issue note"
                self.trigger_uid = trigger_payload["user"]["id"]

Interacting with GitLab and git

src.bot.project

Project

Interact with GitLab and its projects

This class wraps all python-gitlab calls, combining them as suitable and adding output. If called in 'dry mode', no changes will be made to the project, instead only comments written to stdout.

Methods:

Name Description
connect_to_project

mandatory connection to GitLab, used for 'playground' and 'home'
string_match

compare (sub)strings, ignoring whitespace and case
check_avatar

check the bot's avatar, setting it
convert_names_to_ids

convert list of usernames to user ids
convert_roles_to_ids

convert access role to numeric id
obtain_involved_users

collect contributors for an MR
add_comment

add comments to issues, MRs, discussions
start_discussion

start discussion thread (unresolved)
wait_until_its_our_turn

wait out turn in current CI pipeline
get_open_issues

get all open issues
get_open_mrs

get all open MRs
commented_before

check if comment with same string already exists
user_is_in_allow_list

compare user to allow list
role_is_in_auto_allow

compare role to allow list
CLA_is_signed

check list of notes for signature string
mark_thread_resolved

resolve discussion thread
Source code in src/bot/project.py 
 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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
class Project:
    """
    Interact with GitLab and its projects

    This class wraps all `python-gitlab` calls, combining them as suitable and
    adding output. If called in 'dry mode', no changes will be made to the
    project, instead only comments written to stdout.

    Methods:
        connect_to_project: mandatory connection to GitLab, used for 'playground' and 'home'
        string_match: compare (sub)strings, ignoring whitespace and case
        check_avatar: check the bot's avatar, setting it
        convert_names_to_ids: convert list of usernames to user ids
        convert_roles_to_ids: convert access role to numeric id
        obtain_involved_users: collect contributors for an MR
        add_comment: add comments to issues, MRs, discussions
        start_discussion: start discussion thread (unresolved)
        wait_until_its_our_turn: wait out turn in current CI pipeline
        get_open_issues: get all open issues
        get_open_mrs: get all open MRs
        commented_before: check if comment with same string already exists
        user_is_in_allow_list: compare user to allow list
        role_is_in_auto_allow: compare role to allow list
        CLA_is_signed: check list of notes for signature string
        mark_thread_resolved: resolve discussion thread
    """

    def __init__(
        self,
        token=None,
        project=None,
        dry_run=False,
        allow_list="",
        auto_allow="",
        instance="",
        name="",
    ):
        """
        Constructor for Project

        Parameters:
            token (str): access token for project
            project (int): project's id
            dry_run (bool): toggle dry run, changing print mainly statements.
            allow_list (strings): list of allowed usernames
            auto_allow (int): numeric role that is allowed
            instance (str): URL of GitLab server to connect to
            name (str): identifier of project to connect to
        """
        # Do we run in dry-run mode?
        self._dry = dry_run
        # Store allow list from CDL
        self._allow_list = allow_list
        # Store project role to auto allow
        self._auto_allow = auto_allow

        if not instance:
            sys.exit("No GitLab instance provided!")

        # Get an instance of the project
        self._project, self._me, self._gl = self.connect_to_project(instance, token, project, name)

        # If allow_list is supplied on command line, convert names to ids
        if allow_list:
            self._allow_ids = self.convert_names_to_ids(allow_list)

    def connect_to_project(self, instance, token, project_id, name=""):
        """
        Connect to the project

        Using the supplied token, this will connect to the project on the given GitLab
        instance. Can work with project access tokens or CI_JOB_TOKEN alike, the latter
        not allowing to query a user.
        Optionally include an indentifier for the project to connect to.

        Parameters:
            instance (str): URL of GitLab instance to connect to
            token (str): access token for project
            project_id (int): project id to connect to
            name (str): identifier of project to connect to

        Returns:
            project: connected GitLab project
            gluser: current user, connected to the project
            gl: connection to GitLab
        """

        name = " " if not name else " " + name + " "
        dry = "[dry] " if self._dry else ""
        print(f"{dry}Connecting to{name}project on {instance}")

        if not token:
            sys.exit(
                "Token for GitLab authentication required! "
                "Please provide it via CI variable or parameter."
            )

        # Connect to our GitLab instance
        try:
            gl = gitlab.Gitlab(instance, private_token=token)
        except Exception as e:
            print(f" {tc.RED}Could not contact the GitLab instance {instance}.{tc.CLR}")
            print(f" Error: {e}")
            sys.exit()

        # Get 'our' project
        try:
            project = gl.projects.get(project_id)
        except Exception as e:
            print(f" {tc.RED}Could not connect to the project {project_id}.{tc.CLR}")
            print(f" Error: {e}")
            sys.exit()

        print(
            f" {tc.YLW}Connected to project '{project.name}' located at "
            f"{project.namespace['web_url']}{tc.CLR}"
        )

        gl.auth()
        return project, gl.user, gl

    def string_match(self, string_to_search, string_to_match):
        """
        Check if one string is a substring anywhere within the other.
        Ignores any spaces and newlines, also ignores case.

        Parameters:
            string_to_search (str): string to search within
            string_to_match (str): string to search for

        Returns:
            flag (bool): true when search string found
        """
        searched_string = string_to_search.casefold().translate(
            {ord(c): None for c in string.whitespace}
        )
        search_string = string_to_match.casefold().translate(
            {ord(c): None for c in string.whitespace}
        )

        # Comparison should include 0 since the find could be at the start of the string
        return searched_string.find(search_string) >= 0

    def check_avatar(self, gl):
        """
        Check our avatar, set correct one if necessary
        """

        user = gl.users.get(gl.user.id)
        with open("bot/botalicious-avatar.jpg", "rb") as avatar_image:
            user.avatar = avatar_image
        user.save()

    def convert_names_to_ids(self, allow_list):
        """
        Convert list of usernames to list of ids

        Parameters:
            allow_list (str): csv list of user names on GitLab

        Returns:
            id_list (int): list of matching user ids on GitLab
        """

        id_list = []

        try:
            for user in allow_list.split(","):
                id_list.append(self._gl.users.list(username=user)[0].id)
        except Exception as e:
            print(f" {tc.RED}An error occurred reading the list of allowed usernames: {e}{tc.CLR}")
            print(f" This list was provided: __{allow_list}__")

        return id_list

    def convert_roles_to_ids(self, auto_allow):
        """
        Convert list of roles to list of access levels

        Parameters:
            auto_allow (str): list of project roles

        Returns:
            lvl_list (): list of matching access levels

        """

        lvl_list = []

        for role in auto_allow.split(","):
            if role == "Developer":
                lvl_list.append(30)
            if role == "Maintainer":
                lvl_list.append(40)
            if role == "Owner":
                lvl_list.append(50)
            if role == "Allmighty":
                lvl_list.append(99)

        return lvl_list

    def obtain_involved_users(self, mr):
        """
        Return list of user ids that show up in MR along with their names.
        At a minimum, it will return MR author.

        Names that cannot be resolved (or are not unique)
        on the GitLab instance will be listed.

        Parameters:
            mr (): merge request to check

        Returns:
            users (int): list of user id
            names (str): list of user names
            not_found (str): list of names not found
        """

        # MR author is at least one
        users = [mr.author["id"]]
        names = [mr.author["name"]]
        not_found = []
        # Cycle all involved commits and extract authors/committers
        for commit in mr.commits():
            for name in [
                commit.attributes["author_name"],
                commit.attributes["committer_name"],
            ]:
                user = self._gl.users.list(search=name)
                if len(user) == 1:
                    if user[0].id not in users:
                        users.append(user[0].id)
                        names.append(name)
                else:
                    not_found.append(name)

        return users, names, not_found

    def add_comment(self, target, comment, name=None, url=None):
        """
        Add a comment to an issue, merge request, or discussion

        Parameters:
            target (): object to comment on
            comment (str): comment to leave
            name (str): user to address (optional)
            url (str): URL to refer to (optional)

        Returns:
            flag (bool): true if comment added, false if wrong note type
        """

        what_we_comment_on = ""
        if isinstance(target, gitlab.v4.objects.issues.ProjectIssue):
            what_we_comment_on = "issue"
        if isinstance(target, gitlab.v4.objects.merge_requests.ProjectMergeRequest):
            what_we_comment_on = "MR"
        if isinstance(target, gitlab.v4.objects.discussions.ProjectMergeRequestDiscussion):
            what_we_comment_on = "discussion"

        if not what_we_comment_on:
            print(f"{tc.RED}Comment misguided{tc.CLR}")
            return False

        if self._dry:
            print(
                f"{tc.YLW}[dry]  Adding comment to {what_we_comment_on}: {str(target.id)}{tc.CLR}"
            )
            print(f"       {name + comment}", url)
        else:
            print(
                f"       {tc.YLW}Adding comment to {what_we_comment_on}: {str(target.id)}{tc.CLR}"
            )
            if name is not None:
                print("       Addressing " + name)
                if url is not None:
                    target.notes.create({"body": name + comment + " (see [here](" + url + "))"})
                else:
                    target.notes.create({"body": name + comment})
            else:
                if url is not None:
                    target.notes.create({"body": comment + " (see [here](" + url + "))"})
                else:
                    target.notes.create({"body": comment})

        return True

    def start_discussion(self, target, comment, name=None):
        """
        Start a discussion within a merge request, adding a comment

        This will leave an unresolved thread in the process

        Parameters:
            target (): merge request to comment on
            comment (str): comment to leave
            name (str): user to address (optional)

        Returns:
            flag (bool): true if comment added, false if wrong note type
        """

        what_we_comment_on = ""
        if isinstance(target, gitlab.v4.objects.merge_requests.ProjectMergeRequest):
            what_we_comment_on = "MR"

        if not what_we_comment_on:
            print(f"{tc.RED}Discussion misguided{tc.CLR}")
            return False

        if self._dry:
            print(
                f"{tc.YLW}[dry]  Adding discussion to {what_we_comment_on}: "
                f"{str(target.id)}{tc.CLR}"
            )
            print(f"       {name + comment}")
        else:
            print(
                f"       {tc.YLW}Adding comment to {what_we_comment_on}: {str(target.id)}{tc.CLR}"
            )
            if name is not None:
                print("       Addressing " + name)
                target.discussions.create({"body": name + comment})
            else:
                target.discussions.create({"body": comment})

        return True

    def wait_until_its_our_turn(self, current_pipeline_id=None):
        """
        Wait until our pipeline is the one created first out of the running ones.

        Parameters:
            current_pipeline_id (str): active pipeline id, string taken from ENV

        """

        if current_pipeline_id is None:
            sys.exit(f"{tc.RED}Smells like a rotten pipeline - not running in a pipeline?{tc.CLR}")

        # Get first of all running pipelines and wait until this is us (inside HOME)
        while self._project.pipelines.list(status="running", get_all=True, sort="asc")[0].id != int(
            current_pipeline_id
        ):
            print("Paitiently waiting our turn...", flush=True)
            sleep(5)

        return None

    def get_open_issues(self):
        """
        Get all (no pagination) _open_ issues of the project

        Returns:
            issues (iterator): list of issues
        """

        print(" Checking project for open issues...")
        issues = self._project.issues.list(state="opened", get_all=True)
        print(f"  Found {str(len(issues))} issues")

        return issues

    def get_open_mrs(self):
        """
        Get all (no pagination) _open_ merge request of the project, earliest first

        Returns:
            mrs (iterator): list of merge requests
        """

        print(" Checking project for open merge requests")
        mrs = self._project.mergerequests.list(state="opened", get_all=True, sort="asc")
        print(f"  Found {str(len(mrs))} merge requests")

        return mrs

    def commented_before(self, notes, search_str, name=None):
        """
        Check if similar comment was already made

        Parameters:
            notes (): list of notes to check
            search_str (str): string to look for as comment
            name (str): user to address (optional)

        Returns:
            ack (bool): true if note with search_str already added
        """

        ack = False

        if name is not None:
            search_str = name + search_str

        for note in notes:
            # Only look at notes from the bot
            # Check if the magic string appears
            if note.author["id"] == self._me.id and self.string_match(note.body, search_str):
                print("       Found earlier comment")
                ack = True

        return ack

    def user_is_in_allow_list(self, user):
        """
        Check if user is in allow list.
        Will issue a free hall pass.

        Parameters:
            user (int): numeric user id

        Returns:
            hall_pass (bool): true if user is allowed
        """

        hall_pass = False

        if self._allow_list:
            for member in self._allow_ids:
                if user == member:
                    print(f"       {tc.GRN}Contributor is in allow list{tc.CLR}")
                    hall_pass = True

        return hall_pass

    def role_is_in_auto_allow(self, user):
        """
        Check if user is project member and has required role.
        Will issue a free hall pass if allowed.

        Parameters:
            user (int): numeric user id

        Returns:
            hall_pass (bool): true if role is allowed
        """

        hall_pass = False

        if self._auto_allow:
            minimum_level = 60  # is that an admin?
            # GitLab maps roles to a numeric access level, so do we
            if self._auto_allow == "Developer":
                minimum_level = 30
            elif self._auto_allow == "Maintainer":
                minimum_level = 40
            elif self._auto_allow == "Owner":
                minimum_level = 50
            elif self._auto_allow == "Allmighty":
                # We add this level for testing purposes, so a bot developer
                # may not automatically be approved
                minimum_level = 99
            else:
                print(
                    f"       {tc.RED}Role not known, please provide either "
                    f"'Developer', 'Maintainer', or 'Owner', skipping role check{tc.CLR}"
                )
                minimum_level = 666

            for member in self._project.members_all.list():
                member_role = member.attributes["access_level"]
                if user == member.get_id() and member_role >= minimum_level:
                    print(
                        "       Contributor is project member "
                        f"with matching permissions ({str(member_role)})"
                    )
                    hall_pass = True

        return hall_pass

    def CLA_is_signed(self, notes, user, signature_string):
        """
        Check if a list of notes contains the 'signature string'
        added by a specific user.

        Parameters:
            notes (): list of notes to check
            user (int): user id to look for
            signature_string (str): string that qualifies as signature

        Returns:
            signed_in_note (int): numeric id of note containing signature string
        """

        # return 0 aka False by default or the matching note id
        signed_in_note = 0

        for note in notes:
            # Only look at notes from the user
            # Check if the magic string appears
            if note.author["id"] == user and self.string_match(note.body, signature_string):
                print(f"      {tc.GRN}Signature found!{tc.CLR}")
                signed_in_note = note.id

        return signed_in_note

    def mark_thread_resolved(self, mr, search_str, ack_str=None, name=None, url=None):
        """
        Mark a thread as resolved

        Will look for a thread started by the bot, including the correct search string,
        then resolve it with a matching comment.

        Parameters:
            mr (): merge request to work one
            search_str (str): string to look for in discussion
            ack_str (str): string to respond with (optional)
            name (str): user to address (optional)
            url (str): URL to the signature (optional)
        """

        if name is not None:
            search_str = name + search_str

        # Find discussion thread
        for discussion in mr.discussions.list():
            # Scan notes
            for note in discussion.attributes["notes"]:
                # Look for our notes
                # Look for us asking to sign
                if note["author"]["id"] == self._me.id and self.string_match(
                    note["body"], search_str
                ):
                    # Add a note if we want one
                    if ack_str is not None:
                        self.add_comment(discussion, ack_str, name, url)
                    # Mark discussion thread resolved
                    if note["resolved"] is not True:
                        print(f"       {tc.GRN}Marking thread as resolved{tc.CLR}")
                        discussion.resolved = True
                        discussion.save()

CLA_is_signed(notes, user, signature_string)

Check if a list of notes contains the 'signature string' added by a specific user.

Parameters:

Name Type Description Default
notes

list of notes to check

 required
user int

user id to look for

 required
signature_string str

string that qualifies as signature

 required

Returns:

Name Type Description
signed_in_note int

numeric id of note containing signature string
Source code in src/bot/project.py 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
def CLA_is_signed(self, notes, user, signature_string):
    """
    Check if a list of notes contains the 'signature string'
    added by a specific user.

    Parameters:
        notes (): list of notes to check
        user (int): user id to look for
        signature_string (str): string that qualifies as signature

    Returns:
        signed_in_note (int): numeric id of note containing signature string
    """

    # return 0 aka False by default or the matching note id
    signed_in_note = 0

    for note in notes:
        # Only look at notes from the user
        # Check if the magic string appears
        if note.author["id"] == user and self.string_match(note.body, signature_string):
            print(f"      {tc.GRN}Signature found!{tc.CLR}")
            signed_in_note = note.id

    return signed_in_note

__init__(token=None, project=None, dry_run=False, allow_list='', auto_allow='', instance='', name='')

Constructor for Project

Parameters:

Name Type Description Default
token str

access token for project

 None
project int

project's id

 None
dry_run bool

toggle dry run, changing print mainly statements.

 False
allow_list strings

list of allowed usernames

 ''
auto_allow int

numeric role that is allowed

 ''
instance str

URL of GitLab server to connect to

 ''
name str

identifier of project to connect to

 ''
Source code in src/bot/project.py 
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
def __init__(
    self,
    token=None,
    project=None,
    dry_run=False,
    allow_list="",
    auto_allow="",
    instance="",
    name="",
):
    """
    Constructor for Project

    Parameters:
        token (str): access token for project
        project (int): project's id
        dry_run (bool): toggle dry run, changing print mainly statements.
        allow_list (strings): list of allowed usernames
        auto_allow (int): numeric role that is allowed
        instance (str): URL of GitLab server to connect to
        name (str): identifier of project to connect to
    """
    # Do we run in dry-run mode?
    self._dry = dry_run
    # Store allow list from CDL
    self._allow_list = allow_list
    # Store project role to auto allow
    self._auto_allow = auto_allow

    if not instance:
        sys.exit("No GitLab instance provided!")

    # Get an instance of the project
    self._project, self._me, self._gl = self.connect_to_project(instance, token, project, name)

    # If allow_list is supplied on command line, convert names to ids
    if allow_list:
        self._allow_ids = self.convert_names_to_ids(allow_list)

add_comment(target, comment, name=None, url=None)

Add a comment to an issue, merge request, or discussion

Parameters:

Name Type Description Default
target

object to comment on

 required
comment str

comment to leave

 required
name str

user to address (optional)

 None
url str

URL to refer to (optional)

 None

Returns:

Name Type Description
flag bool

true if comment added, false if wrong note type
Source code in src/bot/project.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def add_comment(self, target, comment, name=None, url=None):
    """
    Add a comment to an issue, merge request, or discussion

    Parameters:
        target (): object to comment on
        comment (str): comment to leave
        name (str): user to address (optional)
        url (str): URL to refer to (optional)

    Returns:
        flag (bool): true if comment added, false if wrong note type
    """

    what_we_comment_on = ""
    if isinstance(target, gitlab.v4.objects.issues.ProjectIssue):
        what_we_comment_on = "issue"
    if isinstance(target, gitlab.v4.objects.merge_requests.ProjectMergeRequest):
        what_we_comment_on = "MR"
    if isinstance(target, gitlab.v4.objects.discussions.ProjectMergeRequestDiscussion):
        what_we_comment_on = "discussion"

    if not what_we_comment_on:
        print(f"{tc.RED}Comment misguided{tc.CLR}")
        return False

    if self._dry:
        print(
            f"{tc.YLW}[dry]  Adding comment to {what_we_comment_on}: {str(target.id)}{tc.CLR}"
        )
        print(f"       {name + comment}", url)
    else:
        print(
            f"       {tc.YLW}Adding comment to {what_we_comment_on}: {str(target.id)}{tc.CLR}"
        )
        if name is not None:
            print("       Addressing " + name)
            if url is not None:
                target.notes.create({"body": name + comment + " (see [here](" + url + "))"})
            else:
                target.notes.create({"body": name + comment})
        else:
            if url is not None:
                target.notes.create({"body": comment + " (see [here](" + url + "))"})
            else:
                target.notes.create({"body": comment})

    return True

check_avatar(gl)

Check our avatar, set correct one if necessary

Source code in src/bot/project.py 
176
177
178
179
180
181
182
183
184
def check_avatar(self, gl):
    """
    Check our avatar, set correct one if necessary
    """

    user = gl.users.get(gl.user.id)
    with open("bot/botalicious-avatar.jpg", "rb") as avatar_image:
        user.avatar = avatar_image
    user.save()

commented_before(notes, search_str, name=None)

Check if similar comment was already made

Parameters:

Name Type Description Default
notes

list of notes to check

 required
search_str str

string to look for as comment

 required
name str

user to address (optional)

 None

Returns:

Name Type Description
ack bool

true if note with search_str already added
Source code in src/bot/project.py 
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
def commented_before(self, notes, search_str, name=None):
    """
    Check if similar comment was already made

    Parameters:
        notes (): list of notes to check
        search_str (str): string to look for as comment
        name (str): user to address (optional)

    Returns:
        ack (bool): true if note with search_str already added
    """

    ack = False

    if name is not None:
        search_str = name + search_str

    for note in notes:
        # Only look at notes from the bot
        # Check if the magic string appears
        if note.author["id"] == self._me.id and self.string_match(note.body, search_str):
            print("       Found earlier comment")
            ack = True

    return ack

connect_to_project(instance, token, project_id, name='')

Connect to the project

Using the supplied token, this will connect to the project on the given GitLab instance. Can work with project access tokens or CI_JOB_TOKEN alike, the latter not allowing to query a user. Optionally include an indentifier for the project to connect to.

Parameters:

Name Type Description Default
instance str

URL of GitLab instance to connect to

 required
token str

access token for project

 required
project_id int

project id to connect to

 required
name str

identifier of project to connect to

 ''

Returns:

Name Type Description
project

connected GitLab project
gluser

current user, connected to the project
gl

connection to GitLab
Source code in src/bot/project.py 
 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
151
152
def connect_to_project(self, instance, token, project_id, name=""):
    """
    Connect to the project

    Using the supplied token, this will connect to the project on the given GitLab
    instance. Can work with project access tokens or CI_JOB_TOKEN alike, the latter
    not allowing to query a user.
    Optionally include an indentifier for the project to connect to.

    Parameters:
        instance (str): URL of GitLab instance to connect to
        token (str): access token for project
        project_id (int): project id to connect to
        name (str): identifier of project to connect to

    Returns:
        project: connected GitLab project
        gluser: current user, connected to the project
        gl: connection to GitLab
    """

    name = " " if not name else " " + name + " "
    dry = "[dry] " if self._dry else ""
    print(f"{dry}Connecting to{name}project on {instance}")

    if not token:
        sys.exit(
            "Token for GitLab authentication required! "
            "Please provide it via CI variable or parameter."
        )

    # Connect to our GitLab instance
    try:
        gl = gitlab.Gitlab(instance, private_token=token)
    except Exception as e:
        print(f" {tc.RED}Could not contact the GitLab instance {instance}.{tc.CLR}")
        print(f" Error: {e}")
        sys.exit()

    # Get 'our' project
    try:
        project = gl.projects.get(project_id)
    except Exception as e:
        print(f" {tc.RED}Could not connect to the project {project_id}.{tc.CLR}")
        print(f" Error: {e}")
        sys.exit()

    print(
        f" {tc.YLW}Connected to project '{project.name}' located at "
        f"{project.namespace['web_url']}{tc.CLR}"
    )

    gl.auth()
    return project, gl.user, gl

convert_names_to_ids(allow_list)

Convert list of usernames to list of ids

Parameters:

Name Type Description Default
allow_list str

csv list of user names on GitLab

 required

Returns:

Name Type Description
id_list int

list of matching user ids on GitLab
Source code in src/bot/project.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def convert_names_to_ids(self, allow_list):
    """
    Convert list of usernames to list of ids

    Parameters:
        allow_list (str): csv list of user names on GitLab

    Returns:
        id_list (int): list of matching user ids on GitLab
    """

    id_list = []

    try:
        for user in allow_list.split(","):
            id_list.append(self._gl.users.list(username=user)[0].id)
    except Exception as e:
        print(f" {tc.RED}An error occurred reading the list of allowed usernames: {e}{tc.CLR}")
        print(f" This list was provided: __{allow_list}__")

    return id_list

convert_roles_to_ids(auto_allow)

Convert list of roles to list of access levels

Parameters:

Name Type Description Default
auto_allow str

list of project roles

 required

Returns:

Type Description

lvl_list (): list of matching access levels
Source code in src/bot/project.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def convert_roles_to_ids(self, auto_allow):
    """
    Convert list of roles to list of access levels

    Parameters:
        auto_allow (str): list of project roles

    Returns:
        lvl_list (): list of matching access levels

    """

    lvl_list = []

    for role in auto_allow.split(","):
        if role == "Developer":
            lvl_list.append(30)
        if role == "Maintainer":
            lvl_list.append(40)
        if role == "Owner":
            lvl_list.append(50)
        if role == "Allmighty":
            lvl_list.append(99)

    return lvl_list

get_open_issues()

Get all (no pagination) open issues of the project

Returns:

Name Type Description
issues iterator

list of issues
Source code in src/bot/project.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
def get_open_issues(self):
    """
    Get all (no pagination) _open_ issues of the project

    Returns:
        issues (iterator): list of issues
    """

    print(" Checking project for open issues...")
    issues = self._project.issues.list(state="opened", get_all=True)
    print(f"  Found {str(len(issues))} issues")

    return issues

get_open_mrs()

Get all (no pagination) open merge request of the project, earliest first

Returns:

Name Type Description
mrs iterator

list of merge requests
Source code in src/bot/project.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
def get_open_mrs(self):
    """
    Get all (no pagination) _open_ merge request of the project, earliest first

    Returns:
        mrs (iterator): list of merge requests
    """

    print(" Checking project for open merge requests")
    mrs = self._project.mergerequests.list(state="opened", get_all=True, sort="asc")
    print(f"  Found {str(len(mrs))} merge requests")

    return mrs

mark_thread_resolved(mr, search_str, ack_str=None, name=None, url=None)

Mark a thread as resolved

Will look for a thread started by the bot, including the correct search string, then resolve it with a matching comment.

Parameters:

Name Type Description Default
mr

merge request to work one

 required
search_str str

string to look for in discussion

 required
ack_str str

string to respond with (optional)

 None
name str

user to address (optional)

 None
url str

URL to the signature (optional)

 None
Source code in src/bot/project.py 
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
def mark_thread_resolved(self, mr, search_str, ack_str=None, name=None, url=None):
    """
    Mark a thread as resolved

    Will look for a thread started by the bot, including the correct search string,
    then resolve it with a matching comment.

    Parameters:
        mr (): merge request to work one
        search_str (str): string to look for in discussion
        ack_str (str): string to respond with (optional)
        name (str): user to address (optional)
        url (str): URL to the signature (optional)
    """

    if name is not None:
        search_str = name + search_str

    # Find discussion thread
    for discussion in mr.discussions.list():
        # Scan notes
        for note in discussion.attributes["notes"]:
            # Look for our notes
            # Look for us asking to sign
            if note["author"]["id"] == self._me.id and self.string_match(
                note["body"], search_str
            ):
                # Add a note if we want one
                if ack_str is not None:
                    self.add_comment(discussion, ack_str, name, url)
                # Mark discussion thread resolved
                if note["resolved"] is not True:
                    print(f"       {tc.GRN}Marking thread as resolved{tc.CLR}")
                    discussion.resolved = True
                    discussion.save()

obtain_involved_users(mr)

Return list of user ids that show up in MR along with their names. At a minimum, it will return MR author.

Names that cannot be resolved (or are not unique) on the GitLab instance will be listed.

Parameters:

Name Type Description Default
mr

merge request to check

 required

Returns:

Name Type Description
users int

list of user id
names str

list of user names
not_found str

list of names not found
Source code in src/bot/project.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def obtain_involved_users(self, mr):
    """
    Return list of user ids that show up in MR along with their names.
    At a minimum, it will return MR author.

    Names that cannot be resolved (or are not unique)
    on the GitLab instance will be listed.

    Parameters:
        mr (): merge request to check

    Returns:
        users (int): list of user id
        names (str): list of user names
        not_found (str): list of names not found
    """

    # MR author is at least one
    users = [mr.author["id"]]
    names = [mr.author["name"]]
    not_found = []
    # Cycle all involved commits and extract authors/committers
    for commit in mr.commits():
        for name in [
            commit.attributes["author_name"],
            commit.attributes["committer_name"],
        ]:
            user = self._gl.users.list(search=name)
            if len(user) == 1:
                if user[0].id not in users:
                    users.append(user[0].id)
                    names.append(name)
            else:
                not_found.append(name)

    return users, names, not_found

role_is_in_auto_allow(user)

Check if user is project member and has required role. Will issue a free hall pass if allowed.

Parameters:

Name Type Description Default
user int

numeric user id

 required

Returns:

Name Type Description
hall_pass bool

true if role is allowed
Source code in src/bot/project.py 
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
def role_is_in_auto_allow(self, user):
    """
    Check if user is project member and has required role.
    Will issue a free hall pass if allowed.

    Parameters:
        user (int): numeric user id

    Returns:
        hall_pass (bool): true if role is allowed
    """

    hall_pass = False

    if self._auto_allow:
        minimum_level = 60  # is that an admin?
        # GitLab maps roles to a numeric access level, so do we
        if self._auto_allow == "Developer":
            minimum_level = 30
        elif self._auto_allow == "Maintainer":
            minimum_level = 40
        elif self._auto_allow == "Owner":
            minimum_level = 50
        elif self._auto_allow == "Allmighty":
            # We add this level for testing purposes, so a bot developer
            # may not automatically be approved
            minimum_level = 99
        else:
            print(
                f"       {tc.RED}Role not known, please provide either "
                f"'Developer', 'Maintainer', or 'Owner', skipping role check{tc.CLR}"
            )
            minimum_level = 666

        for member in self._project.members_all.list():
            member_role = member.attributes["access_level"]
            if user == member.get_id() and member_role >= minimum_level:
                print(
                    "       Contributor is project member "
                    f"with matching permissions ({str(member_role)})"
                )
                hall_pass = True

    return hall_pass

start_discussion(target, comment, name=None)

Start a discussion within a merge request, adding a comment

This will leave an unresolved thread in the process

Parameters:

Name Type Description Default
target

merge request to comment on

 required
comment str

comment to leave

 required
name str

user to address (optional)

 None

Returns:

Name Type Description
flag bool

true if comment added, false if wrong note type
Source code in src/bot/project.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def start_discussion(self, target, comment, name=None):
    """
    Start a discussion within a merge request, adding a comment

    This will leave an unresolved thread in the process

    Parameters:
        target (): merge request to comment on
        comment (str): comment to leave
        name (str): user to address (optional)

    Returns:
        flag (bool): true if comment added, false if wrong note type
    """

    what_we_comment_on = ""
    if isinstance(target, gitlab.v4.objects.merge_requests.ProjectMergeRequest):
        what_we_comment_on = "MR"

    if not what_we_comment_on:
        print(f"{tc.RED}Discussion misguided{tc.CLR}")
        return False

    if self._dry:
        print(
            f"{tc.YLW}[dry]  Adding discussion to {what_we_comment_on}: "
            f"{str(target.id)}{tc.CLR}"
        )
        print(f"       {name + comment}")
    else:
        print(
            f"       {tc.YLW}Adding comment to {what_we_comment_on}: {str(target.id)}{tc.CLR}"
        )
        if name is not None:
            print("       Addressing " + name)
            target.discussions.create({"body": name + comment})
        else:
            target.discussions.create({"body": comment})

    return True

string_match(string_to_search, string_to_match)

Check if one string is a substring anywhere within the other. Ignores any spaces and newlines, also ignores case.

Parameters:

Name Type Description Default
string_to_search str

string to search within

 required
string_to_match str

string to search for

 required

Returns:

Name Type Description
flag bool

true when search string found
Source code in src/bot/project.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def string_match(self, string_to_search, string_to_match):
    """
    Check if one string is a substring anywhere within the other.
    Ignores any spaces and newlines, also ignores case.

    Parameters:
        string_to_search (str): string to search within
        string_to_match (str): string to search for

    Returns:
        flag (bool): true when search string found
    """
    searched_string = string_to_search.casefold().translate(
        {ord(c): None for c in string.whitespace}
    )
    search_string = string_to_match.casefold().translate(
        {ord(c): None for c in string.whitespace}
    )

    # Comparison should include 0 since the find could be at the start of the string
    return searched_string.find(search_string) >= 0

user_is_in_allow_list(user)

Check if user is in allow list. Will issue a free hall pass.

Parameters:

Name Type Description Default
user int

numeric user id

 required

Returns:

Name Type Description
hall_pass bool

true if user is allowed
Source code in src/bot/project.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
def user_is_in_allow_list(self, user):
    """
    Check if user is in allow list.
    Will issue a free hall pass.

    Parameters:
        user (int): numeric user id

    Returns:
        hall_pass (bool): true if user is allowed
    """

    hall_pass = False

    if self._allow_list:
        for member in self._allow_ids:
            if user == member:
                print(f"       {tc.GRN}Contributor is in allow list{tc.CLR}")
                hall_pass = True

    return hall_pass

wait_until_its_our_turn(current_pipeline_id=None)

Wait until our pipeline is the one created first out of the running ones.

Parameters:

Name Type Description Default
current_pipeline_id str

active pipeline id, string taken from ENV

 None
Source code in src/bot/project.py 
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
def wait_until_its_our_turn(self, current_pipeline_id=None):
    """
    Wait until our pipeline is the one created first out of the running ones.

    Parameters:
        current_pipeline_id (str): active pipeline id, string taken from ENV

    """

    if current_pipeline_id is None:
        sys.exit(f"{tc.RED}Smells like a rotten pipeline - not running in a pipeline?{tc.CLR}")

    # Get first of all running pipelines and wait until this is us (inside HOME)
    while self._project.pipelines.list(status="running", get_all=True, sort="asc")[0].id != int(
        current_pipeline_id
    ):
        print("Paitiently waiting our turn...", flush=True)
        sleep(5)

    return None

ProjectData dataclass

Dataclass to store project data. Includes name and location (either GitLab internal id or URL) as well as necessary access tokens.

Source code in src/bot/project.py 
20
21
22
23
24
25
26
27
28
29
30
@dataclass(frozen=True)
class ProjectData:
    """
    Dataclass to store project data. Includes name and location (either GitLab
    internal id or URL) as well as necessary access tokens.
    """

    name: str
    id: int
    url: str
    token: str

src.bot.repository

Repository

Interact with git.

Required to store signature database within the git repository, will track signatures over time. If used in 'dry mode', new/changed files will not be committed to the repository. The final git commit/push relies on the destructor being executed sucessfully.

Source code in src/bot/repository.py 
 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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class Repository:
    """
    Interact with git.

    Required to store signature database within the git repository, will track
    signatures over time. If used in 'dry mode', new/changed files will not be
    committed to the repository.
    The final git commit/push relies on the destructor being executed
    sucessfully.
    """

    def __init__(self, dry_run=False, token=None, store_url=None, store_branch="main"):
        """
        Constructor for `Repository`.

        This will clone the repository and check out the database file.
        Committing/pushing changes will happen in the class destructor!

        Parameters:
            dry_run (bool): toggle dry run w/o actually using a repository
            token (str): access token (read/write repo) to access repository
            store_url (str): URL of repository
            store_branch (str): branch to check out
        """

        # Do we run in dry-run mode?
        self._dry = dry_run
        # Access token for repo
        self._token = token
        # Storage URL
        self._store_url = store_url
        # Branch inside repository
        self._store_branch = store_branch

        if dry_run:
            # to keep the temp files for later or artifacts
            self._sig_store_path = tempfile.mkdtemp(dir=".")
        else:
            self._sig_store_tmp = tempfile.TemporaryDirectory()  # dir=".")
            self._sig_store_path = self._sig_store_tmp.name

        # Now get an instance of our project to store signatures
        self._sig_store = self.__connect_to_project()

    def __del__(self):
        """
        Destructor for `Repository`.

        This will commit changed and untracked files.
        If there are any changes, those will be pushed to the repository.
        This will again rely on the access token and variables used in the
        constructor.
        """

        print("Closing signature store")

        author = git.Actor("CLA Bot", "anon@nowhere.net")
        committer = git.Actor("CLA Bot", "anon@nowhere.net")

        # Check for untracked files
        print(" Checking for new database")
        if len(self._sig_store.untracked_files) > 0:
            print(f"  {tc.YLW}Adding new files{tc.CLR}")
            print(self._sig_store.untracked_files)
            self._sig_store.index.add(self._sig_store.untracked_files)
            # Commit changes
            self._sig_store.index.commit(
                "CLA Bot adding new database", author=author, committer=committer
            )

        # Check for changes
        print(" Checking for changes in database")
        if len(self._sig_store.index.diff(None)) > 0:
            print(f"  {tc.YLW}Adding changed files{tc.CLR}")
            self._sig_store.git.add(u=True)  # no idea why this is preferred...
            # Commit changes
            self._sig_store.index.commit(
                "CLA Bot adding signature", author=author, committer=committer
            )
        # Push if appropriate
        if self._dry:
            print("[dry] Pushing back changes to signature repository")
        else:
            print(f"      {tc.YLW}Pushing back changes to signature repository{tc.CLR}")
            if self._new_branch:
                self._sig_store.git.push("origin", "-u", self._store_branch)
            else:
                origin = self._sig_store.remotes.origin
                try:
                    origin.push().raise_if_error()
                except Exception as e1:
                    print(
                        f"      {tc.RED}Failed to push changes to store."
                        " Perhaps a protected branch and not a high enough role"
                        f" for the access token?{tc.CLR}\nThe error is {e1}."
                    )

    def __connect_to_project(self):
        """
        Connect to the project storing signatures.

        Use access token to clone git repository and check out database file.
        Will be used from the class constructor.

        Returns:
            repo (Repo): instance pointing to the cloned directory
        """

        if self._dry:
            print("[dry] Cloning signature store")
        else:
            print("Cloning signature store")

        # Connect to our git repo
        if self._token == "":
            sys.exit(
                f"{tc.RED}Token for git authentication required! Please provide it via CI"
                " variable or parameter.{tc.CLR}"
            )
        #  The repo should be _private_, so we need access tokens to get to it
        self._store_url_tokenised = self._store_url.replace(
            "https://", "https://BOT:" + self._token + "@"
        )
        repo = git.Repo.clone_from(self._store_url_tokenised, self._sig_store_path)

        print(
            f" {tc.YLW}Cloned repository located at {self._store_url} "
            f"into {self._sig_store_path}{tc.CLR}"
        )

        # Switch to the appropriate branch
        try:
            repo.git.checkout(self._store_branch)
            # We have encountered an existing branch, so keep everything as is
            self._new_branch = False
        except Exception as e1:
            print(
                f"   {tc.YLW}{e1}\n  Failed to checkout store branch\n  Creating it now...{tc.CLR}"
            )
            try:
                repo.git.checkout("HEAD", b=self._store_branch)
                # This is a new branch, so start from scratch, deleting all files
                for (_path, _stage), _entry in repo.index.entries.items():
                    repo.git.rm([_path])
            except Exception as e2:
                print(
                    f"   {tc.YLW}{e2}\n  This seems to be a fresh repository"
                    f"\n  Starting from scratch...{tc.CLR}"
                )
                repo.git.switch("--create", self._store_branch)
            self._new_branch = True

        return repo

__connect_to_project()

Connect to the project storing signatures.

Use access token to clone git repository and check out database file. Will be used from the class constructor.

Returns:

Name Type Description
repo Repo

instance pointing to the cloned directory
Source code in src/bot/repository.py 
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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def __connect_to_project(self):
    """
    Connect to the project storing signatures.

    Use access token to clone git repository and check out database file.
    Will be used from the class constructor.

    Returns:
        repo (Repo): instance pointing to the cloned directory
    """

    if self._dry:
        print("[dry] Cloning signature store")
    else:
        print("Cloning signature store")

    # Connect to our git repo
    if self._token == "":
        sys.exit(
            f"{tc.RED}Token for git authentication required! Please provide it via CI"
            " variable or parameter.{tc.CLR}"
        )
    #  The repo should be _private_, so we need access tokens to get to it
    self._store_url_tokenised = self._store_url.replace(
        "https://", "https://BOT:" + self._token + "@"
    )
    repo = git.Repo.clone_from(self._store_url_tokenised, self._sig_store_path)

    print(
        f" {tc.YLW}Cloned repository located at {self._store_url} "
        f"into {self._sig_store_path}{tc.CLR}"
    )

    # Switch to the appropriate branch
    try:
        repo.git.checkout(self._store_branch)
        # We have encountered an existing branch, so keep everything as is
        self._new_branch = False
    except Exception as e1:
        print(
            f"   {tc.YLW}{e1}\n  Failed to checkout store branch\n  Creating it now...{tc.CLR}"
        )
        try:
            repo.git.checkout("HEAD", b=self._store_branch)
            # This is a new branch, so start from scratch, deleting all files
            for (_path, _stage), _entry in repo.index.entries.items():
                repo.git.rm([_path])
        except Exception as e2:
            print(
                f"   {tc.YLW}{e2}\n  This seems to be a fresh repository"
                f"\n  Starting from scratch...{tc.CLR}"
            )
            repo.git.switch("--create", self._store_branch)
        self._new_branch = True

    return repo

__del__()

Destructor for Repository.

This will commit changed and untracked files. If there are any changes, those will be pushed to the repository. This will again rely on the access token and variables used in the constructor.

Source code in src/bot/repository.py 
 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
def __del__(self):
    """
    Destructor for `Repository`.

    This will commit changed and untracked files.
    If there are any changes, those will be pushed to the repository.
    This will again rely on the access token and variables used in the
    constructor.
    """

    print("Closing signature store")

    author = git.Actor("CLA Bot", "anon@nowhere.net")
    committer = git.Actor("CLA Bot", "anon@nowhere.net")

    # Check for untracked files
    print(" Checking for new database")
    if len(self._sig_store.untracked_files) > 0:
        print(f"  {tc.YLW}Adding new files{tc.CLR}")
        print(self._sig_store.untracked_files)
        self._sig_store.index.add(self._sig_store.untracked_files)
        # Commit changes
        self._sig_store.index.commit(
            "CLA Bot adding new database", author=author, committer=committer
        )

    # Check for changes
    print(" Checking for changes in database")
    if len(self._sig_store.index.diff(None)) > 0:
        print(f"  {tc.YLW}Adding changed files{tc.CLR}")
        self._sig_store.git.add(u=True)  # no idea why this is preferred...
        # Commit changes
        self._sig_store.index.commit(
            "CLA Bot adding signature", author=author, committer=committer
        )
    # Push if appropriate
    if self._dry:
        print("[dry] Pushing back changes to signature repository")
    else:
        print(f"      {tc.YLW}Pushing back changes to signature repository{tc.CLR}")
        if self._new_branch:
            self._sig_store.git.push("origin", "-u", self._store_branch)
        else:
            origin = self._sig_store.remotes.origin
            try:
                origin.push().raise_if_error()
            except Exception as e1:
                print(
                    f"      {tc.RED}Failed to push changes to store."
                    " Perhaps a protected branch and not a high enough role"
                    f" for the access token?{tc.CLR}\nThe error is {e1}."
                )

__init__(dry_run=False, token=None, store_url=None, store_branch='main')

Constructor for Repository.

This will clone the repository and check out the database file. Committing/pushing changes will happen in the class destructor!

Parameters:

Name Type Description Default
dry_run bool

toggle dry run w/o actually using a repository

 False
token str

access token (read/write repo) to access repository

 None
store_url str

URL of repository

 None
store_branch str

branch to check out

 'main'
Source code in src/bot/repository.py 
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
def __init__(self, dry_run=False, token=None, store_url=None, store_branch="main"):
    """
    Constructor for `Repository`.

    This will clone the repository and check out the database file.
    Committing/pushing changes will happen in the class destructor!

    Parameters:
        dry_run (bool): toggle dry run w/o actually using a repository
        token (str): access token (read/write repo) to access repository
        store_url (str): URL of repository
        store_branch (str): branch to check out
    """

    # Do we run in dry-run mode?
    self._dry = dry_run
    # Access token for repo
    self._token = token
    # Storage URL
    self._store_url = store_url
    # Branch inside repository
    self._store_branch = store_branch

    if dry_run:
        # to keep the temp files for later or artifacts
        self._sig_store_path = tempfile.mkdtemp(dir=".")
    else:
        self._sig_store_tmp = tempfile.TemporaryDirectory()  # dir=".")
        self._sig_store_path = self._sig_store_tmp.name

    # Now get an instance of our project to store signatures
    self._sig_store = self.__connect_to_project()

'Database' access

src.bot.csvlite

CSVLite

Interact with csv file acting as database that holds known contributors. The class will read all contents of the csv file into memory. Adding a new contributor will happen in memory and disk at the same time. 'dry mode' will still edit data on disk, the class relies on data not being committed to git in 'dry mode'.

Data stored for contributors
  • id: internal numbering
  • user_id: GitLab user id
  • name: user's name
  • email: user's email
  • individual: 0/1 if signed as individual or instiution
  • start_date: user first seen
  • cla_version: version of CLA that has been signed, ie. the date of it
  • signed_with_mr: if signed via comment, note the corresponding MR
  • last_sign_date: last time CLA has been signed
  • end_date: if we want to expire the user

Methods:

Name Description
connect_to_csvfile

Open 'database' file, returns all contents.
add_signature

Adding entry to 'database'.
signature_valid

Check for valid contributor.
Source code in src/bot/csvlite.py 
 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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
class CSVLite:
    """
    Interact with csv file acting as database that holds known contributors.
    The class will read all contents of the csv file into memory. Adding a new
    contributor will happen in memory and disk at the same time. 'dry mode'
    will still edit data on disk, the class relies on data not being committed
    to git in 'dry mode'.

    Data stored for contributors:
        - `id`: internal numbering
        - `user_id`: GitLab user id
        - `name`: user's name
        - `email`: user's email
        - `individual`: 0/1 if signed as individual or instiution
        - `start_date`: user first seen
        - `cla_version`: version of CLA that has been signed, ie. the date of it
        - `signed_with_mr`: if signed via comment, note the corresponding MR
        - `last_sign_date`: last time CLA has been signed
        - `end_date`: if we want to expire the user

    Methods:
        connect_to_csvfile: Open 'database' file, returns all contents.
        add_signature: Adding entry to 'database'.
        signature_valid: Check for valid contributor.
    """

    def __init__(self, dry_run=False, path=None, project_id=None):
        """
        Constructor for `CSV`.

        Parameters:
            dry_run (bool): toggle dry run, effectively only changing print statements.
            path (str): path to store database file.
            project_id (int): numeric project id (used to extend database path).
        """

        # Do we run in dry-run mode?
        self._dry = dry_run
        # File path
        self._path = path
        self._full_path = None
        # Project id
        self._id = project_id
        # Database name
        self._db_name = "signatures.csv"
        # Are we working with a maiden db?
        self._new_database = False
        # Define headers for our csv file (UPDATE ABOVE!)
        #     id: just because
        #     user_id: GitLab user id
        #     name: user's name
        #     email: user's email
        #     individual: 0/1 if signed as individual or instiution
        #     start_date: user first seen
        #     cla_version: version of CLA that has been signed, ie. the date of it
        #     signed_with_mr: if signed via comment, note the corresponding MR
        #     last_sign_date: last time CLA has been signed
        #     end_date: if we want to expire the user
        self._headers = [
            "id",
            "user_id",
            "name",
            "email",
            "individual",
            "start_date",
            "cla_version",
            "signed_with_mr",
            "last_sign_date",
            "end_date",
        ]

        # Now get an instance of the 'database'
        #   will return a list with Dicts containing the data
        self._dict_list = self.connect_to_csvfile()

    def __del__(self):
        """
        Destructor for `CSV`.

        If instantiated in `dry_run` mode, will print the final content of the database.
        """

        if self._dry:
            print("[dry]        Final content of database")
            for row in self._dict_list:
                print(row)

    def connect_to_csvfile(self):
        """
        Connect to the csv file (the 'database').

        This will in fact either read the file and return a Dict
        or create an emtpy Dict and return that...

        Returns:
           dict (Dict): all of the the csv data

        """

        if self._dry:
            print("[dry] Connecting csv file")
        else:
            print("Connecting csv file")

        # Check if we start from scatch
        if not os.path.exists(self._path + "/" + str(self._id)):
            self._new_database = True
            os.mkdir(self._path + "/" + str(self._id))

        # Construct path to csv file in the git repo
        tmp_path = str(self._id) + "/" + self._db_name
        self._full_path = self._path + "/" + tmp_path

        print(f" {tc.YLW}Database kept in {tmp_path}{tc.CLR}")

        if self._new_database:
            # Init our csv file with a header
            with open(self._full_path, "w") as csvfile:
                # Creating a csv dict writer object
                writer = csv.DictWriter(csvfile, fieldnames=self._headers)
                # Writing headers (field names)
                writer.writeheader()

        # Open the csv file
        with open(self._full_path) as csvfile:
            # Create a CSV reader with DictReader
            reader = csv.DictReader(csvfile)
            # Initialize an empty list to store the dictionaries
            dict_list = []
            # Iterate through each row in the CSV file
            for row in reader:
                # Append each row (as a dictionary) to the list
                dict_list.append(row)

        if self._dry:
            print("[dry]        Initial content of database")
            for row in dict_list:
                print(row)

        return dict_list

    def add_signature(self, user_id=None, name=None, email=None, individual=True, mr=None):
        """
        Add a signature for a single contributor to database.

        Parameters:
            user_id (int): user id in GitLab (to match commits/MRs)
            name (str): user's name in GitLab (human readable name)
            email (str): user's email in GitLab (contact email)
            individual (bool): signing as individual? (unless a legal entity)
            mr (int): numeric id of MR that had signature

        """

        start_date = str(datetime.datetime.now())
        last_sign_date = start_date
        cla_version = ""
        signed_with_mr = mr

        payload = [
            {
                "id": -1,
                "user_id": user_id,
                "name": name,
                "email": email,
                "individual": int(individual),
                "start_date": start_date,
                "cla_version": cla_version,
                "signed_with_mr": signed_with_mr,
                "last_sign_date": last_sign_date,
                "end_date": -1,
            }
        ]

        # Append to our 'in-memory' database
        self._dict_list.append(payload[0])

        # Append to file on disk
        with open(self._full_path, "a") as csvfile:
            # Creating a csv dict writer object
            writer = csv.DictWriter(csvfile, fieldnames=self._headers)
            # Writing data row
            writer.writerows(payload)

        return None

    def signature_valid(self, user_id=None):
        """
        Check if a signature is (still) valid.
        Will also fail for no entry in database for that user.

        Parameters:
            user_id (int): user id in GitLab (to match commits/MRs)

        Returns:
            is_valid (bool): true if signature found and valid

        """

        # now = datetime.datetime.now()

        for row in self._dict_list:  # noqa: SIM110
            # TODO: have more checks than just the uid?
            if row["user_id"] == str(user_id):
                # TODO: check if still valid!

                # TODO: this still needs the CLA version...
                return True

        return False

__del__()

Destructor for CSV.

If instantiated in dry_run mode, will print the final content of the database.

Source code in src/bot/csvlite.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __del__(self):
    """
    Destructor for `CSV`.

    If instantiated in `dry_run` mode, will print the final content of the database.
    """

    if self._dry:
        print("[dry]        Final content of database")
        for row in self._dict_list:
            print(row)

__init__(dry_run=False, path=None, project_id=None)

Constructor for CSV.

Parameters:

Name Type Description Default
dry_run bool

toggle dry run, effectively only changing print statements.

 False
path str

path to store database file.

 None
project_id int

numeric project id (used to extend database path).

 None
Source code in src/bot/csvlite.py 
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
def __init__(self, dry_run=False, path=None, project_id=None):
    """
    Constructor for `CSV`.

    Parameters:
        dry_run (bool): toggle dry run, effectively only changing print statements.
        path (str): path to store database file.
        project_id (int): numeric project id (used to extend database path).
    """

    # Do we run in dry-run mode?
    self._dry = dry_run
    # File path
    self._path = path
    self._full_path = None
    # Project id
    self._id = project_id
    # Database name
    self._db_name = "signatures.csv"
    # Are we working with a maiden db?
    self._new_database = False
    # Define headers for our csv file (UPDATE ABOVE!)
    #     id: just because
    #     user_id: GitLab user id
    #     name: user's name
    #     email: user's email
    #     individual: 0/1 if signed as individual or instiution
    #     start_date: user first seen
    #     cla_version: version of CLA that has been signed, ie. the date of it
    #     signed_with_mr: if signed via comment, note the corresponding MR
    #     last_sign_date: last time CLA has been signed
    #     end_date: if we want to expire the user
    self._headers = [
        "id",
        "user_id",
        "name",
        "email",
        "individual",
        "start_date",
        "cla_version",
        "signed_with_mr",
        "last_sign_date",
        "end_date",
    ]

    # Now get an instance of the 'database'
    #   will return a list with Dicts containing the data
    self._dict_list = self.connect_to_csvfile()

add_signature(user_id=None, name=None, email=None, individual=True, mr=None)

Add a signature for a single contributor to database.

Parameters:

Name Type Description Default
user_id int

user id in GitLab (to match commits/MRs)

 None
name str

user's name in GitLab (human readable name)

 None
email str

user's email in GitLab (contact email)

 None
individual bool

signing as individual? (unless a legal entity)

 True
mr int

numeric id of MR that had signature

 None
Source code in src/bot/csvlite.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def add_signature(self, user_id=None, name=None, email=None, individual=True, mr=None):
    """
    Add a signature for a single contributor to database.

    Parameters:
        user_id (int): user id in GitLab (to match commits/MRs)
        name (str): user's name in GitLab (human readable name)
        email (str): user's email in GitLab (contact email)
        individual (bool): signing as individual? (unless a legal entity)
        mr (int): numeric id of MR that had signature

    """

    start_date = str(datetime.datetime.now())
    last_sign_date = start_date
    cla_version = ""
    signed_with_mr = mr

    payload = [
        {
            "id": -1,
            "user_id": user_id,
            "name": name,
            "email": email,
            "individual": int(individual),
            "start_date": start_date,
            "cla_version": cla_version,
            "signed_with_mr": signed_with_mr,
            "last_sign_date": last_sign_date,
            "end_date": -1,
        }
    ]

    # Append to our 'in-memory' database
    self._dict_list.append(payload[0])

    # Append to file on disk
    with open(self._full_path, "a") as csvfile:
        # Creating a csv dict writer object
        writer = csv.DictWriter(csvfile, fieldnames=self._headers)
        # Writing data row
        writer.writerows(payload)

    return None

connect_to_csvfile()

Connect to the csv file (the 'database').

This will in fact either read the file and return a Dict or create an emtpy Dict and return that...

Returns:

Name Type Description
dict Dict

all of the the csv data
Source code in src/bot/csvlite.py 
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
151
152
153
154
155
def connect_to_csvfile(self):
    """
    Connect to the csv file (the 'database').

    This will in fact either read the file and return a Dict
    or create an emtpy Dict and return that...

    Returns:
       dict (Dict): all of the the csv data

    """

    if self._dry:
        print("[dry] Connecting csv file")
    else:
        print("Connecting csv file")

    # Check if we start from scatch
    if not os.path.exists(self._path + "/" + str(self._id)):
        self._new_database = True
        os.mkdir(self._path + "/" + str(self._id))

    # Construct path to csv file in the git repo
    tmp_path = str(self._id) + "/" + self._db_name
    self._full_path = self._path + "/" + tmp_path

    print(f" {tc.YLW}Database kept in {tmp_path}{tc.CLR}")

    if self._new_database:
        # Init our csv file with a header
        with open(self._full_path, "w") as csvfile:
            # Creating a csv dict writer object
            writer = csv.DictWriter(csvfile, fieldnames=self._headers)
            # Writing headers (field names)
            writer.writeheader()

    # Open the csv file
    with open(self._full_path) as csvfile:
        # Create a CSV reader with DictReader
        reader = csv.DictReader(csvfile)
        # Initialize an empty list to store the dictionaries
        dict_list = []
        # Iterate through each row in the CSV file
        for row in reader:
            # Append each row (as a dictionary) to the list
            dict_list.append(row)

    if self._dry:
        print("[dry]        Initial content of database")
        for row in dict_list:
            print(row)

    return dict_list

signature_valid(user_id=None)

Check if a signature is (still) valid. Will also fail for no entry in database for that user.

Parameters:

Name Type Description Default
user_id int

user id in GitLab (to match commits/MRs)

 None

Returns:

Name Type Description
is_valid bool

true if signature found and valid
Source code in src/bot/csvlite.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def signature_valid(self, user_id=None):
    """
    Check if a signature is (still) valid.
    Will also fail for no entry in database for that user.

    Parameters:
        user_id (int): user id in GitLab (to match commits/MRs)

    Returns:
        is_valid (bool): true if signature found and valid

    """

    # now = datetime.datetime.now()

    for row in self._dict_list:  # noqa: SIM110
        # TODO: have more checks than just the uid?
        if row["user_id"] == str(user_id):
            # TODO: check if still valid!

            # TODO: this still needs the CLA version...
            return True

    return False

src.bot.sqlite

SQLite

Interact with SQLite file acting as database that holds known contributors. The class will open the SQLite file and use a handle to query or add data. Adding a new contributor will happen add an entry to the file, so 'dry mode' will still edit data on disk, the class relies on data not being committed to git in 'dry mode'.

Data stored for contributors
  • id: internal numbering
  • user_id: GitLab user id
  • name: user's name
  • email: user's email
  • individual: 0/1 if signed as individual or instiution
  • start_date: user first seen
  • cla_version: version of CLA that has been signed, ie. the date of it
  • signed_with_mr: if signed via comment, note the corresponding MR
  • last_sign_date: last time CLA has been signed
  • end_date: if we want to expire the user

Methods:

Name Description
connect_to_sqlite_db

Open 'database' file, returns handle.
add_signature

Adding entry to 'database'.
signature_valid

Check for valid contributor.
Source code in src/bot/sqlite.py 
 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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
class SQLite:
    """
    Interact with SQLite file acting as database that holds known contributors.
    The class will open the SQLite file and use a handle to query or add data.
    Adding a new contributor will happen add an entry to the file, so 'dry
    mode' will still edit data on disk, the class relies on data not being
    committed to git in 'dry mode'.

    Data stored for contributors:
        - `id`: internal numbering
        - `user_id`: GitLab user id
        - `name`: user's name
        - `email`: user's email
        - `individual`: 0/1 if signed as individual or instiution
        - `start_date`: user first seen
        - `cla_version`: version of CLA that has been signed, ie. the date of it
        - `signed_with_mr`: if signed via comment, note the corresponding MR
        - `last_sign_date`: last time CLA has been signed
        - `end_date`: if we want to expire the user

    Methods:
        connect_to_sqlite_db: Open 'database' file, returns handle.
        add_signature: Adding entry to 'database'.
        signature_valid: Check for valid contributor.
    """

    def __init__(self, dry_run=False, path=None, project_id=None):
        """
        Constructor for `SQLite`.

        Parameters:
            dry_run (bool): toggle dry run, effectively only changing print statements.
            path (str): path to store database file.
            project_id (int): numeric project id (used to extend database path).
        """

        # Do we run in dry-run mode?
        self._dry = dry_run
        # File path
        self._path = path
        # Project id
        self._id = project_id
        # Database name
        self._db_name = "signatures.sqlite"
        # Are we working with a maiden db?
        self._new_database = False

        # Now get an instance of the database and a 'cursor' (what a name...)
        self._conn, self._db = self.connect_to_sqlite_db()

    def __del__(self):
        """
        Destructor for `SQLite`.

        If instantiated in `dry_run` mode, will print the final content of the database.
        """

        if self._dry:
            print("[dry]        Final content of database")
            self._db.execute("SELECT * FROM approved_contributors")
            rows = self._db.fetchall()
            print(rows)

    def connect_to_sqlite_db(self):
        """
        Connect to the SQLite database file.

        This will in fact either connect to an existing file
        or create a new emtpy one and return handles to that...

        Returns:
            connection: database connection
            db: connection cursor for queries
        """

        if self._dry:
            print("[dry] Connecting SQLite file")
        else:
            print("Connecting SQLite file")

        # Check if we start from scatch
        if not os.path.exists(self._path + "/" + str(self._id)):
            self._new_database = True
            os.mkdir(self._path + "/" + str(self._id))

        # Connect to our SQLite file in the git repo
        tmp_path = str(self._id) + "/" + self._db_name
        full_path = self._path + "/" + tmp_path
        connection = sqlite3.connect(full_path)

        print(f" {tc.YLW}Database kept in {tmp_path}{tc.CLR}")

        # Get cursor for the actual interaction
        cursor = connection.cursor()

        if self._new_database:
            # Init our table (UPDATE ABOVE!)
            #     id: just because
            #     user_id: GitLab user id
            #     name: user's name
            #     email: user's email
            #     individual: 0/1 if signed as individual or instiution
            #     start_date: user first seen
            #     cla_version: version of CLA that has been signed, ie. the date of it
            #     signed_with_mr: if signed via comment, note the corresponding MR
            #     last_sign_date: last time CLA has been signed
            #     end_date: if we want to expire the user
            create_table = """CREATE TABLE IF NOT EXISTS approved_contributors (
                                   id INTEGER PRIMARY KEY,
                                   user_id INTEGER NOT NULL,
                                   name text NOT NULL,
                                   email text NOT NULL,
                                   individual INTEGER NOT NULL,
                                   start_date TEXT NOT NULL,
                                   cla_version TEXT,
                                   signed_with_mr TEXT,
                                   last_sign_date TEXT NOT NULL,
                                   end_date TEXT
                           );"""
            cursor.execute(create_table)
            connection.commit()

        if self._dry:
            print("[dry]        Initial content of database")
            cursor.execute("SELECT * FROM approved_contributors")
            rows = cursor.fetchall()
            print(rows)

        return connection, cursor

    def add_signature(self, user_id=None, name=None, email=None, individual=True, mr=None):
        """
        Add a signature for a single contributor to database.

        Parameters:
            user_id (int): user id in GitLab (to match commits/MRs)
            name (str): user's name in GitLab (human readable name)
            email (str): user's email in GitLab (contact email)
            individual (bool): signing as individual? (unless a legal entity)
            mr (int): numeric id of MR that had signature
        """

        start_date = str(datetime.datetime.now())
        last_sign_date = start_date
        cla_version = ""
        signed_with_mr = mr

        payload = (
            user_id,
            name,
            email,
            int(individual),
            start_date,
            cla_version,
            signed_with_mr,
            last_sign_date,
        )

        create_entry = """ INSERT INTO approved_contributors(
                              user_id,
                              name,
                              email,
                              individual,
                              start_date,
                              cla_version,
                              signed_with_mr,
                              last_sign_date)
                        VALUES(?,?,?,?,?,?,?,?) """

        self._db.execute(create_entry, payload)
        self._conn.commit()

        return None

    def signature_valid(self, user_id=None):
        """
        Check if a signature is (still) valid.
        Will also fail for no entry in database for that user.

        Parameters:
            user_id (int): user id in GitLab (to match commits/MRs)

        Returns:
            is_valid (bool): true if signature found and valid
        """

        # now = datetime.datetime.now()

        self._db.execute("SELECT * FROM approved_contributors WHERE user_id = ?", (user_id,))
        row = self._db.fetchone()

        # TODO: have more checks than just the uid?
        if row is not None:  # noqa: SIM103
            # TODO: check if still valid!

            # TODO: this still needs the CLA version...
            return True

        return False

__del__()

Destructor for SQLite.

If instantiated in dry_run mode, will print the final content of the database.

Source code in src/bot/sqlite.py 
68
69
70
71
72
73
74
75
76
77
78
79
def __del__(self):
    """
    Destructor for `SQLite`.

    If instantiated in `dry_run` mode, will print the final content of the database.
    """

    if self._dry:
        print("[dry]        Final content of database")
        self._db.execute("SELECT * FROM approved_contributors")
        rows = self._db.fetchall()
        print(rows)

__init__(dry_run=False, path=None, project_id=None)

Constructor for SQLite.

Parameters:

Name Type Description Default
dry_run bool

toggle dry run, effectively only changing print statements.

 False
path str

path to store database file.

 None
project_id int

numeric project id (used to extend database path).

 None
Source code in src/bot/sqlite.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(self, dry_run=False, path=None, project_id=None):
    """
    Constructor for `SQLite`.

    Parameters:
        dry_run (bool): toggle dry run, effectively only changing print statements.
        path (str): path to store database file.
        project_id (int): numeric project id (used to extend database path).
    """

    # Do we run in dry-run mode?
    self._dry = dry_run
    # File path
    self._path = path
    # Project id
    self._id = project_id
    # Database name
    self._db_name = "signatures.sqlite"
    # Are we working with a maiden db?
    self._new_database = False

    # Now get an instance of the database and a 'cursor' (what a name...)
    self._conn, self._db = self.connect_to_sqlite_db()

add_signature(user_id=None, name=None, email=None, individual=True, mr=None)

Add a signature for a single contributor to database.

Parameters:

Name Type Description Default
user_id int

user id in GitLab (to match commits/MRs)

 None
name str

user's name in GitLab (human readable name)

 None
email str

user's email in GitLab (contact email)

 None
individual bool

signing as individual? (unless a legal entity)

 True
mr int

numeric id of MR that had signature

 None
Source code in src/bot/sqlite.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def add_signature(self, user_id=None, name=None, email=None, individual=True, mr=None):
    """
    Add a signature for a single contributor to database.

    Parameters:
        user_id (int): user id in GitLab (to match commits/MRs)
        name (str): user's name in GitLab (human readable name)
        email (str): user's email in GitLab (contact email)
        individual (bool): signing as individual? (unless a legal entity)
        mr (int): numeric id of MR that had signature
    """

    start_date = str(datetime.datetime.now())
    last_sign_date = start_date
    cla_version = ""
    signed_with_mr = mr

    payload = (
        user_id,
        name,
        email,
        int(individual),
        start_date,
        cla_version,
        signed_with_mr,
        last_sign_date,
    )

    create_entry = """ INSERT INTO approved_contributors(
                          user_id,
                          name,
                          email,
                          individual,
                          start_date,
                          cla_version,
                          signed_with_mr,
                          last_sign_date)
                    VALUES(?,?,?,?,?,?,?,?) """

    self._db.execute(create_entry, payload)
    self._conn.commit()

    return None

connect_to_sqlite_db()

Connect to the SQLite database file.

This will in fact either connect to an existing file or create a new emtpy one and return handles to that...

Returns:

Name Type Description
connection

database connection
db

connection cursor for queries
Source code in src/bot/sqlite.py 
 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
def connect_to_sqlite_db(self):
    """
    Connect to the SQLite database file.

    This will in fact either connect to an existing file
    or create a new emtpy one and return handles to that...

    Returns:
        connection: database connection
        db: connection cursor for queries
    """

    if self._dry:
        print("[dry] Connecting SQLite file")
    else:
        print("Connecting SQLite file")

    # Check if we start from scatch
    if not os.path.exists(self._path + "/" + str(self._id)):
        self._new_database = True
        os.mkdir(self._path + "/" + str(self._id))

    # Connect to our SQLite file in the git repo
    tmp_path = str(self._id) + "/" + self._db_name
    full_path = self._path + "/" + tmp_path
    connection = sqlite3.connect(full_path)

    print(f" {tc.YLW}Database kept in {tmp_path}{tc.CLR}")

    # Get cursor for the actual interaction
    cursor = connection.cursor()

    if self._new_database:
        # Init our table (UPDATE ABOVE!)
        #     id: just because
        #     user_id: GitLab user id
        #     name: user's name
        #     email: user's email
        #     individual: 0/1 if signed as individual or instiution
        #     start_date: user first seen
        #     cla_version: version of CLA that has been signed, ie. the date of it
        #     signed_with_mr: if signed via comment, note the corresponding MR
        #     last_sign_date: last time CLA has been signed
        #     end_date: if we want to expire the user
        create_table = """CREATE TABLE IF NOT EXISTS approved_contributors (
                               id INTEGER PRIMARY KEY,
                               user_id INTEGER NOT NULL,
                               name text NOT NULL,
                               email text NOT NULL,
                               individual INTEGER NOT NULL,
                               start_date TEXT NOT NULL,
                               cla_version TEXT,
                               signed_with_mr TEXT,
                               last_sign_date TEXT NOT NULL,
                               end_date TEXT
                       );"""
        cursor.execute(create_table)
        connection.commit()

    if self._dry:
        print("[dry]        Initial content of database")
        cursor.execute("SELECT * FROM approved_contributors")
        rows = cursor.fetchall()
        print(rows)

    return connection, cursor

signature_valid(user_id=None)

Check if a signature is (still) valid. Will also fail for no entry in database for that user.

Parameters:

Name Type Description Default
user_id int

user id in GitLab (to match commits/MRs)

 None

Returns:

Name Type Description
is_valid bool

true if signature found and valid
Source code in src/bot/sqlite.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def signature_valid(self, user_id=None):
    """
    Check if a signature is (still) valid.
    Will also fail for no entry in database for that user.

    Parameters:
        user_id (int): user id in GitLab (to match commits/MRs)

    Returns:
        is_valid (bool): true if signature found and valid
    """

    # now = datetime.datetime.now()

    self._db.execute("SELECT * FROM approved_contributors WHERE user_id = ?", (user_id,))
    row = self._db.fetchone()

    # TODO: have more checks than just the uid?
    if row is not None:  # noqa: SIM103
        # TODO: check if still valid!

        # TODO: this still needs the CLA version...
        return True

    return False