4864891028

Build Type

push

github

Committed by rcerven

Commit Message

pin urllib3 for compatibility

Run Details

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

98.34

/osbs/api.py

"""
Copyright (c) 2015-2022 Red Hat, Inc
All rights reserved.

This software may be modified and distributed under the terms
of the BSD license. See the LICENSE file for details.
"""
from __future__ import print_function, unicode_literals, absolute_import

from collections import namedtuple
import logging
import sys
import warnings
import yaml
from functools import wraps
from typing import Any, Dict
from string import Template

from osbs.build.user_params import (
    BuildUserParams,
    SourceContainerUserParams
)
from osbs.constants import (RELEASE_LABEL_FORMAT, VERSION_LABEL_FORBIDDEN_CHARS,
                            ISOLATED_RELEASE_FORMAT)
from osbs.tekton import Openshift, PipelineRun
from osbs.exceptions import (OsbsException, OsbsValidationException, OsbsResponseException)
from osbs.utils.labels import Labels
# import utils in this way, so that we can mock standalone functions with flexmock
from osbs import utils


def _load_pipeline_from_template(pipeline_run_path, substitutions):
    """Load pipeline run from template and apply substitutions"""
    with open(pipeline_run_path) as f:
        yaml_data = f.read()
    template = Template(yaml_data)
    return yaml.safe_load(template.safe_substitute(substitutions))


# Decorator for API methods.
def osbsapi(func):
    @wraps(func)
    def catch_exceptions(*args, **kwargs):
        if kwargs.pop("namespace", None):
            warnings.warn("OSBS.%s: the 'namespace' argument is no longer supported" %
                          func.__name__)
        try:
            return func(*args, **kwargs)
        except OsbsException:
            # Re-raise OsbsExceptions
            raise
        except Exception as ex:
            # Propogate flexmock errors immediately (used in test cases)
            if getattr(ex, '__module__', None) == 'flexmock':
                raise

            # Convert anything else to OsbsException

            # Python 3 has implicit exception chaining and enhanced
            # reporting, so you get the original traceback as well as
            # the one originating here.
            # For Python 2, let's do that explicitly.
            raise OsbsException(cause=ex, traceback=sys.exc_info()[2])

    return catch_exceptions


_REQUIRED_PARAM = object()

logger = logging.getLogger(__name__)

LogEntry = namedtuple('LogEntry', ['platform', 'line'])


class OSBS(object):

    _GIT_LABEL_KEYS = ('git-repo-name', 'git-branch', 'git-full-repo')
    _OLD_LABEL_KEYS = ('git-repo-name', 'git-branch')

    @osbsapi
    def __init__(self, openshift_configuration):
        """ """
        self.os_conf = openshift_configuration
        self.os = Openshift(openshift_api_url=self.os_conf.get_openshift_base_uri(),
                            openshift_oauth_url=self.os_conf.get_openshift_oauth_api_uri(),
                            k8s_api_url=self.os_conf.get_k8s_api_uri(),
                            verbose=self.os_conf.get_verbosity(),
                            username=self.os_conf.get_username(),
                            password=self.os_conf.get_password(),
                            use_kerberos=self.os_conf.get_use_kerberos(),
                            client_cert=self.os_conf.get_client_cert(),
                            client_key=self.os_conf.get_client_key(),
                            kerberos_keytab=self.os_conf.get_kerberos_keytab(),
                            kerberos_principal=self.os_conf.get_kerberos_principal(),
                            kerberos_ccache=self.os_conf.get_kerberos_ccache(),
                            use_auth=self.os_conf.get_use_auth(),
                            verify_ssl=self.os_conf.get_verify_ssl(),
                            token=self.os_conf.get_oauth2_token(),
                            namespace=self.os_conf.get_namespace())
        self._bm = None

    def _check_labels(self, repo_info):
        labels = repo_info.labels

        required_missing = False
        req_labels = {}
        # required labels which needs to have explicit value (not from env variable)
        explicit_labels = [Labels.LABEL_TYPE_NAME,
                           Labels.LABEL_TYPE_COMPONENT]
        # version label isn't used here, but is required label in Dockerfile
        # and is used and required for atomic reactor
        # if we don't catch error here, it will fail in atomic reactor later
        for label in [Labels.LABEL_TYPE_NAME,
                      Labels.LABEL_TYPE_COMPONENT,
                      Labels.LABEL_TYPE_VERSION]:
            try:
                _, req_labels[label] = labels.get_name_and_value(label)

                if label in explicit_labels and not req_labels[label]:
                    required_missing = True
                    logger.error("required label doesn't have explicit value in Dockerfile : %s",
                                 labels.get_name(label))
            except KeyError:
                required_missing = True
                logger.error("required label missing from Dockerfile : %s",
                             labels.get_name(label))

        try:
            _, release_value = labels.get_name_and_value(Labels.LABEL_TYPE_RELEASE)
            if release_value and not RELEASE_LABEL_FORMAT.match(release_value):
                logger.error("release label '%s' doesn't match regex : %s", release_value,
                             RELEASE_LABEL_FORMAT.pattern)
                raise OsbsValidationException("release label doesn't have proper format")
        except KeyError:
            pass

        try:
            _, version_value = labels.get_name_and_value(Labels.LABEL_TYPE_VERSION)
        # version doesn't exist
        except KeyError:
            pass
        else:
            if version_value:
                wrong_chars = \
                    [denied for denied in VERSION_LABEL_FORBIDDEN_CHARS if denied in version_value]

                if wrong_chars:
                    msg = "version label '{}' contains not allowed chars : '{}'".\
                        format(version_value, wrong_chars)
                    logger.error(msg)
                    raise OsbsValidationException(msg)

        if required_missing:
            raise OsbsValidationException("required label missing from Dockerfile")

        # Verify the name label meets requirements.
        # It is made up of slash-separated name components.
        #
        # When pulling an image, the first component of the name
        # pulled is interpreted as a registry name if it contains a
        # '.' character, and otherwise the configured registries are
        # queried in turn.
        #
        # Due to this, a name with '.' in its initial component will
        # be awkward to pull from a registry because the registry name
        # will have to be explicitly supplied, e.g. "docker pull
        # foo.bar/baz" will fail because the "foo.bar" registry cannot
        # be contacted.
        #
        # Avoid this awkwardness by forbidding '.' in the initial
        # component of the image name.
        name_components = req_labels[Labels.LABEL_TYPE_NAME].split('/', 1)
        if '.' in name_components[0]:
            raise OsbsValidationException("initial image name component "
                                          "must not contain '.'")

        return req_labels

    # Gives flexmock something to mock
    def get_user_params(self, component=None, req_labels=None, **kwargs):
        req_labels = req_labels or {}
        user_component = component or req_labels[Labels.LABEL_TYPE_COMPONENT]
        return BuildUserParams.make_params(build_conf=self.os_conf,
                                           component=user_component,
                                           name_label=req_labels[Labels.LABEL_TYPE_NAME],
                                           **kwargs)

    def _checks_for_isolated(self, user_params):
        if user_params.isolated:
            if not user_params.release:
                raise OsbsValidationException(
                    'The release parameter is required for isolated builds.')

            if not ISOLATED_RELEASE_FORMAT.match(user_params.release):
                raise OsbsValidationException(
                    'For isolated builds, the release value must be in the format: {}'
                    .format(ISOLATED_RELEASE_FORMAT.pattern))

    def _checks_for_flatpak(self, flatpak, repo_info):
        if flatpak and not repo_info.configuration.is_flatpak:
            raise OsbsException(
                "Flatpak build, "
                "but repository doesn't have a container.yaml with a flatpak: section")

        if not flatpak and repo_info.configuration.is_flatpak:
            raise OsbsException(
                "Not a flatpak build, "
                "but repository has a container.yaml with a flatpak: section")

    def _get_pipeline_template_substitutions(
            self, *, buildtime_limit: int = 0, user_params, pipeline_run_name) -> Dict[str, str]:
        """Return map of substitutions used by pipeline run template
        to construct pipeline run data

        These substitutions can be used in pipeline run template

        Substitutions:
          $osbs_configmap_name - name of configmap defined by user
          $osbs_namespace - namespace where pipeline runs
          $osbs_pipeline_run_name - name of pipeline run
          $osbs_user_params_json - user params in json format
        """
        return {
            'osbs_buildtime_limit': f'{buildtime_limit}s',
            'osbs_configmap_name': user_params.reactor_config_map,
            'osbs_namespace': self.os_conf.get_namespace(),
            'osbs_pipeline_run_name': pipeline_run_name,
            'osbs_user_params_json': user_params.to_json(),
        }

    def _get_binary_container_pipeline_name(self, user_params):
        pipeline_run_postfix = utils.generate_random_postfix()
        pipeline_run_name = user_params.name

        if user_params.isolated:
            pipeline_run_name = f'isolated-{pipeline_run_postfix}'

        elif user_params.scratch:
            pipeline_run_name = f'scratch-{pipeline_run_postfix}'
        return pipeline_run_name

    def _get_binary_container_pipeline_data(self, *, buildtime_limit, user_params,
                                            pipeline_run_name):
        pipeline_run_path = self.os_conf.get_pipeline_run_path()

        substitutions = self._get_pipeline_template_substitutions(
            buildtime_limit=buildtime_limit,
            user_params=user_params,
            pipeline_run_name=pipeline_run_name
        )
        pipeline_run_data = _load_pipeline_from_template(pipeline_run_path, substitutions)

        return pipeline_run_data

    @osbsapi
    def create_binary_container_build(self, **kwargs):
        return self.create_binary_container_pipeline_run(**kwargs)

    @osbsapi
    def create_binary_container_pipeline_run(self,
                                             git_uri=_REQUIRED_PARAM, git_ref=_REQUIRED_PARAM,
                                             git_branch=_REQUIRED_PARAM,
                                             component=None,
                                             flatpak=None,
                                             git_commit_depth=None,
                                             isolated=None,
                                             koji_task_id=None,
                                             target=None,
                                             operator_csv_modifications_url=None,
                                             **kwargs):

        required_params = {"git_uri": git_uri, "git_ref": git_ref, "git_branch": git_branch}
        missing_params = []
        for param_name, param_arg in required_params.items():
            if param_arg is _REQUIRED_PARAM or not param_arg:
                missing_params.append(param_name)
        if missing_params:
            raise OsbsException('required parameter {} missing'.format(", ".join(missing_params)))

        if operator_csv_modifications_url and not isolated:
            raise OsbsException('Only isolated build can update operator CSV metadata')

        repo_info = utils.get_repo_info(git_uri, git_ref, git_branch=git_branch,
                                        depth=git_commit_depth)

        self._checks_for_flatpak(flatpak, repo_info)

        default_buildtime_limit = kwargs.get('default_buildtime_limit')
        max_buildtime_limit = kwargs.get('max_buildtime_limit')

        if repo_info.configuration.buildtime_limit > max_buildtime_limit:
            raise OsbsException(f'Build limit cannot be more than {max_buildtime_limit} '
                                f'and is set to {repo_info.configuration.buildtime_limit}')

        if repo_info.configuration.buildtime_limit == 0:
            logger.info('No build time limit is set, using default limit: %s',
                        default_buildtime_limit)
            repo_info.configuration.buildtime_limit = default_buildtime_limit

        req_labels = self._check_labels(repo_info)

        user_params = self.get_user_params(
            base_image=repo_info.base_image,
            component=component,
            flatpak=flatpak,
            isolated=isolated,
            koji_target=target,
            koji_task_id=koji_task_id,
            req_labels=req_labels,
            repo_info=repo_info,
            operator_csv_modifications_url=operator_csv_modifications_url,
            **kwargs)

        self._checks_for_isolated(user_params)

        pipeline_run_name = self._get_binary_container_pipeline_name(user_params)
        pipeline_run_data = self._get_binary_container_pipeline_data(
            buildtime_limit=repo_info.configuration.buildtime_limit,
            user_params=user_params,
            pipeline_run_name=pipeline_run_name)

        logger.info("creating binary container image pipeline run: %s", pipeline_run_name)

        pipeline_run = PipelineRun(self.os, pipeline_run_name, pipeline_run_data)

        try:
            logger.info("pipeline run created: %s", pipeline_run.start_pipeline_run())
        except OsbsResponseException:
            logger.error("failed to create pipeline run %s", pipeline_run_name)
            raise

        return pipeline_run

    def _get_source_container_pipeline_name(self):
        pipeline_run_postfix = utils.generate_random_postfix()
        pipeline_run_name = f'source-{pipeline_run_postfix}'
        return pipeline_run_name

    def _get_source_container_pipeline_data(self, *, user_params, pipeline_run_name):

        pipeline_run_path = self.os_conf.get_pipeline_run_path()

        substitutions = self._get_pipeline_template_substitutions(
            user_params=user_params,
            pipeline_run_name=pipeline_run_name,
        )
        pipeline_run_data = _load_pipeline_from_template(pipeline_run_path, substitutions)

        return pipeline_run_data

    @osbsapi
    def create_source_container_build(self, **kwargs):
        return self.create_source_container_pipeline_run(**kwargs)

    @osbsapi
    def create_source_container_pipeline_run(self,
                                             component=None,
                                             koji_task_id=None,
                                             target=None,
                                             **kwargs):
        """
        Take input args, create source pipeline run

        :return: instance of PiplelineRun
        """
        error_messages = []
        # most likely can be removed, source build should get component name
        # from binary build OSBS2 TBD
        if not component:
            error_messages.append("required argument 'component' can't be empty")
        if error_messages:
            raise OsbsValidationException(", ".join(error_messages))

        user_params = SourceContainerUserParams.make_params(
            build_conf=self.os_conf,
            component=component,
            koji_target=target,
            koji_task_id=koji_task_id,
            **kwargs
        )

        pipeline_run_name = self._get_source_container_pipeline_name()
        pipeline_run_data = self._get_source_container_pipeline_data(
            user_params=user_params,
            pipeline_run_name=pipeline_run_name,
        )

        logger.info("creating source container image pipeline run: %s", pipeline_run_name)

        pipeline_run = PipelineRun(self.os, pipeline_run_name, pipeline_run_data)

        try:
            logger.info("pipeline run created: %s", pipeline_run.start_pipeline_run())
        except OsbsResponseException:
            logger.error("failed to create pipeline run %s", pipeline_run_name)
            raise

        return pipeline_run

    @osbsapi
    def get_build_name(self, build_response: PipelineRun):
        return build_response.pipeline_run_name

    @osbsapi
    def get_build(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.get_info()

    @osbsapi
    def get_final_platforms(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.get_final_platforms()

    @osbsapi
    def get_build_reason(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.status_reason

    @osbsapi
    def build_has_succeeded(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.has_succeeded()

    @osbsapi
    def build_not_finished(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.has_not_finished()

    @osbsapi
    def wait_for_build_to_finish(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.wait_for_finish()

    @osbsapi
    def build_was_cancelled(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.was_cancelled()

    @osbsapi
    def build_has_any_failed_tasks(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.any_task_failed()

    @osbsapi
    def build_has_any_cancelled_tasks(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.any_task_was_cancelled()

    @osbsapi
    def cancel_build(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.cancel_pipeline_run()

    @osbsapi
    def remove_build(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.remove_pipeline_run()

    @osbsapi
    def get_build_logs(self, build_name, follow=False, wait=False):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.get_logs(follow=follow, wait=wait)

    @osbsapi
    def get_build_error_message(self, build_name):
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.get_error_message()

    @osbsapi
    def get_build_results(self, build_name) -> Dict[str, Any]:
        """Fetch the pipelineResults for this build."""
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.pipeline_results

    @osbsapi
    def get_task_results(self, build_name) -> Dict[str, Any]:
        """Fetch tasks results for this build."""
        pipeline_run = PipelineRun(self.os, build_name)
        return pipeline_run.get_task_results()

1	"""
2	Copyright (c) 2015-2022 Red Hat, Inc
3	All rights reserved.
4
5	This software may be modified and distributed under the terms
6	of the BSD license. See the LICENSE file for details.
7	"""
8	from __future__ import print_function, unicode_literals, absolute_import	1✔
9
10	from collections import namedtuple	1✔
11	import logging	1✔
12	import sys	1✔
13	import warnings	1✔
14	import yaml	1✔
15	from functools import wraps	1✔
16	from typing import Any, Dict	1✔
17	from string import Template	1✔
18
19	from osbs.build.user_params import (	1✔
20	BuildUserParams,
21	SourceContainerUserParams
22	)
23	from osbs.constants import (RELEASE_LABEL_FORMAT, VERSION_LABEL_FORBIDDEN_CHARS,	1✔
24	ISOLATED_RELEASE_FORMAT)
25	from osbs.tekton import Openshift, PipelineRun	1✔
26	from osbs.exceptions import (OsbsException, OsbsValidationException, OsbsResponseException)	1✔
27	from osbs.utils.labels import Labels	1✔
28	# import utils in this way, so that we can mock standalone functions with flexmock
29	from osbs import utils	1✔
30
31
32	def _load_pipeline_from_template(pipeline_run_path, substitutions):	1✔
33	"""Load pipeline run from template and apply substitutions"""
34	with open(pipeline_run_path) as f:	1✔
35	yaml_data = f.read()	1✔
36	template = Template(yaml_data)	1✔
37	return yaml.safe_load(template.safe_substitute(substitutions))	1✔
38
39
40	# Decorator for API methods.
41	def osbsapi(func):	1✔
42	@wraps(func)	1✔
43	def catch_exceptions(args, *kwargs):	1✔
44	if kwargs.pop("namespace", None):	1✔
45	warnings.warn("OSBS.%s: the 'namespace' argument is no longer supported" %	×
46	func.__name__)
47	try:	1✔
48	return func(args, *kwargs)	1✔
49	except OsbsException:	1✔
50	# Re-raise OsbsExceptions
51	raise	1✔
52	except Exception as ex:	1✔
53	# Propogate flexmock errors immediately (used in test cases)
54	if getattr(ex, '__module__', None) == 'flexmock':	1✔
55	raise	×
56
57	# Convert anything else to OsbsException
58
59	# Python 3 has implicit exception chaining and enhanced
60	# reporting, so you get the original traceback as well as
61	# the one originating here.
62	# For Python 2, let's do that explicitly.
63	raise OsbsException(cause=ex, traceback=sys.exc_info()[2])	1✔
64
65	return catch_exceptions	1✔
66
67
68	_REQUIRED_PARAM = object()	1✔
69
70	logger = logging.getLogger(__name__)	1✔
71
72	LogEntry = namedtuple('LogEntry', ['platform', 'line'])	1✔
73
74
75	class OSBS(object):	1✔
76
77	_GIT_LABEL_KEYS = ('git-repo-name', 'git-branch', 'git-full-repo')	1✔
78	_OLD_LABEL_KEYS = ('git-repo-name', 'git-branch')	1✔
79
80	@osbsapi	1✔
81	def __init__(self, openshift_configuration):	1✔
82	""" """
83	self.os_conf = openshift_configuration	1✔
84	self.os = Openshift(openshift_api_url=self.os_conf.get_openshift_base_uri(),	1✔
85	openshift_oauth_url=self.os_conf.get_openshift_oauth_api_uri(),
86	k8s_api_url=self.os_conf.get_k8s_api_uri(),
87	verbose=self.os_conf.get_verbosity(),
88	username=self.os_conf.get_username(),
89	password=self.os_conf.get_password(),
90	use_kerberos=self.os_conf.get_use_kerberos(),
91	client_cert=self.os_conf.get_client_cert(),
92	client_key=self.os_conf.get_client_key(),
93	kerberos_keytab=self.os_conf.get_kerberos_keytab(),
94	kerberos_principal=self.os_conf.get_kerberos_principal(),
95	kerberos_ccache=self.os_conf.get_kerberos_ccache(),
96	use_auth=self.os_conf.get_use_auth(),
97	verify_ssl=self.os_conf.get_verify_ssl(),
98	token=self.os_conf.get_oauth2_token(),
99	namespace=self.os_conf.get_namespace())
100	self._bm = None	1✔
101
102	def _check_labels(self, repo_info):	1✔
103	labels = repo_info.labels	1✔
104
105	required_missing = False	1✔
106	req_labels = {}	1✔
107	# required labels which needs to have explicit value (not from env variable)
108	explicit_labels = [Labels.LABEL_TYPE_NAME,	1✔
109	Labels.LABEL_TYPE_COMPONENT]
110	# version label isn't used here, but is required label in Dockerfile
111	# and is used and required for atomic reactor
112	# if we don't catch error here, it will fail in atomic reactor later
113	for label in [Labels.LABEL_TYPE_NAME,	1✔
114	Labels.LABEL_TYPE_COMPONENT,
115	Labels.LABEL_TYPE_VERSION]:
116	try:	1✔
117	_, req_labels[label] = labels.get_name_and_value(label)	1✔
118
119	if label in explicit_labels and not req_labels[label]:	1✔
120	required_missing = True	1✔
121	logger.error("required label doesn't have explicit value in Dockerfile : %s",	1✔
122	labels.get_name(label))
123	except KeyError:	1✔
124	required_missing = True	1✔
125	logger.error("required label missing from Dockerfile : %s",	1✔
126	labels.get_name(label))
127
128	try:	1✔
129	_, release_value = labels.get_name_and_value(Labels.LABEL_TYPE_RELEASE)	1✔
130	if release_value and not RELEASE_LABEL_FORMAT.match(release_value):	1✔
131	logger.error("release label '%s' doesn't match regex : %s", release_value,	1✔
132	RELEASE_LABEL_FORMAT.pattern)
133	raise OsbsValidationException("release label doesn't have proper format")	1✔
134	except KeyError:	1✔
135	pass	1✔
136
137	try:	1✔
138	_, version_value = labels.get_name_and_value(Labels.LABEL_TYPE_VERSION)	1✔
139	# version doesn't exist
140	except KeyError:	1✔
141	pass	1✔
142	else:
143	if version_value:	1✔
144	wrong_chars = \	1✔
145	[denied for denied in VERSION_LABEL_FORBIDDEN_CHARS if denied in version_value]
146
147	if wrong_chars:	1✔
148	msg = "version label '{}' contains not allowed chars : '{}'".\	1✔
149	format(version_value, wrong_chars)
150	logger.error(msg)	1✔
151	raise OsbsValidationException(msg)	1✔
152
153	if required_missing:	1✔
154	raise OsbsValidationException("required label missing from Dockerfile")	1✔
155
156	# Verify the name label meets requirements.
157	# It is made up of slash-separated name components.
158	#
159	# When pulling an image, the first component of the name
160	# pulled is interpreted as a registry name if it contains a
161	# '.' character, and otherwise the configured registries are
162	# queried in turn.
163	#
164	# Due to this, a name with '.' in its initial component will
165	# be awkward to pull from a registry because the registry name
166	# will have to be explicitly supplied, e.g. "docker pull
167	# foo.bar/baz" will fail because the "foo.bar" registry cannot
168	# be contacted.
169	#
170	# Avoid this awkwardness by forbidding '.' in the initial
171	# component of the image name.
172	name_components = req_labels[Labels.LABEL_TYPE_NAME].split('/', 1)	1✔
173	if '.' in name_components[0]:	1✔
174	raise OsbsValidationException("initial image name component "	1✔
175	"must not contain '.'")
176
177	return req_labels	1✔
178
179	# Gives flexmock something to mock
180	def get_user_params(self, component=None, req_labels=None, **kwargs):	1✔
181	req_labels = req_labels or {}	1✔
182	user_component = component or req_labels[Labels.LABEL_TYPE_COMPONENT]	1✔
183	return BuildUserParams.make_params(build_conf=self.os_conf,	1✔
184	component=user_component,
185	name_label=req_labels[Labels.LABEL_TYPE_NAME],
186	**kwargs)
187
188	def _checks_for_isolated(self, user_params):	1✔
189	if user_params.isolated:	1✔
190	if not user_params.release:	1✔
191	raise OsbsValidationException(	1✔
192	'The release parameter is required for isolated builds.')
193
194	if not ISOLATED_RELEASE_FORMAT.match(user_params.release):	1✔
195	raise OsbsValidationException(	1✔
196	'For isolated builds, the release value must be in the format: {}'
197	.format(ISOLATED_RELEASE_FORMAT.pattern))
198
199	def _checks_for_flatpak(self, flatpak, repo_info):	1✔
200	if flatpak and not repo_info.configuration.is_flatpak:	1✔
201	raise OsbsException(	1✔
202	"Flatpak build, "
203	"but repository doesn't have a container.yaml with a flatpak: section")
204
205	if not flatpak and repo_info.configuration.is_flatpak:	1✔
206	raise OsbsException(	1✔
207	"Not a flatpak build, "
208	"but repository has a container.yaml with a flatpak: section")
209
210	def _get_pipeline_template_substitutions(	1✔
211	self, *, buildtime_limit: int = 0, user_params, pipeline_run_name) -> Dict[str, str]:
212	"""Return map of substitutions used by pipeline run template
213	to construct pipeline run data
214
215	These substitutions can be used in pipeline run template
216
217	Substitutions:
218	$osbs_configmap_name - name of configmap defined by user
219	$osbs_namespace - namespace where pipeline runs
220	$osbs_pipeline_run_name - name of pipeline run
221	$osbs_user_params_json - user params in json format
222	"""
223	return {	1✔
224	'osbs_buildtime_limit': f'{buildtime_limit}s',
225	'osbs_configmap_name': user_params.reactor_config_map,
226	'osbs_namespace': self.os_conf.get_namespace(),
227	'osbs_pipeline_run_name': pipeline_run_name,
228	'osbs_user_params_json': user_params.to_json(),
229	}
230
231	def _get_binary_container_pipeline_name(self, user_params):	1✔
232	pipeline_run_postfix = utils.generate_random_postfix()	1✔
233	pipeline_run_name = user_params.name	1✔
234
235	if user_params.isolated:	1✔
236	pipeline_run_name = f'isolated-{pipeline_run_postfix}'	1✔
237
238	elif user_params.scratch:	1✔
239	pipeline_run_name = f'scratch-{pipeline_run_postfix}'	1✔
240	return pipeline_run_name	1✔
241
242	def _get_binary_container_pipeline_data(self, *, buildtime_limit, user_params,	1✔
243	pipeline_run_name):
244	pipeline_run_path = self.os_conf.get_pipeline_run_path()	1✔
245
246	substitutions = self._get_pipeline_template_substitutions(	1✔
247	buildtime_limit=buildtime_limit,
248	user_params=user_params,
249	pipeline_run_name=pipeline_run_name
250	)
251	pipeline_run_data = _load_pipeline_from_template(pipeline_run_path, substitutions)	1✔
252
253	return pipeline_run_data	1✔
254
255	@osbsapi	1✔
256	def create_binary_container_build(self, **kwargs):	1✔
257	return self.create_binary_container_pipeline_run(**kwargs)	1✔
258
259	@osbsapi	1✔
260	def create_binary_container_pipeline_run(self,	1✔
261	git_uri=_REQUIRED_PARAM, git_ref=_REQUIRED_PARAM,
262	git_branch=_REQUIRED_PARAM,
263	component=None,
264	flatpak=None,
265	git_commit_depth=None,
266	isolated=None,
267	koji_task_id=None,
268	target=None,
269	operator_csv_modifications_url=None,
270	**kwargs):
271
272	required_params = {"git_uri": git_uri, "git_ref": git_ref, "git_branch": git_branch}	1✔
273	missing_params = []	1✔
274	for param_name, param_arg in required_params.items():	1✔
275	if param_arg is _REQUIRED_PARAM or not param_arg:	1✔
276	missing_params.append(param_name)	1✔
277	if missing_params:	1✔
278	raise OsbsException('required parameter {} missing'.format(", ".join(missing_params)))	1✔
279
280	if operator_csv_modifications_url and not isolated:	1✔
281	raise OsbsException('Only isolated build can update operator CSV metadata')	1✔
282
283	repo_info = utils.get_repo_info(git_uri, git_ref, git_branch=git_branch,	1✔
284	depth=git_commit_depth)
285
286	self._checks_for_flatpak(flatpak, repo_info)	1✔
287
288	default_buildtime_limit = kwargs.get('default_buildtime_limit')	1✔
289	max_buildtime_limit = kwargs.get('max_buildtime_limit')	1✔
290
291	if repo_info.configuration.buildtime_limit > max_buildtime_limit:	1✔
292	raise OsbsException(f'Build limit cannot be more than {max_buildtime_limit} '	1✔
293	f'and is set to {repo_info.configuration.buildtime_limit}')
294
295	if repo_info.configuration.buildtime_limit == 0:	1✔
296	logger.info('No build time limit is set, using default limit: %s',	1✔
297	default_buildtime_limit)
298	repo_info.configuration.buildtime_limit = default_buildtime_limit	1✔
299
300	req_labels = self._check_labels(repo_info)	1✔
301
302	user_params = self.get_user_params(	1✔
303	base_image=repo_info.base_image,
304	component=component,
305	flatpak=flatpak,
306	isolated=isolated,
307	koji_target=target,
308	koji_task_id=koji_task_id,
309	req_labels=req_labels,
310	repo_info=repo_info,
311	operator_csv_modifications_url=operator_csv_modifications_url,
312	**kwargs)
313
314	self._checks_for_isolated(user_params)	1✔
315
316	pipeline_run_name = self._get_binary_container_pipeline_name(user_params)	1✔
317	pipeline_run_data = self._get_binary_container_pipeline_data(	1✔
318	buildtime_limit=repo_info.configuration.buildtime_limit,
319	user_params=user_params,
320	pipeline_run_name=pipeline_run_name)
321
322	logger.info("creating binary container image pipeline run: %s", pipeline_run_name)	1✔
323
324	pipeline_run = PipelineRun(self.os, pipeline_run_name, pipeline_run_data)	1✔
325
326	try:	1✔
327	logger.info("pipeline run created: %s", pipeline_run.start_pipeline_run())	1✔
328	except OsbsResponseException:	1✔
329	logger.error("failed to create pipeline run %s", pipeline_run_name)	1✔
330	raise	1✔
331
332	return pipeline_run	1✔
333
334	def _get_source_container_pipeline_name(self):	1✔
335	pipeline_run_postfix = utils.generate_random_postfix()	1✔
336	pipeline_run_name = f'source-{pipeline_run_postfix}'	1✔
337	return pipeline_run_name	1✔
338
339	def _get_source_container_pipeline_data(self, *, user_params, pipeline_run_name):	1✔
340
341	pipeline_run_path = self.os_conf.get_pipeline_run_path()	1✔
342
343	substitutions = self._get_pipeline_template_substitutions(	1✔
344	user_params=user_params,
345	pipeline_run_name=pipeline_run_name,
346	)
347	pipeline_run_data = _load_pipeline_from_template(pipeline_run_path, substitutions)	1✔
348
349	return pipeline_run_data	1✔
350
351	@osbsapi	1✔
352	def create_source_container_build(self, **kwargs):	1✔
353	return self.create_source_container_pipeline_run(**kwargs)	1✔
354
355	@osbsapi	1✔
356	def create_source_container_pipeline_run(self,	1✔
357	component=None,
358	koji_task_id=None,
359	target=None,
360	**kwargs):
361	"""
362	Take input args, create source pipeline run
363
364	:return: instance of PiplelineRun
365	"""
366	error_messages = []	1✔
367	# most likely can be removed, source build should get component name
368	# from binary build OSBS2 TBD
369	if not component:	1✔
370	error_messages.append("required argument 'component' can't be empty")	1✔
371	if error_messages:	1✔
372	raise OsbsValidationException(", ".join(error_messages))	1✔
373
374	user_params = SourceContainerUserParams.make_params(	1✔
375	build_conf=self.os_conf,
376	component=component,
377	koji_target=target,
378	koji_task_id=koji_task_id,
379	**kwargs
380	)
381
382	pipeline_run_name = self._get_source_container_pipeline_name()	1✔
383	pipeline_run_data = self._get_source_container_pipeline_data(	1✔
384	user_params=user_params,
385	pipeline_run_name=pipeline_run_name,
386	)
387
388	logger.info("creating source container image pipeline run: %s", pipeline_run_name)	1✔
389
390	pipeline_run = PipelineRun(self.os, pipeline_run_name, pipeline_run_data)	1✔
391
392	try:	1✔
393	logger.info("pipeline run created: %s", pipeline_run.start_pipeline_run())	1✔
394	except OsbsResponseException:	1✔
395	logger.error("failed to create pipeline run %s", pipeline_run_name)	1✔
396	raise	1✔
397
398	return pipeline_run	1✔
399
400	@osbsapi	1✔
401	def get_build_name(self, build_response: PipelineRun):	1✔
402	return build_response.pipeline_run_name	1✔
403
404	@osbsapi	1✔
405	def get_build(self, build_name):	1✔
406	pipeline_run = PipelineRun(self.os, build_name)	1✔
407	return pipeline_run.get_info()	1✔
408
409	@osbsapi	1✔
410	def get_final_platforms(self, build_name):	1✔
411	pipeline_run = PipelineRun(self.os, build_name)	1✔
412	return pipeline_run.get_final_platforms()	1✔
413
414	@osbsapi	1✔
415	def get_build_reason(self, build_name):	1✔
416	pipeline_run = PipelineRun(self.os, build_name)	1✔
417	return pipeline_run.status_reason	1✔
418
419	@osbsapi	1✔
420	def build_has_succeeded(self, build_name):	1✔
421	pipeline_run = PipelineRun(self.os, build_name)	1✔
422	return pipeline_run.has_succeeded()	1✔
423
424	@osbsapi	1✔
425	def build_not_finished(self, build_name):	1✔
426	pipeline_run = PipelineRun(self.os, build_name)	1✔
427	return pipeline_run.has_not_finished()	1✔
428
429	@osbsapi	1✔
430	def wait_for_build_to_finish(self, build_name):	1✔
431	pipeline_run = PipelineRun(self.os, build_name)	×
432	return pipeline_run.wait_for_finish()	×
433
434	@osbsapi	1✔
435	def build_was_cancelled(self, build_name):	1✔
436	pipeline_run = PipelineRun(self.os, build_name)	1✔
437	return pipeline_run.was_cancelled()	1✔
438
439	@osbsapi	1✔
440	def build_has_any_failed_tasks(self, build_name):	1✔
441	pipeline_run = PipelineRun(self.os, build_name)	1✔
442	return pipeline_run.any_task_failed()	1✔
443
444	@osbsapi	1✔
445	def build_has_any_cancelled_tasks(self, build_name):	1✔
446	pipeline_run = PipelineRun(self.os, build_name)	1✔
447	return pipeline_run.any_task_was_cancelled()	1✔
448
449	@osbsapi	1✔
450	def cancel_build(self, build_name):	1✔
451	pipeline_run = PipelineRun(self.os, build_name)	1✔
452	return pipeline_run.cancel_pipeline_run()	1✔
453
454	@osbsapi	1✔
455	def remove_build(self, build_name):	1✔
456	pipeline_run = PipelineRun(self.os, build_name)	1✔
457	return pipeline_run.remove_pipeline_run()	1✔
458
459	@osbsapi	1✔
460	def get_build_logs(self, build_name, follow=False, wait=False):	1✔
461	pipeline_run = PipelineRun(self.os, build_name)	1✔
462	return pipeline_run.get_logs(follow=follow, wait=wait)	1✔
463
464	@osbsapi	1✔
465	def get_build_error_message(self, build_name):	1✔
466	pipeline_run = PipelineRun(self.os, build_name)	1✔
467	return pipeline_run.get_error_message()	1✔
468
469	@osbsapi	1✔
470	def get_build_results(self, build_name) -> Dict[str, Any]:	1✔
471	"""Fetch the pipelineResults for this build."""
472	pipeline_run = PipelineRun(self.os, build_name)	1✔
473	return pipeline_run.pipeline_results	1✔
474
475	@osbsapi	1✔
476	def get_task_results(self, build_name) -> Dict[str, Any]:	1✔
477	"""Fetch tasks results for this build."""
478	pipeline_run = PipelineRun(self.os, build_name)	1✔
479	return pipeline_run.get_task_results()	1✔

containerbuildsystem / osbs-client / 4864891028

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