Coverage for sources/copiertv/engine.py: 99%
89 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-14 02:11 +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 for cmd in config.commands:
157 args, cwd = _config.interpolate_command(
158 cmd, template_directory, project_directory,
159 temporary_directory, variant )
160 _scribe.debug(
161 f"Running validation command: {' '.join( args )}" )
162 _execute_command(
163 args, cwd, temporary_directory, preserve,
164 executor = executor )
167def survey_variants(
168 answers_directory: __.Path,
169) -> tuple[ str, ... ]:
170 ''' Discovers variant names from ``answers-*.yaml`` files. '''
171 if not answers_directory.is_dir( ):
172 raise _exceptions.ConfigurationAbsence( answers_directory )
173 return tuple( sorted(
174 fsent.stem[ len( 'answers-' ): ]
175 for fsent in answers_directory.glob( 'answers-*.yaml' )
176 if fsent.is_file( )
177 ) )
180def validate_variant(
181 variant: str,
182 config: _config.Configuration,
183 copier: __.Absential[
184 __.cabc.Callable[ ..., __.typx.Any ]
185 ] = __.absent,
186 executor: __.cabc.Callable[ ..., __.typx.Any ] = __.subprocess.run,
187) -> ValidationResult:
188 ''' Validates a single template variant. '''
189 answers_dir = config.answers_directory
190 if __.is_absent( answers_dir ):
191 raise _exceptions.ConfigurationInvalidity( )
192 answers_file = answers_dir / f"answers-{variant}.yaml"
193 if not answers_file.is_file( ):
194 raise _exceptions.ConfigurationAbsence( answers_file )
195 template_directory = _resolve_template_directory( config )
196 _scribe.info( f"Validating variant: {variant}" )
197 temporary_directory = _create_temporary_directory( variant )
198 preserve = (
199 bool( config.preserve )
200 if not __.is_absent( config.preserve )
201 else False )
202 unsafe = (
203 bool( config.unsafe )
204 if not __.is_absent( config.unsafe )
205 else False )
206 try:
207 project_directory = temporary_directory / variant
208 copy_template(
209 answers_file, project_directory, template_directory,
210 config.vcs_ref, unsafe,
211 copier = copier )
212 execute_validation_commands(
213 config, template_directory, project_directory,
214 temporary_directory, variant,
215 executor = executor )
216 items = len( config.commands ) + 1
217 result = ValidationResult(
218 variant = variant,
219 temporary_directory = temporary_directory,
220 items_attempted = items,
221 items_generated = items,
222 preserved = preserve,
223 )
224 except _exceptions.ValidationCommandFailure:
225 if not preserve: 225 ↛ 227line 225 didn't jump to line 227 because the condition on line 225 was always true
226 _remove_temporary_directory( temporary_directory )
227 raise
228 except Exception:
229 if not preserve:
230 _remove_temporary_directory( temporary_directory )
231 raise
232 if not preserve:
233 _remove_temporary_directory( temporary_directory )
234 return result
237def _create_temporary_directory( variant: str ) -> __.Path:
238 ''' Creates a temporary directory for validation. '''
239 try: return __.Path( __.tempfile.mkdtemp(
240 prefix = f"copiertv-{variant}-" ) )
241 except ( OSError, IOError ) as exception: # pragma: no cover
242 raise _exceptions.FileOperationFailure(
243 __.Path( __.tempfile.gettempdir( ) ),
244 'create temporary directory' ) from exception
247def _remove_temporary_directory( path: __.Path ) -> None:
248 ''' Removes a temporary directory, suppressing errors. '''
249 _scribe.debug( f"Cleaning up temporary directory: {path}" )
250 with __.ctxl.suppress( OSError, IOError ):
251 __.shutil.rmtree( path )
254def _resolve_template_directory(
255 config: _config.Configuration,
256) -> __.Path:
257 ''' Resolves template directory from configuration. '''
258 if not __.is_absent( config.template_directory ):
259 return config.template_directory
260 raise _exceptions.ConfigurationInvalidity( )