feat(core): wire refresh mode through scaffold pipeline

packages/create-awesome-python-app/src/create_awesome_python_app/cli.py

@@ -20,6 +20,7 @@
  print_env_info,
  resolve_source,
 )
+from create_python_app_core.git_cache import RefreshMode
 from rich.console import Console
 
 from create_awesome_python_app import __version__
@@ -57,6 +58,16 @@ def _parse_set_options(set_opt: list[str] | None) -> dict[str, str]:
  return set_map
 
 
+def _normalize_refresh(refresh: str | None) -> RefreshMode | None:
+ if refresh == "always":
+ return "always"
+ if refresh == "stale":
+ return "stale"
+ if refresh == "manual":
+ return "manual"
+ return None
+
+
 def _stringify_option_value(value: Any) -> str:
  if value is None:
  return ""
@@ -88,12 +99,18 @@ def _prompt_custom_options(
  set_map: dict[str, str],
  cache_dir: Path | None,
  offline: bool,
+ refresh: str | None,
  registry_options: list[dict[str, Any]] | None = None,
 ) -> dict[str, str]:
  import questionary
 
  source = resolve_source(template, cache_dir=cache_dir)
- root = download_repository(source, offline=offline, cache_root=cache_dir)
+ root = download_repository(
+ source,
+ offline=offline,
+ refresh=_normalize_refresh(refresh),
+ cache_root=cache_dir,
+ )
  try:
  config = load_cpa_config(_template_config_path(source.subdir, root))
  except ConfigParseError as err:
@@ -189,14 +206,22 @@ def scaffold(
  la(template)
  raise typer.Exit(0)
 
+ effective_refresh = _normalize_refresh(refresh)
+ if refresh and effective_refresh is None:
+ console.print(
+ "[red]Invalid --refresh mode: "
+ f"'{refresh}'. Use one of: always, stale, manual.[/red]"
+ )
+ raise typer.Exit(2)
+
  # env wiring (#36)
  if no_cache:
  os.environ["CPA_NO_CATALOG_CACHE"] = "1"
-os.environ["CPA_REFRESH"] = "always"
+effective_refresh = effective_refresh or "always"
  if cache_dir:
  os.environ["CPA_CACHE_DIR"] = str(cache_dir)
- if refresh:
- os.environ["CPA_REFRESH"] = refresh
+ if effective_refresh:
+ os.environ["CPA_REFRESH"] = effective_refresh
  if offline:
  pass # passed to core
 
@@ -339,6 +364,7 @@ def scaffold(
  set_map=set_map,
  cache_dir=cache_dir,
  offline=offline,
+ refresh=effective_refresh,
  registry_options=registry_options,
  )
  except ImportError:
@@ -371,6 +397,7 @@ def scaffold(
  "force": force,
  "verbose": verbose,
  "offline": offline,
+ "refresh": effective_refresh,
  "keep_on_failure": keep_on_failure,
  "cache_dir": str(cache_dir) if cache_dir else None,
  "set": set_map,

packages/create-awesome-python-app/tests/test_cli.py

@@ -1,7 +1,20 @@
+from pathlib import Path
+
+import pytest
 from create_awesome_python_app.cli import app
 from typer.testing import CliRunner
 
 runner = CliRunner()
+_CPA_ENV_VARS = ("CPA_REFRESH", "CPA_NO_CATALOG_CACHE", "CPA_CACHE_DIR")
+
+
+@pytest.fixture(autouse=True)
+def _clean_cpa_env(monkeypatch):
+ for name in _CPA_ENV_VARS:
+ monkeypatch.delenv(name, raising=False)
+ yield
+ for name in _CPA_ENV_VARS:
+ monkeypatch.delenv(name, raising=False)
 
 
 def test_version() -> None:
@@ -15,3 +28,143 @@ def test_help() -> None:
  result = runner.invoke(app, ["--help"])
  assert result.exit_code == 0
  assert "Scaffold" in result.stdout or "create" in result.stdout.lower()
+
+
+def test_refresh_flag_is_forwarded(tmp_path: Path, monkeypatch) -> None:
+ tpl = tmp_path / "tpl"
+ tpl.mkdir()
+ captured: dict[str, object] = {}
+
+ async def fake_check_for_latest_version(_package_name):
+ return None
+
+ async def fake_create_python_app(project_directory, options, *_args, **_kwargs):
+ captured["project_directory"] = project_directory
+ captured["options"] = options
+
+ monkeypatch.setattr(
+ "create_awesome_python_app.cli.check_for_latest_version",
+ fake_check_for_latest_version,
+ )
+ monkeypatch.setattr(
+ "create_awesome_python_app.cli.create_python_app",
+ fake_create_python_app,
+ )
+
+ result = runner.invoke(
+ app,
+ [
+ "--template",
+ f"file://{tpl}",
+ "--refresh",
+ "always",
+ "--no-install",
+ "--no-interactive",
+ "api",
+ ],
+ )
+
+ assert result.exit_code == 0, result.stdout + result.stderr
+ options = captured["options"]
+ assert isinstance(options, dict)
+ assert options["refresh"] == "always"
+
+
+def test_invalid_refresh_mode_fails(tmp_path: Path) -> None:
+ tpl = tmp_path / "tpl"
+ tpl.mkdir()
+
+ result = runner.invoke(
+ app,
+ [
+ "--template",
+ f"file://{tpl}",
+ "--refresh",
+ "bogus",
+ "--no-install",
+ "--no-interactive",
+ "api",
+ ],
+ )
+
+ assert result.exit_code == 2
+ assert "Invalid --refresh mode" in result.stdout + result.stderr
+
+
+def test_no_cache_sets_explicit_refresh(tmp_path: Path, monkeypatch) -> None:
+ tpl = tmp_path / "tpl"
+ tpl.mkdir()
+ captured: dict[str, object] = {}
+
+ async def fake_check_for_latest_version(_package_name):
+ return None
+
+ async def fake_create_python_app(project_directory, options, *_args, **_kwargs):
+ captured["project_directory"] = project_directory
+ captured["options"] = options
+
+ monkeypatch.setattr(
+ "create_awesome_python_app.cli.check_for_latest_version",
+ fake_check_for_latest_version,
+ )
+ monkeypatch.setattr(
+ "create_awesome_python_app.cli.create_python_app",
+ fake_create_python_app,
+ )
+
+ result = runner.invoke(
+ app,
+ [
+ "--template",
+ f"file://{tpl}",
+ "--no-cache",
+ "--no-install",
+ "--no-interactive",
+ "api",
+ ],
+ )
+
+ assert result.exit_code == 0, result.stdout + result.stderr
+ options = captured["options"]
+ assert isinstance(options, dict)
+ assert options["refresh"] == "always"
+
+
+def test_pin_appends_ref_to_template_url(tmp_path: Path, monkeypatch) -> None:
+ tpl = tmp_path / "tpl"
+ tpl.mkdir()
+ captured: dict[str, object] = {}
+
+ async def fake_check_for_latest_version(_package_name):
+ return None
+
+ async def fake_create_python_app(project_directory, options, *_args, **_kwargs):
+ captured["project_directory"] = project_directory
+ captured["options"] = options
+
+ monkeypatch.setattr(
+ "create_awesome_python_app.cli.check_for_latest_version",
+ fake_check_for_latest_version,
+ )
+ monkeypatch.setattr(
+ "create_awesome_python_app.cli.create_python_app",
+ fake_create_python_app,
+ )
+
+ result = runner.invoke(
+ app,
+ [
+ "--template",
+ f"file://{tpl}",
+ "--pin",
+ "abc123",
+ "--no-install",
+ "--no-interactive",
+ "api",
+ ],
+ )
+
+ assert result.exit_code == 0, result.stdout + result.stderr
+ options = captured["options"]
+ assert isinstance(options, dict)
+ assert options["template"] == f"file://{tpl}?ref=abc123"

packages/create-python-app-core/src/create_python_app_core/api.py

@@ -10,6 +10,7 @@
 from typing import Any
 
 from create_python_app_core._version import __version__
+from create_python_app_core.git_cache import RefreshMode
 
 CPA_USER_AGENT = (
  f"create-python-app-core/{__version__} "
@@ -84,6 +85,10 @@ async def create_python_app(
  options = await transform_options(options)
 
  cache = options.get("cache_dir")
+ refresh = options.get("refresh")
+ refresh_mode: RefreshMode | None = (
+ refresh if refresh in ("always", "stale", "manual") else None
+ )
  scaffold_project(
  project_directory,
  template=str(options.get("template") or ""),
@@ -92,6 +97,7 @@ async def create_python_app(
  force=bool(options.get("force", False)),
  install=bool(options.get("install", True)),
  offline=bool(options.get("offline", False)),
+ refresh=refresh_mode,
  keep_on_failure=bool(options.get("keep_on_failure", False)),
  cache_dir=Path(cache) if cache else None,
  options=options,

packages/create-python-app-core/src/create_python_app_core/installer.py

@@ -14,7 +14,7 @@
  load_cpa_config,
 )
 from create_python_app_core.errors import CpaError, ScaffoldAbortedError
-from create_python_app_core.git_cache import download_repository
+from create_python_app_core.git_cache import RefreshMode, download_repository
 from create_python_app_core.loaders import merge_layers
 from create_python_app_core.paths import ResolvedSource, resolve_source
 
@@ -67,6 +67,7 @@ def scaffold_project(
  force: bool = False,
  install: bool = True,
  offline: bool = False,
+ refresh: RefreshMode | None = None,
  keep_on_failure: bool = False,
  cache_dir: Path | None = None,
  options: dict[str, Any] | None = None,
@@ -82,7 +83,12 @@ def scaffold_project(
  try:
  for spec in specs:
  source = resolve_source(spec, cache_dir=cache_dir)
- root = download_repository(source, offline=offline, cache_root=cache_dir)
+ root = download_repository(
+ source,
+ offline=offline,
+ refresh=refresh,
+ cache_root=cache_dir,
+ )
  layers.append((source, root))
  configs.append(load_cpa_config(_config_path(source, root)))
 

packages/create-python-app-core/tests/test_api.py

@@ -1,5 +1,7 @@
+import asyncio
+
 import pytest
-from create_python_app_core import check_python_version
+from create_python_app_core import check_python_version, create_python_app
 
 
 def test_check_python_version_accepts_current() -> None:
@@ -9,3 +11,30 @@ def test_check_python_version_accepts_current() -> None:
 def test_check_python_version_rejects_impossible() -> None:
  with pytest.raises(SystemExit):
  check_python_version(">=99.0", "create-python-app-core")
+
+
+def test_create_python_app_forwards_refresh(monkeypatch) -> None:
+ captured: dict[str, object] = {}
+
+ def fake_scaffold_project(project_directory: str, **kwargs) -> None:
+ captured["project_directory"] = project_directory
+ captured.update(kwargs)
+
+ monkeypatch.setattr(
+ "create_python_app_core.installer.scaffold_project",
+ fake_scaffold_project,
+ )
+
+ asyncio.run(
+ create_python_app(
+ "api",
+ {
+ "template": "file:///template",
+ "install": False,
+ "refresh": "always",
+ },
+ )
+ )
+
+ assert captured["project_directory"] == "api"
+ assert captured["refresh"] == "always"

packages/create-python-app-core/tests/test_installer.py

@@ -2,6 +2,7 @@
 
 import pytest
 from create_python_app_core.installer import scaffold_project
+from create_python_app_core.paths import ResolvedSource
 
 
 def _tpl(tmp: Path, name: str) -> str:
@@ -20,3 +21,34 @@ def test_scaffold_file_template(
  url = _tpl(tmp_path, "tpl")
  scaffold_project(str(dest), template=url, install=False)
  assert (dest / "hello.txt").read_text() == "hi"
+
+
+def test_scaffold_forwards_refresh_to_download_repository(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ monkeypatch.setenv("CPA_SKIP_GIT", "1")
+ dest = tmp_path / "app"
+ url = _tpl(tmp_path, "tpl")
+ refresh_values: list[str | None] = []
+
+ def fake_download_repository(
+ source: ResolvedSource,
+ *,
+ offline: bool = False,
+ refresh: str | None = None,
+ cache_root: Path | None = None,
+ ) -> Path:
+ refresh_values.append(refresh)
+ assert offline is False
+ assert cache_root is None
+ assert source.local_path is not None
+ return source.local_path
+
+ monkeypatch.setattr(
+ "create_python_app_core.installer.download_repository",
+ fake_download_repository,
+ )
+
+ scaffold_project(str(dest), template=url, install=False, refresh="always")
+
+ assert refresh_values == ["always"]