Coverage for sources/copiertv/engine.py: 99%
91 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-10 00:24 +0000
1# vim: set filetype=python fileencoding=utf-8:
2# -*- coding: utf-8 -*-
4#============================================================================#
5# #
6# Licensed under the Apache License, Version 2.0 (the "License"); #
7# you may not use this file except in compliance with the License. #
8# You may obtain a copy of the License at #
9# #
10# http://www.apache.org/licenses/LICENSE-2.0 #
11# #
12# Unless required by applicable law or agreed to in writing, software #
13# distributed under the License is distributed on an "AS IS" BASIS, #
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
15# See the License for the specific language governing permissions and #
16# limitations under the License. #
17# #
18#============================================================================#
21''' Core template validation engine. '''
24from . import __
25from . import configuration as _config
26from . import exceptions as _exceptions
29_scribe = __.logging.getLogger( __name__ )
32class ValidationResult( __.immut.DataclassObject ):
33 ''' Result of a template validation run. '''
35 variant: str
36 temporary_directory: __.Path
37 items_attempted: int
38 items_generated: int
39 preserved: bool
41 def render_as_markdown( self ) -> tuple[ str, ... ]:
42 ''' Renders validation result as Markdown lines. '''
43 lines = [
44 f"\u2705 Validation complete for "
45 f"'{self.variant}' variant:",
46 f" * Temporary Directory: "
47 f"{self.temporary_directory}",
48 f" * Items: {self.items_generated}"
49 f"/{self.items_attempted} generated",
50 ]
51 if self.preserved:
52 lines.append(
53 f" * \U0001f4c1 Files preserved for inspection"
54 f" at: {self.temporary_directory}" )
55 else:
56 lines.append(
57 ' * \U0001f5d1\ufe0f Temporary files cleaned up' )
58 return tuple( lines )
61def _acquire_answers_file(
62 path: __.Path,
63) -> dict[ str, __.typx.Any ]:
64 ''' Reads a YAML answers file. '''
65 import yaml
66 try: content = path.read_text( encoding = 'utf-8' )
67 except ( OSError, IOError ) as exception: # pragma: no cover
68 raise _exceptions.FileOperationFailure(
69 path, 'read answers file' ) from exception
70 try: data: dict[ str, __.typx.Any ] = yaml.safe_load( content )
71 except Exception as exception: # pragma: no cover
72 raise _exceptions.DataInvalidity(
73 path, 'Invalid YAML' ) from exception
74 if not isinstance( data, dict ): # pragma: no cover
75 raise _exceptions.DataInvalidity(
76 path, 'Answers file must be a mapping' )
77 return data
80def _execute_command(
81 command: tuple[ str, ... ],
82 working_directory: __.Path,
83 temporary_directory: __.Path,
84 preserve: bool,
85 executor: __.cabc.Callable[ ..., __.typx.Any ] = __.subprocess.run,
86) -> None:
87 ''' Runs a command and wraps errors. '''
88 try: executor(
89 command, cwd = working_directory, check = True,
90 stdout = __.subprocess.PIPE, stderr = __.subprocess.PIPE,
91 )
92 except FileNotFoundError as exception:
93 raise _exceptions.ConfigurationInvalidity(
94 str( exception ) ) from exception
95 except __.subprocess.CalledProcessError as exception:
96 temp_ref: __.Absential[ __.Path ] = (
97 temporary_directory if preserve else __.absent )
98 stderr_text: __.Absential[ str ] = (
99 exception.stderr.decode( 'utf-8', errors = 'replace' )
100 if exception.stderr else __.absent )
101 raise _exceptions.ValidationCommandFailure(
102 command, exception.returncode, temp_ref, stderr_text
103 ) from exception
106def copy_template( # noqa: PLR0913
107 answers_file: __.Path,
108 project_directory: __.Path,
109 template_directory: __.Path,
110 vcs_ref: __.Absential[ str ] = __.absent,
111 unsafe: bool = False,
112 answers_reader: __.cabc.Callable[
113 [ __.Path ], dict[ str, __.typx.Any ]
114 ] = _acquire_answers_file,
115 copier: __.Absential[
116 __.cabc.Callable[ ..., __.typx.Any ]
117 ] = __.absent,
118) -> None:
119 ''' Copies template using Copier Python API. '''
120 copier_copy: __.cabc.Callable[ ..., __.typx.Any ]
121 if __.is_absent( copier ):
122 from copier import run_copy as copier_copy
123 else: copier_copy = copier
124 copy_kwargs: dict[ str, __.typx.Any ] = dict(
125 data = answers_reader( answers_file ),
126 defaults = True,
127 overwrite = True,
128 quiet = True,
129 )
130 if not __.is_absent( vcs_ref ):
131 copy_kwargs[ 'vcs_ref' ] = vcs_ref
132 if unsafe: copy_kwargs[ 'unsafe' ] = True
133 try: copier_copy(
134 str( template_directory ),
135 project_directory,
136 **copy_kwargs,
137 )
138 except Exception as exception:
139 raise _exceptions.ConfigurationInvalidity(
140 str( exception ) ) from exception
143def execute_validation_commands( # noqa: PLR0913
144 config: _config.Configuration,
145 template_directory: __.Path,
146 project_directory: __.Path,
147 temporary_directory: __.Path,
148 variant: str,
149 executor: __.cabc.Callable[ ..., __.typx.Any ] = __.subprocess.run,
150) -> None:
151 ''' Executes validation commands sequentially. '''
152 preserve = (
153 bool( config.preserve )
154 if not __.is_absent( config.preserve )
155 else False )
156 if __.is_absent( config.commands ): return
157 for cmd in config.commands:
158 args, cwd = _config.interpolate_command(
159 cmd, template_directory, project_directory,
160 temporary_directory, variant )
161 _scribe.debug(
162 f"Running validation command: {' '.join( args )}" )
163 _execute_command(
164 args, cwd, temporary_directory, preserve,
165 executor = executor )
168def survey_variants(
169 answers_directory: __.Path,
170) -> tuple[ str, ... ]:
171 ''' Discovers variant names from ``answers-*.yaml`` files. '''
172 if not answers_directory.is_dir( ):
173 raise _exceptions.ConfigurationAbsence( answers_directory )
174 return tuple( sorted(
175 fsent.stem[ len( 'answers-' ): ]
176 for fsent in answers_directory.glob( 'answers-*.yaml' )
177 if fsent.is_file( )
178 ) )
181def validate_variant(
182 variant: str,
183 config: _config.Configuration,
184 copier: __.Absential[
185 __.cabc.Callable[ ..., __.typx.Any ]
186 ] = __.absent,
187 executor: __.cabc.Callable[ ..., __.typx.Any ] = __.subprocess.run,
188) -> ValidationResult:
189 ''' Validates a single template variant. '''
190 answers_dir = config.answers_directory
191 if __.is_absent( answers_dir ):
192 raise _exceptions.ConfigurationInvalidity(
193 subject = 'answers directory' )
194 answers_file = answers_dir / f"answers-{variant}.yaml"
195 if not answers_file.is_file( ):
196 raise _exceptions.ConfigurationAbsence( answers_file )
197 template_directory = _resolve_template_directory( config )
198 _scribe.info( f"Validating variant: {variant}" )
199 temporary_directory = _create_temporary_directory( variant )
200 preserve = (
201 bool( config.preserve )
202 if not __.is_absent( config.preserve )
203 else False )
204 unsafe = (
205 bool( config.unsafe )
206 if not __.is_absent( config.unsafe )
207 else False )
208 try:
209 project_directory = temporary_directory / variant
210 copy_template(
211 answers_file, project_directory, template_directory,
212 config.vcs_ref, unsafe,
213 copier = copier )
214 execute_validation_commands(
215 config, template_directory, project_directory,
216 temporary_directory, variant,
217 executor = executor )
218 commands_count = (
219 0 if __.is_absent( config.commands )
220 else len( config.commands ) )
221 items = commands_count + 1
222 result = ValidationResult(
223 variant = variant,
224 temporary_directory = temporary_directory,
225 items_attempted = items,
226 items_generated = items,
227 preserved = preserve,
228 )
229 except _exceptions.ValidationCommandFailure:
230 if not preserve: 230 ↛ 232line 230 didn't jump to line 232 because the condition on line 230 was always true
231 _remove_temporary_directory( temporary_directory )
232 raise
233 except Exception:
234 if not preserve:
235 _remove_temporary_directory( temporary_directory )
236 raise
237 if not preserve:
238 _remove_temporary_directory( temporary_directory )
239 return result
242def _create_temporary_directory( variant: str ) -> __.Path:
243 ''' Creates a temporary directory for validation. '''
244 try: return __.Path( __.tempfile.mkdtemp(
245 prefix = f"copiertv-{variant}-" ) )
246 except ( OSError, IOError ) as exception: # pragma: no cover
247 raise _exceptions.FileOperationFailure(
248 __.Path( __.tempfile.gettempdir( ) ),
249 'create temporary directory' ) from exception
252def _remove_temporary_directory( path: __.Path ) -> None:
253 ''' Removes a temporary directory, suppressing errors. '''
254 _scribe.debug( f"Cleaning up temporary directory: {path}" )
255 with __.ctxl.suppress( OSError, IOError ):
256 __.shutil.rmtree( path )
259def _resolve_template_directory(
260 config: _config.Configuration,
261) -> __.Path:
262 ''' Resolves template directory from configuration. '''
263 if not __.is_absent( config.template_directory ):
264 return config.template_directory
265 raise _exceptions.ConfigurationInvalidity(
266 subject = 'template directory' )