14519448990

Committed 17 Apr 2025 03:38PM UTC coverage: 96.774% (-0.4%) from 97.128%

Build # 14519448990

Build Type

Pull #95

github

Committed by

valentinaschueller

Commit Message

Update server_url and project_id

Pull Request Pull Request #95: Gitlab presentation task

Run Details

127 of 139 new or added lines in 2 files covered. (91.37%)

2190 of 2263 relevant lines covered (96.77%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.21

/monitoring/gitlab.py

"""Presentation Task that uploads Data and Plots to the EC-Earth 4 Gitlab."""

import re
import urllib.parse
from pathlib import Path

import gitlab
import jinja2
import requests.exceptions
from scriptengine.exceptions import ScriptEngineTaskRunError
from scriptengine.jinja import filters as j2filters
from scriptengine.tasks.core import Task, timed_runner

from helpers.exceptions import PresentationException
from helpers.presentation_objects import PresentationObject


class Gitlab(Task):
    """Gitlab Presentation Task"""

    _required_arguments = (
        "src",
        "local_dst",
        "subject",
        "template",
        "api_key",
    )

    def __init__(self, arguments=None):
        Gitlab.check_arguments(arguments)
        super().__init__(arguments)

    @timed_runner
    def run(self, context):
        sources = self.getarg("src", context)
        dst_folder = self.getarg("local_dst", context)
        issue_subject = self.getarg("subject", context, parse_yaml=False)
        template_path = self.getarg("template", context)
        key = self.getarg("api_key", context)
        self.log_info(f"Create Gitlab issue '{issue_subject}'.")
        self.log_debug(f"Template: {template_path}, Source File(s): {sources}")

        presentation_list = self.get_presentation_list(sources, dst_folder)
        gitlab_template = self.get_template(context, template_path)
        gitlab_template.globals["urlencode"] = urllib.parse.quote

        server_url = "https://git.smhi.se"
        gl = gitlab.Gitlab(server_url, private_token=key)

        self.log_debug("Connecting to Gitlab.")

        project, issue = self.get_project_and_issue(gl, issue_subject)

        self.log_debug("Uploading attachments.")
        issue.uploads = []
        for item in presentation_list:
            if item["presentation_type"] == "image":
                file_name = Path(item["path"]).name
                upload = project.upload(file_name, filepath=f"{dst_folder}/{file_name}")
                item["image_link"] = upload[
                    "markdown"
                ]  # string of the form "![]()" generated by Gitlab

        # render the template and add as description
        self.log_debug("Updating the issue description.")
        issue.description = gitlab_template.render(
            presentation_list=presentation_list,
            issue_url=issue.web_url,
            create_anchor=create_anchor,
        )
        self.log_debug("Saving issue.")
        issue.save()

    def get_presentation_list(self, sources, dst_folder):
        """create a list of presentation objects"""
        self.log_debug("Getting list of presentation objects.")
        presentation_list = []
        for src in sources:
            try:
                try:
                    pres_object = PresentationObject(dst_folder, **src)
                except TypeError:
                    pres_object = PresentationObject(dst_folder, src)
                self.log_debug(
                    f"Loading {pres_object.loader.diag_type} diagnostic from {pres_object.loader.path}."
                )
                presentation_list.append(pres_object.create_dict())
            except PresentationException as msg:
                self.log_warning(f"Can not present diagnostic: {msg}")
        return presentation_list

    def get_template(self, context, template):
        """get Jinja2 template file"""
        search_path = [".", "templates"]
        if "_se_cmd_cwd" in context:
            search_path.extend(
                [
                    context["_se_cmd_cwd"],
                    Path(context["_se_cmd_cwd"]) / "templates",
                ]
            )
        self.log_debug(f"Search path for template: {search_path}")

        loader = jinja2.FileSystemLoader(search_path)
        environment = jinja2.Environment(loader=loader)
        for name, function in j2filters().items():
            environment.filters[name] = function
        return environment.get_template(template)

    def get_project_and_issue(self, gl, issue_subject):
        """Connect to Gitlab server, find and return correct issue and project based on user input"""

        project_id = 1982
        try:
            project = gl.projects.get(project_id)
        except (
            gitlab.GitlabAuthenticationError,
            requests.exceptions.ConnectionError,
        ) as e:
            msg = f"Could not log in to Gitlab server ({e})"
            self.log_error(msg)
            raise ScriptEngineTaskRunError()

        # Find issue or create if none exists
        issues = project.issues.list()
        issue = next((issue for issue in issues if issue.title == issue_subject), None)
        if issue is None:
            issue = project.issues.create({"title": issue_subject})
            issue.save()
        return project, issue


def create_anchor(anchor: str) -> str:
    """
    create correct anchors for Gitlab issue URLs
    """
    anchor = re.sub(r"[^a-zA-Z\s-]", "", anchor)
    anchor = re.sub(r"\s+(\-+\s*)?", "-", anchor)
    anchor = anchor.lower()
    return anchor

1	"""Presentation Task that uploads Data and Plots to the EC-Earth 4 Gitlab."""
2
3	import re	1✔
4	import urllib.parse	1✔
5	from pathlib import Path	1✔
6
7	import gitlab	1✔
8	import jinja2	1✔
9	import requests.exceptions	1✔
10	from scriptengine.exceptions import ScriptEngineTaskRunError	1✔
11	from scriptengine.jinja import filters as j2filters	1✔
12	from scriptengine.tasks.core import Task, timed_runner	1✔
13
14	from helpers.exceptions import PresentationException	1✔
15	from helpers.presentation_objects import PresentationObject	1✔
16
17
18	class Gitlab(Task):	1✔
19	"""Gitlab Presentation Task"""
20
21	_required_arguments = (	1✔
22	"src",
23	"local_dst",
24	"subject",
25	"template",
26	"api_key",
27	)
28
29	def __init__(self, arguments=None):	1✔
30	Gitlab.check_arguments(arguments)	1✔
31	super().__init__(arguments)	1✔
32
33	@timed_runner	1✔
34	def run(self, context):	1✔
35	sources = self.getarg("src", context)	1✔
36	dst_folder = self.getarg("local_dst", context)	1✔
37	issue_subject = self.getarg("subject", context, parse_yaml=False)	1✔
38	template_path = self.getarg("template", context)	1✔
39	key = self.getarg("api_key", context)	1✔
40	self.log_info(f"Create Gitlab issue '{issue_subject}'.")	1✔
41	self.log_debug(f"Template: {template_path}, Source File(s): {sources}")	1✔
42
43	presentation_list = self.get_presentation_list(sources, dst_folder)	1✔
44	gitlab_template = self.get_template(context, template_path)	1✔
45	gitlab_template.globals["urlencode"] = urllib.parse.quote	1✔
46
47	server_url = "https://git.smhi.se"	1✔
48	gl = gitlab.Gitlab(server_url, private_token=key)	1✔
49
50	self.log_debug("Connecting to Gitlab.")	1✔
51
52	project, issue = self.get_project_and_issue(gl, issue_subject)	1✔
53
54	self.log_debug("Uploading attachments.")	1✔
55	issue.uploads = []	1✔
56	for item in presentation_list:	1✔
NEW 57	if item["presentation_type"] == "image":	×
NEW 58	file_name = Path(item["path"]).name	×
NEW 59	upload = project.upload(file_name, filepath=f"{dst_folder}/{file_name}")	×
NEW 60	item["image_link"] = upload[	×
61	"markdown"
62	] # string of the form "![]()" generated by Gitlab
63
64	# render the template and add as description
65	self.log_debug("Updating the issue description.")	1✔
66	issue.description = gitlab_template.render(	1✔
67	presentation_list=presentation_list,
68	issue_url=issue.web_url,
69	create_anchor=create_anchor,
70	)
71	self.log_debug("Saving issue.")	1✔
72	issue.save()	1✔
73
74	def get_presentation_list(self, sources, dst_folder):	1✔
75	"""create a list of presentation objects"""
76	self.log_debug("Getting list of presentation objects.")	1✔
77	presentation_list = []	1✔
78	for src in sources:	1✔
79	try:	1✔
80	try:	1✔
81	pres_object = PresentationObject(dst_folder, **src)	1✔
82	except TypeError:	1✔
83	pres_object = PresentationObject(dst_folder, src)	1✔
84	self.log_debug(	1✔
85	f"Loading {pres_object.loader.diag_type} diagnostic from {pres_object.loader.path}."
86	)
87	presentation_list.append(pres_object.create_dict())	1✔
88	except PresentationException as msg:	1✔
89	self.log_warning(f"Can not present diagnostic: {msg}")	1✔
90	return presentation_list	1✔
91
92	def get_template(self, context, template):	1✔
93	"""get Jinja2 template file"""
94	search_path = [".", "templates"]	1✔
95	if "_se_cmd_cwd" in context:	1✔
NEW 96	search_path.extend(	×
97	[
98	context["_se_cmd_cwd"],
99	Path(context["_se_cmd_cwd"]) / "templates",
100	]
101	)
102	self.log_debug(f"Search path for template: {search_path}")	1✔
103
104	loader = jinja2.FileSystemLoader(search_path)	1✔
105	environment = jinja2.Environment(loader=loader)	1✔
106	for name, function in j2filters().items():	1✔
107	environment.filters[name] = function	1✔
108	return environment.get_template(template)	1✔
109
110	def get_project_and_issue(self, gl, issue_subject):	1✔
111	"""Connect to Gitlab server, find and return correct issue and project based on user input"""
112
113	project_id = 1982	1✔
114	try:	1✔
115	project = gl.projects.get(project_id)	1✔
116	except (	1✔
117	gitlab.GitlabAuthenticationError,
118	requests.exceptions.ConnectionError,
119	) as e:
120	msg = f"Could not log in to Gitlab server ({e})"	1✔
121	self.log_error(msg)	1✔
122	raise ScriptEngineTaskRunError()	1✔
123
124	# Find issue or create if none exists
NEW 125	issues = project.issues.list()	×
NEW 126	issue = next((issue for issue in issues if issue.title == issue_subject), None)	×
NEW 127	if issue is None:	×
NEW 128	issue = project.issues.create({"title": issue_subject})	×
NEW 129	issue.save()	×
NEW 130	return project, issue	×
131
132
133	def create_anchor(anchor: str) -> str:	1✔
134	"""
135	create correct anchors for Gitlab issue URLs
136	"""
137	anchor = re.sub(r"[^a-zA-Z\s-]", "", anchor)	1✔
138	anchor = re.sub(r"\s+(\-+\s*)?", "-", anchor)	1✔
139	anchor = anchor.lower()	1✔
140	return anchor	1✔

uwefladrich / scriptengine-tasks-ecearth / 14519448990

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous