1# Copyright 2022 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15"""This file provides the Sample class which contains methods for converting
16
17DCL samples to KCC format
18"""
19from __future__ import annotations
20
21import os
22import re
23import ruamel.yaml
24import string
25
26from absl import logging
27from collections import Counter
28from collections.abc import Mapping
29from collections.abc import Set
30from datetime import date
31from typing import Any
32from typing import List
33
34import config
35import strings
36
37APACHE_LICENSE = f"""# Copyright {date.today().year} Google LLC
38#
39# Licensed under the Apache License, Version 2.0 (the "License");
40# you may not use this file except in compliance with the License.
41# You may obtain a copy of the License at
42#
43# http://www.apache.org/licenses/LICENSE-2.0
44#
45# Unless required by applicable law or agreed to in writing, software
46# distributed under the License is distributed on an "AS IS" BASIS,
47# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
48# See the License for the specific language governing permissions and
49# limitations under the License.
50
51"""
52
53
54def kcc_service_version_and_kind(kcc_config: Mapping[str, Any]):
55 """Return the service and kind from a kcc resource config."""
56 version_parts = kcc_config['apiVersion'].split('/')
57 version_parts[0] = version_parts[0].split('.')[0]
58 return (version_parts[0], version_parts[1], kcc_config['kind'].lower())
59
60
61class Sample:
62 """Corresponds to a sample yaml file in DCL and a directory in KCC for a resource."""
63
64 def __init__(self, service: str, dcl_dir: str, kcc_testdata_dir: str,
65 kcc_samples_dir: str, file_name: str):
66 """Load and convert the configs for a single sample."""
67 self.service = service
68 self.dcl_dir = dcl_dir
69 self.kcc_testdata_dir = kcc_testdata_dir
70 self.kcc_samples_dir = kcc_samples_dir
71 # Load sample file.
72 dcl_sample_file = f'{self.dcl_dir}/samples/{file_name}'
73 logging.info(f'Loading sample file:\n{dcl_sample_file}')
74 yaml = ruamel.yaml.YAML(typ='safe', pure=True)
75 with open(dcl_sample_file) as file_object:
76 self.dcl_config = yaml.load(file_object)
77 self.name = self.dcl_config.get('name')
78 dependencies = self.dcl_config.get('dependencies')
79 if dependencies:
80 self.dcl_dependencies = {
81 dep: config.Config(self.service, f'{self.dcl_dir}/{dep}')
82 for dep in dependencies
83 }
84 else:
85 self.dcl_dependencies = {}
86 self.dcl_create = config.Config(
87 self.service, f'{self.dcl_dir}/{self.dcl_config.get("resource")}')
88 updates = self.dcl_config.get('updates')
89 if updates:
90 if len(updates) > 1:
91 logging.warning(
92 f'{strings.colors.WARNING}Sample {self.name} has more than one update.{strings.colors.ENDC}'
93 )
94 self.dcl_update = config.Config(
95 self.service, f'{self.dcl_dir}/{updates[0].get("resource")}')
96 #TODO: Handle update dependencies which are not already in the create dependencies array.
97 else:
98 self.dcl_update = None
99 self.variables = self.dcl_config.get('variables', [])
100 # Replace variables in all fields.
101 self.replacements = {
102 '{{org_id}}':
103 '${TEST_ORG_ID}', #TODO(kcc-eng): Handle resources that expect long form reference for org id.
104 '{{project}}': '${projectId}',
105 '{{region}}': 'us-west2',
106 '{{zone}}': 'us-west2-a',
107 }
108 self.add_replacements_for_variables()
109 self.kcc_dependencies = []
110 self.kcc_create = {}
111 self.kcc_update = {}
112
113 def add_replacements_for_variables(self) -> None:
114 """Add the proper replacement to this sample's conversion map to convert each DCL variable to KCC format."""
115 for variable in self.variables:
116 variable_type = variable.get('type')
117 if variable_type == 'resource_name':
118 variable_name = variable.get('name')
119 dcl_var = '{{' + variable_name + '}}'
120 kcc_var = variable_name.replace('_', '-')
121 self.replacements[dcl_var] = f'{kcc_var}-${{uniqueId}}'
122
123 def convert(self) -> None:
124 """Convert all dependency configs to KCC format."""
125 converted_dependencies = set()
126 while len(converted_dependencies) < len(self.dcl_dependencies):
127 # Convert the first resource config that doesn't depend on any resources
128 # that haven't been converted.
129 for file_name, dependency in self.dcl_dependencies.items():
130 if file_name in converted_dependencies:
131 continue
132 if not (dependency.depends_on() - converted_dependencies):
133 converted_dependencies.add(file_name)
134 self.kcc_dependencies.append(
135 dependency.convert(self.dcl_dependencies, self.replacements))
136 break
137 else:
138 logging.error(
139 f'{strings.colors.FAIL}Could not find next dependency to apply.{strings.colors.ENDC}'
140 )
141 # Convert main resource to KCC format.
142 self.kcc_create = self.dcl_create.convert(self.dcl_dependencies,
143 self.replacements)
144 if self.dcl_update:
145 self.kcc_update = self.dcl_update.convert(self.dcl_dependencies,
146 self.replacements)
147
148 def write(self, is_only: bool) -> None:
149 """After converting samples write them to files."""
150 if is_only:
151 # No subdirectories should be used if there is only one test case.
152 kcc_testdata_path = self.kcc_testdata_dir
153 kcc_samples_path = self.kcc_samples_dir
154 else:
155 kcc_name = self.dcl_config.get('name').replace('_', '-')
156 kcc_short_name = kcc_name.replace('-', '')
157 kcc_testdata_path = f'{self.kcc_testdata_dir}/{kcc_short_name}'
158 kcc_samples_path = f'{self.kcc_samples_dir}/{kcc_name}'
159 # Ensure that testdata and samples paths exist.
160 for path in (kcc_testdata_path, kcc_samples_path):
161 cur_path = '/' if path[0] == '/' else ''
162 for step in path.split(os.sep):
163 if not step:
164 continue
165 cur_path = os.path.join(cur_path, step)
166 if not os.path.isdir(cur_path):
167 os.mkdir(cur_path)
168 self.write_testdata(kcc_testdata_path)
169 self.write_samples(kcc_samples_path,
170 '' if is_only else f'-{kcc_short_name}')
171
172 def total_kind_counts(self) -> Counter:
173 """Return a counter of how many samples have each kcc kind."""
174 counts = Counter()
175 for kcc_dependency in self.kcc_dependencies:
176 counts[kcc_dependency['kind'].lower()] += 1
177 return counts
178
179 def rename_replacements(
180 self, renamer: Callable[[str, int], str]) -> Mapping[str, str]:
181 """Return a map of current config names to new names using the given renaming
182
183 function which takes the kind of the current sample and the index of that
184 sample within its kind if there is more than one.
185 """
186 create_name = self.kcc_create.get('metadata', {}).get('name')
187 if create_name:
188 replacements = {create_name: renamer(self.kcc_create['kind'].lower(), 0)}
189 else:
190 replacements = {}
191 total_kind_counts = self.total_kind_counts()
192 kind_indices = Counter()
193 for kcc_dependency in self.kcc_dependencies:
194 service, _, kind = kcc_service_version_and_kind(kcc_dependency)
195 kcc_dependency_name = kcc_dependency.get('metadata', {}).get('name')
196 if not kcc_dependency_name:
197 continue
198 if total_kind_counts[kind] > 1:
199 kind_indices[kind] += 1
200 replacements[kcc_dependency_name] = renamer(kind, kind_indices[kind])
201 else:
202 replacements[kcc_dependency_name] = renamer(kind, 0)
203 return replacements
204
205 def write_testdata(self, kcc_testdata_path: str):
206 """Write testdata samples."""
207 # Count total number of dependencies of each kind to determine which ones
208 # need a number suffix.
209 replacements = self.rename_replacements(
210 lambda k, i: f'{k}-{i}-${{uniqueId}}' if i else f'{k}-${{uniqueId}}')
211 replacer = lambda s: strings.replace_map(replacements, s)
212 yaml = ruamel.yaml.YAML()
213 yaml.indent(mapping=2)
214 kcc_dependencies_file = f'{kcc_testdata_path}/dependencies.yaml'
215 if self.kcc_dependencies:
216 logging.info(f'Writing dependencies to:\n{kcc_dependencies_file}')
217 with open(kcc_dependencies_file, 'w') as file_object:
218 yaml.dump_all([
219 config.quote_spec(config.replace_strings(dep, replacer))
220 for dep in self.kcc_dependencies
221 ], file_object)
222 kcc_create_file = f'{kcc_testdata_path}/create.yaml'
223 logging.info(f'Writing create sample to:\n{kcc_create_file}')
224 with open(kcc_create_file, 'w') as file_object:
225 yaml.dump(
226 config.quote_spec(config.replace_strings(self.kcc_create, replacer)),
227 file_object)
228 if self.kcc_update:
229 kcc_update_file = f'{kcc_testdata_path}/update.yaml'
230 logging.info(f'Writing update sample to:\n{kcc_update_file}')
231 with open(kcc_update_file, 'w') as file_object:
232 yaml.dump(
233 config.quote_spec(
234 config.replace_strings(self.kcc_update, replacer)), file_object)
235
236 def write_samples(self, kcc_samples_path: str, name_suffix: str):
237 """Write config samples."""
238 _, kcc_create_version, kcc_create_kind = kcc_service_version_and_kind(
239 self.kcc_create)
240 replacements = self.rename_replacements(
241 lambda k, i: f'{kcc_create_kind}-dep{i}{name_suffix}'
242 if i else f'{kcc_create_kind}-dep{name_suffix}')
243 create_name = self.kcc_create.get('metadata', {}).get('name')
244 if create_name:
245 replacements[create_name] = f'{kcc_create_kind}-sample{name_suffix}'
246 replacements['${TEST_ORG_ID}'] = '${ORG_ID?}'
247 replacements['${projectId}'] = '${PROJECT_ID?}'
248 replacer = lambda s: strings.replace_map(replacements, s)
249 # Map file names to configs so that all configs of a given kind go in the
250 # same file.
251 file_names_to_kcc_configs = {}
252 for kcc_dependency in self.kcc_dependencies:
253 service, version, kind = kcc_service_version_and_kind(kcc_dependency)
254 file_name = f'{kcc_samples_path}/{service}_{version}_{kind}.yaml'
255 existing_configs = file_names_to_kcc_configs.get(file_name)
256 if existing_configs:
257 # There is already at least one dependency of this kind.
258 if isinstance(existing_configs, dict):
259 # There is already exactly one dependency of this kind.
260 existing_configs = [existing_configs]
261 file_names_to_kcc_configs[file_name] = existing_configs
262 existing_configs.append(config.prepare_sample(kcc_dependency, replacer))
263 else:
264 file_names_to_kcc_configs[file_name] = config.prepare_sample(
265 kcc_dependency, replacer)
266 file_names_to_kcc_configs[
267 f'{kcc_samples_path}/{self.service}_{kcc_create_version}_{kcc_create_kind}.yaml'] = config.prepare_sample(
268 self.kcc_create, replacer)
269 for file_name, kcc_config in file_names_to_kcc_configs.items():
270 logging.info(f'Writing sample to:\n{file_name}')
271 with open(file_name, 'w') as file_object:
272 file_object.write(APACHE_LICENSE)
273 yaml = ruamel.yaml.YAML()
274 yaml.indent(mapping=2)
275 if isinstance(kcc_config, list):
276 yaml.dump_all(kcc_config, file_object)
277 else:
278 yaml.dump(kcc_config, file_object)
View as plain text