job: add is_branch and is_tag functions

Author: bergzand

murdock/job.py

@@ -80,6 +80,12 @@ def remove_dir(work_dir):
         except FileNotFoundError:
             LOGGER.debug(f"Directory '{work_dir}' doesn't exist, cannot remove")
 
+    def is_branch(self) -> bool:
+        return self.ref is not None and self.ref.startswith("refs/heads/")
+
+    def is_tag(self) -> bool:
+        return self.ref is not None and self.ref.startswith("refs/tags/")
+
     @property
     def start_time(self) -> datetime:
         return self._start_time
@@ -183,9 +189,9 @@ def env(self):
                     "CI_BUILD_REPO": GITHUB_CONFIG.repo,
                 }
             )
-            if self.ref.startswith("refs/tags"):
+            if self.is_tag():
                 _env.update({"CI_BUILD_TAG": self.ref[10:]})
-            if self.ref.startswith("refs/heads"):
+            if self.is_branch():
                 _env.update({"CI_BUILD_BRANCH": self.ref[11:]})
 
         return _env
@@ -202,9 +208,9 @@ def title(self):
         commit = self.commit.sha[0:7]
         if self.pr is not None:
             return f"PR #{self.pr.number} ({commit})"
-        elif self.ref is not None and self.ref.startswith("refs/tags"):
+        elif self.is_tag():
             return f"tag {self.ref[10:]} ({commit})"
-        elif self.ref is not None and self.ref.startswith("refs/heads"):
+        elif self.is_branch():
             return f"branch {self.ref[11:]} ({commit})"
         else:
             return f"commit {commit}"

murdock/job_containers.py

@@ -79,13 +79,7 @@ def search_with_query(self, query: JobQueryModel) -> List[MurdockJob]:
         if query.is_branch is not None:
             if query.is_branch is True:
                 is_branch_jobs = {
-                    job
-                    for job in self.jobs
-                    if (
-                        job is not None
-                        and job.ref is not None
-                        and job.ref.startswith("refs/heads/")
-                    )
+                    job for job in self.jobs if job is not None and job.is_branch
                 }
             else:
                 is_branch_jobs = {
@@ -105,13 +99,7 @@ def search_with_query(self, query: JobQueryModel) -> List[MurdockJob]:
         if query.is_tag is not None:
             if query.is_tag is True:
                 is_tag_jobs = {
-                    job
-                    for job in self.jobs
-                    if (
-                        job is not None
-                        and job.ref is not None
-                        and job.ref.startswith("refs/tags/")
-                    )
+                    job for job in self.jobs if job is not None and job.is_tag()
                 }
             else:
                 is_tag_jobs = {

murdock/notify.py

@@ -99,13 +99,13 @@ async def notify(self, job: MurdockJob):
                 f'(<a href="{commit_url}" target="_blank" rel="noreferrer noopener">{commit_short}</a>) '
                 f"by {author}"
             )
-        elif job.ref is not None and job.ref.startswith("refs/tags"):
+        elif job.is_tag():
             tag_url = f"https://github.com/{GITHUB_CONFIG.repo}/tree/{job.ref[10:]}"
             job_html_description = (
                 f'tag <a href="{tag_url}" target="_blank" rel="noreferrer noopener">{job.ref[10:]}</a> '
                 f'(<a href="{commit_url}" target="_blank" rel="noreferrer noopener">{commit_short}</a>)'
             )
-        elif job.ref is not None and job.ref.startswith("refs/heads"):
+        elif job.is_branch():
             branch_url = f"https://github.com/{GITHUB_CONFIG.repo}/tree/{job.ref[11:]}"
             job_html_description = (
                 f'branch <a href="{branch_url}" target="_blank" rel="noreferrer noopener">{job.ref[11:]}</a> '
@@ -167,10 +167,10 @@ async def notify(self, job: MurdockJob, db: Database):
                         f"{job} result still successful, skipping notification"
                     )
                     return
-            if job.ref.startswith("refs/heads"):
+            if job.is_branch():
                 for notifier_type in self.config.branch:
                     await self._notifiers[notifier_type].notify(job)
-            elif job.ref.startswith("refs/tags"):
+            elif job.is_tag():
                 for notifier_type in self.config.tag:
                     await self._notifiers[notifier_type].notify(job)
         else: