# Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. from collections import namedtuple from dataclasses import dataclass, field from typing import List from .constants import MAGIC_MANAGE_STRING from .errors import Context, JenkinsJobsException from .loc_loader import LocDict, LocString from .position import Pos from .formatter import enum_str_format_required_params, enum_str_format_param_defaults from .expander import Expander, expand_parameters from .defaults import Defaults from .dimensions import enum_dimensions_params, is_point_included @dataclass class JobViewData: """Expanded job or view data, with added source context. Fed into xml generator""" data: LocDict context: List[Context] = field(default_factory=list) @property def name(self): return self.data["name"] def with_context(self, message, pos): return JobViewData(self.data, [Context(message, pos), *self.context]) @dataclass class RootBase: """Base class for YAML root elements - job, view or template""" _defaults: dict _expander: Expander _keep_descriptions: bool _id: str name: str pos: Pos description: str defaults_name: str params: dict contents: dict @property def id(self): if self._id: return self._id else: return self.name @property def title(self): return str(self).capitalize() def _format_description(self, params): if self.description is None: defaults = self._pick_defaults(self.defaults_name) description = defaults.params.get("description") else: if type(self.description) is LocString: description = str(self.description) else: description = self.description if description is None and self._keep_descriptions: return {} expanded_desc = self._expander.expand(description, params) return {"description": (expanded_desc or "") + MAGIC_MANAGE_STRING} def _pick_defaults(self, name, merge_global=True): try: defaults = self._defaults[name] except KeyError: if name == "global": return Defaults.empty() raise JenkinsJobsException( f"{self.title} wants defaults {self.defaults_name!r}" " but it was never defined", pos=name.pos, ) if name == "global": return defaults if merge_global: return defaults.merged_with_global(self._pick_defaults("global")) else: return defaults class NonTemplateRootMixin: def top_level_generate_items(self): try: defaults = self._pick_defaults(self.defaults_name, merge_global=False) description = self._format_description(params={}) raw_data = self._as_dict() contents = self._expander.expand(raw_data, self.params) data = LocDict.merge( defaults.contents, contents, description, pos=self.pos, ) context = [Context(f"In {self}", self.pos)] yield JobViewData(data, context) except JenkinsJobsException as x: raise x.with_context(f"In {self}", pos=self.pos) def generate_items(self, defaults_name, params): # Do not produce jobs/views from under project - they are produced when # processed directly from roots, by top_level_generate_items. return [] class TemplateRootMixin: def generate_items(self, defaults_name, params): try: defaults = self._pick_defaults(defaults_name or self.defaults_name) item_params = LocDict.merge( defaults.params, self.params, params, {"template-name": self.name}, ) if self._id: item_params["id"] = self._id contents = LocDict.merge( defaults.contents, self._as_dict(), ) axes = list(enum_str_format_required_params(self.name, self.name.pos)) axes_defaults = dict(enum_str_format_param_defaults(self.name)) for dim_params in enum_dimensions_params(axes, item_params, axes_defaults): instance_params = LocDict.merge( item_params, dim_params, ) expanded_params = expand_parameters(self._expander, instance_params) if not is_point_included( exclude_list=expanded_params.get("exclude"), params=expanded_params, key_pos=expanded_params.key_pos.get("exclude"), ): continue description = self._format_description(expanded_params) expanded_contents = self._expander.expand(contents, expanded_params) data = LocDict.merge( expanded_contents, description, pos=self.pos, ) context = [Context(f"In {self}", self.pos)] yield JobViewData(data, context) except JenkinsJobsException as x: raise x.with_context(f"In {self}", pos=self.pos) class GroupBase: Spec = namedtuple("Spec", "name params pos") def __repr__(self): return f"<{self}>" @classmethod def _specs_from_list(cls, spec_list=None): if spec_list is None: return [] return [ cls._spec_from_dict(item, spec_list.value_pos[idx]) for idx, item in enumerate(spec_list) ] @classmethod def _spec_from_dict(cls, d, pos): if isinstance(d, (str, LocString)): return cls.Spec(d, params={}, pos=pos) if not isinstance(d, dict): raise JenkinsJobsException( "Job/view spec should name or dict," f" but is {type(d)} ({d!r}). Missing indent?", pos=pos, ) if len(d) != 1: raise JenkinsJobsException( "Job/view dict should be single-item," f" but have keys {list(d.keys())}. Missing indent?", pos=d.pos, ) name, params = next(iter(d.items())) if params is None: params = {} else: if not isinstance(params, dict): raise JenkinsJobsException( f"Job/view {name!r} params type should be dict," f" but is {params!r}.", pos=params.pos, ) return cls.Spec(name, params, pos) def _generate_items(self, root_dicts, spec_list, defaults_name, params): try: for spec in spec_list: item = self._pick_spec_item(root_dicts, spec) item_params = LocDict.merge( params, self.params, self._my_params, spec.params, ) for job_data in item.generate_items(defaults_name, item_params): yield ( job_data.with_context("Defined here", spec.pos).with_context( f"In {self}", self.pos ) ) except JenkinsJobsException as x: raise x.with_context(f"In {self}", self.pos) @property def _my_params(self): return {} def _pick_spec_item(self, root_dict_list, spec): for roots_dict in root_dict_list: try: return roots_dict[spec.name] except KeyError: pass raise JenkinsJobsException( f"Failed to find suitable job/view/template named '{spec.name}'", pos=spec.pos, ) @dataclass class Group(GroupBase): name: str pos: Pos specs: list # list[Spec] params: dict def generate_items(self, defaults_name, params): return self._generate_items(self._root_dicts, self.specs, defaults_name, params)