255 lines
8.5 KiB
Python
255 lines
8.5 KiB
Python
# 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)
|