diff --git a/copier/_cli.py b/copier/_cli.py index 3d5f65a3e..ef24d97ef 100644 --- a/copier/_cli.py +++ b/copier/_cli.py @@ -239,6 +239,13 @@ class CopierCopySubApp(_Subcommand): ["-w", "--overwrite"], help="Overwrite files that already exist, without asking.", ) + version_subscription = cli.SwitchAttr( + ["-V", "--version-subscription"], + str, + default=None, + help="Prevent copying from a version of the template that does not adhere to " + "the version specification.", + ) def main(self, template_src: str, destination_path: str) -> int: """Call [run_copy][copier.run_copy]. @@ -270,6 +277,7 @@ def inner() -> None: quiet=self.quiet, unsafe=self.unsafe, skip_tasks=self.skip_tasks, + version_subscription=self.version_subscription, ) return _handle_exceptions(inner) @@ -318,6 +326,13 @@ class CopierRecopySubApp(_Subcommand): default=False, help="Skip questions that have already been answered", ) + version_subscription = cli.SwitchAttr( + ["-V", "--version-subscription"], + str, + default=None, + help="Prevent copying from a version of the template that does not adhere to " + "the version specification.", + ) def main(self, destination_path: cli.ExistingDirectory = ".") -> int: """Call [run_recopy][copier.run_recopy]. @@ -347,6 +362,7 @@ def inner() -> None: unsafe=self.unsafe, skip_answered=self.skip_answered, skip_tasks=self.skip_tasks, + version_subscription=self.version_subscription, ) return _handle_exceptions(inner) @@ -401,6 +417,13 @@ class CopierUpdateSubApp(_Subcommand): default=False, help="Skip questions that have already been answered", ) + version_subscription = cli.SwitchAttr( + ["-V", "--version-subscription"], + str, + default=None, + help="Prevent updates to any version that does not adhere to the version " + "specification.", + ) def main(self, destination_path: cli.ExistingDirectory = ".") -> int: """Call [run_update][copier.run_update]. @@ -432,6 +455,7 @@ def inner() -> None: unsafe=self.unsafe, skip_answered=self.skip_answered, skip_tasks=self.skip_tasks, + version_subscription=self.version_subscription, ) return _handle_exceptions(inner) diff --git a/copier/_main.py b/copier/_main.py index 57a029438..0c34371a0 100644 --- a/copier/_main.py +++ b/copier/_main.py @@ -32,6 +32,7 @@ from jinja2.loaders import FileSystemLoader from jinja2.sandbox import SandboxedEnvironment +from packaging.specifiers import SpecifierSet from packaging.version import Version from pathspec import PathSpec, __version__ as pathspec_version from plumbum import ProcessExecutionError, colors @@ -249,6 +250,7 @@ class Worker: unsafe: bool = False skip_answered: bool = False skip_tasks: bool = False + version_subscription: str | None = None answers: AnswersMap = field(default_factory=AnswersMap, init=False) _cleanup_hooks: list[Callable[[], None]] = field(default_factory=list, init=False) @@ -360,7 +362,12 @@ def _answers_to_remember(self) -> Mapping[str, Any]: answers: AnyByStrDict = {} commit = self.template.commit src = self.template.url - for key, value in (("_commit", commit), ("_src_path", src)): + version_subscription = self.template.version_subscription + for key, value in ( + ("_commit", commit), + ("_src_path", src), + ("_version_subscription", version_subscription), + ): if value is not None: answers[key] = value # Other data goes next @@ -1065,7 +1072,16 @@ def template(self) -> Template: raise TypeError("Template not found") url = str(self.subproject.template.url) ref = self.resolved_vcs_ref - result = Template(url=url, ref=ref, use_prereleases=self.use_prereleases) + version_subscription = ( + self.version_subscription + or self.subproject.last_answers.get("_version_subscription", None) + ) + result = Template( + url=url, + ref=ref, + use_prereleases=self.use_prereleases, + version_subscription=version_subscription, + ) self._cleanup_hooks.append(result._cleanup) return result @@ -1189,6 +1205,13 @@ def run_update(self) -> None: # review the diff before committing; so we can safely avoid # asking for confirmation raise UserMessageError("Enable overwrite to update a subproject.") + if self.version_subscription and not SpecifierSet( + self.version_subscription + ).contains(self.template.version): + raise UserMessageError( + f"Cannot update: new version {self.template.version}, as it does " + f'not adhere to specification "{self.version_subscription}".' + ) self._print_message(self.template.message_before_update) self._print_template_update_info(self.subproject.template) with suppress(AttributeError): @@ -1544,6 +1567,7 @@ def run_copy( quiet: bool = False, unsafe: bool = False, skip_tasks: bool = False, + version_subscription: SpecifierSet | None = None, ) -> Worker: """Copy a template to a destination, from zero.""" with Worker( @@ -1572,6 +1596,7 @@ def run_copy( quiet=quiet, unsafe=unsafe, skip_tasks=skip_tasks, + version_subscription=version_subscription, ) as worker: worker.run_copy() return worker @@ -1596,6 +1621,7 @@ def run_recopy( unsafe: bool = False, skip_answered: bool = False, skip_tasks: bool = False, + version_subscription: SpecifierSet | None = None, ) -> Worker: """Update a subproject from its template, discarding subproject evolution.""" with Worker( @@ -1624,6 +1650,7 @@ def run_recopy( unsafe=unsafe, skip_answered=skip_answered, skip_tasks=skip_tasks, + version_subscription=version_subscription, ) as worker: worker.run_recopy() return worker @@ -1650,6 +1677,7 @@ def run_update( unsafe: bool = False, skip_answered: bool = False, skip_tasks: bool = False, + version_subscription: SpecifierSet | None = None, ) -> Worker: """Update a subproject, from its template.""" with Worker( @@ -1680,6 +1708,7 @@ def run_update( unsafe=unsafe, skip_answered=skip_answered, skip_tasks=skip_tasks, + version_subscription=version_subscription, ) as worker: worker.run_update() return worker diff --git a/copier/_subproject.py b/copier/_subproject.py index fbd2e5b6f..d989ec348 100644 --- a/copier/_subproject.py +++ b/copier/_subproject.py @@ -67,7 +67,8 @@ def last_answers(self) -> AnyByStrDict: return { key: value for key, value in self._raw_answers.items() - if key in {"_src_path", "_commit"} or not key.startswith("_") + if key in {"_src_path", "_commit", "_version_subscription"} + or not key.startswith("_") } @cached_property diff --git a/copier/_template.py b/copier/_template.py index 7650e6662..0ae9b1119 100644 --- a/copier/_template.py +++ b/copier/_template.py @@ -18,6 +18,7 @@ import packaging.version import yaml from funcy import lflatten +from packaging.specifiers import SpecifierSet from packaging.version import Version, parse from plumbum.machines import local from pydantic.dataclasses import dataclass @@ -219,6 +220,7 @@ class Template: url: str ref: str | None = None use_prereleases: bool = False + version_subscription: str | None = None def _cleanup(self) -> None: if temp_clone := self._temp_clone(): @@ -570,7 +572,14 @@ def local_abspath(self) -> Path: result = Path( clone( self.url_expanded, - self.ref or get_latest_tag(self.url_expanded, self.use_prereleases), + self.ref + or get_latest_tag( + self.url_expanded, + self.use_prereleases, + self.version_subscription + and SpecifierSet(self.version_subscription) + or None, + ), ) ) if not result.is_dir(): diff --git a/copier/_vcs.py b/copier/_vcs.py index 3b2ebf7a8..b89b3b521 100644 --- a/copier/_vcs.py +++ b/copier/_vcs.py @@ -10,12 +10,13 @@ from warnings import warn from packaging import version +from packaging.specifiers import SpecifierSet from packaging.version import InvalidVersion, Version from plumbum import TF, ProcessExecutionError, colors, local from plumbum.machines import LocalCommand from ._types import OptBool, OptStrOrPath, StrOrPath -from .errors import DirtyLocalWarning, ShallowCloneWarning +from .errors import DirtyLocalWarning, ShallowCloneWarning, UserMessageError GIT_USER_NAME = "Copier" GIT_USER_EMAIL = "copier@copier" @@ -126,7 +127,9 @@ def get_repo(url: str) -> str | None: return None -def get_latest_tag(url: str, use_prereleases: OptBool = False) -> str: +def get_latest_tag( + url: str, use_prereleases: OptBool = False, spec: SpecifierSet | None = None +) -> str: """Get latest git tag, sorted by PEP 440. Args: @@ -135,6 +138,9 @@ def get_latest_tag(url: str, use_prereleases: OptBool = False) -> str: [get_repo][copier.vcs.get_repo]. use_prereleases: If `False`, skip prerelease git tags. + spec: + An optional pep440 specifier set to filter the available tags + with. Returns: The latest git tag, or `HEAD` if no valid tags are found. @@ -147,10 +153,16 @@ def get_latest_tag(url: str, use_prereleases: OptBool = False) -> str: all_tags = (tag for tag in all_tags if valid_version(tag)) if not use_prereleases: all_tags = (tag for tag in all_tags if not version.parse(tag).is_prerelease) + if spec: + all_tags = spec.filter(all_tags, use_prereleases) sorted_tags = sorted(all_tags, key=version.parse, reverse=True) try: return str(sorted_tags[0]) - except IndexError: + except IndexError as e: + if spec: + raise UserMessageError( + f"No git tag found that matches version spec {spec}." + ) from e print( colors.warn | "No git tags found in template; using HEAD as ref", file=sys.stderr, diff --git a/tests/test_cli.py b/tests/test_cli.py index 8ed4cf5ac..ce243f996 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -392,6 +392,10 @@ def test_copy_help(capsys: pytest.CaptureFixture[str]) -> None: -T, --skip-tasks Skip template tasks execution --UNSAFE, --trust Allow templates with unsafe features (Jinja extensions, migrations, tasks) + -V, --version-subscription VALUE:str + Prevent copying from a version of the + template that does not adhere to the version + specification. -a, --answers-file VALUE:str Update using this path (relative to `destination_path`) to find the answers file -d, --data VARIABLE=VALUE:str Make VARIABLE available as VALUE when @@ -458,6 +462,9 @@ def test_update_help(capsys: pytest.CaptureFixture[str]) -> None: -T, --skip-tasks Skip template tasks execution --UNSAFE, --trust Allow templates with unsafe features (Jinja extensions, migrations, tasks) + -V, --version-subscription VALUE:str + Prevent updates to any version that does not + adhere to the version specification. -a, --answers-file VALUE:str Update using this path (relative to `destination_path`) to find the answers file -c, --context-lines VALUE:int Lines of context to use for detecting diff --git a/tests/test_vcs.py b/tests/test_vcs.py index c6c8668b7..c9787c66a 100644 --- a/tests/test_vcs.py +++ b/tests/test_vcs.py @@ -213,10 +213,20 @@ def test_invalid_version(tmp_path: Path) -> None: assert get_latest_tag(str(tmp_path)) == "v2" -@pytest.mark.parametrize("sorter", [iter, reversed]) +@pytest.mark.parametrize( + ("sorter", "expected_version", "version_subscription"), + [ + (iter, "v1.0.1", ""), + (reversed, "v1.0.1", ""), + (iter, "v1", "==1.0.0"), + (iter, "v1.0.1", ">=1.0.0"), + ], +) def test_select_latest_version_tag( tmp_path_factory: pytest.TempPathFactory, sorter: Callable[[Sequence[str]], Iterator[str]], + expected_version: str, + version_subscription: str, ) -> None: src, dst = map(tmp_path_factory.mktemp, ("src", "dst")) filename = "version.txt" @@ -232,12 +242,18 @@ def test_select_latest_version_tag( git("commit", "-m", version) git("tag", version) - run_copy(str(src), dst) + args = { + "src_path": str(src), + "dst_path": dst, + } + if version_subscription: + args["version_subscription"] = version_subscription + run_copy(**args) assert (dst / filename).is_file() - assert (dst / filename).read_text() == "v1.0.1" + assert (dst / filename).read_text() == expected_version answers = load_answersfile_data(dst) - assert answers["_commit"] == "v1.0.1" + assert answers["_commit"] == expected_version @pytest.mark.parametrize(