Implement exercise version pinning

app/cli.py

@@ -10,8 +10,8 @@
 from app.utils.gitmastery import (
     find_exercise_root,
     find_gitmastery_root,
-    read_gitmastery_config,
     read_exercise_config,
+    read_gitmastery_config,
 )
 from app.utils.version import Version
 from app.version import __version__
@@ -45,6 +45,10 @@ def cli(ctx, verbose) -> None:
         ctx.obj[CliContextKey.GITMASTERY_EXERCISE_CONFIG] = exercise_root_config
 
     ctx.obj[CliContextKey.VERBOSE] = verbose
+    # We make the assumption that within a single command run, the "state of the world"
+    # is immutable, allowing us to cache things
+    ctx.obj[CliContextKey.WEB_CACHE] = {}
+    ctx.obj[CliContextKey.TAG_CACHE] = []
 
     current_version = Version.parse_version_string(__version__)
     ctx.obj[CliContextKey.VERSION] = current_version

app/commands/setup_folder.py

@@ -6,6 +6,7 @@
 from app.commands.check.git import git
 from app.commands.progress.constants import PROGRESS_LOCAL_FOLDER_NAME
 from app.utils.click import error, info, invoke_command, prompt
+from app.utils.exercises import get_latest_release_exercise_version
 
 
 @click.command("setup")
@@ -37,8 +38,22 @@ def setup() -> None:
     info("Setting up your local progress tracker...")
     os.makedirs(PROGRESS_LOCAL_FOLDER_NAME, exist_ok=True)
     with open(".gitmastery.json", "w") as gitmastery_file:
+        version_to_pin = get_latest_release_exercise_version()
+        if version_to_pin is None:
+            # For now, we just error out because we should never be in this bad state.
+            raise ValueError(
+                "Unexpected error occurred when fetching exercises due to missing exercises tag. Contact the Git-Mastery team."
+            )
+        version_to_pin.pinned = True
+        info(f"Pinning your exercises to {version_to_pin}")
         gitmastery_file.write(
-            json.dumps({"progress_local": True, "progress_remote": False})
+            json.dumps(
+                {
+                    "progress_local": True,
+                    "progress_remote": False,
+                    "exercises_version": version_to_pin.to_version_string(),
+                }
+            )
         )
 
     with open("progress/progress.json", "a") as progress_file:

app/gitmastery_config.py

@@ -2,11 +2,14 @@
 from dataclasses import dataclass
 from pathlib import Path
 
+from app.utils.version import Version
+
 
 @dataclass
 class GitMasteryConfig:
     progress_local: bool
     progress_remote: bool
+    exercises_version: Version
 
     path: Path
     cds: int

app/utils/click.py

@@ -1,7 +1,7 @@
 import logging
 import sys
 from enum import StrEnum
-from typing import Any, Dict, NoReturn, Optional
+from typing import Any, Dict, List, NoReturn, Optional
 
 import click
 
@@ -16,6 +16,8 @@ class CliContextKey(StrEnum):
     GITMASTERY_EXERCISE_CONFIG = "GITMASTERY_EXERCISE_CONFIG"
     VERBOSE = "VERBOSE"
     VERSION = "VERSION"
+    WEB_CACHE = "WEB_CACHE"
+    TAG_CACHE = "TAG_CACHE"
 
 
 class ClickColor(StrEnum):
@@ -109,6 +111,14 @@ def get_exercise_root_config() -> Optional[ExerciseConfig]:
     )
 
 
+def get_web_cache() -> Dict[str, str | bytes | Any]:
+    return click.get_current_context().obj.get(CliContextKey.WEB_CACHE, {})
+
+
+def get_tag_cache() -> List[str]:
+    return click.get_current_context().obj.get(CliContextKey.TAG_CACHE, [])
+
+
 def invoke_command(command: click.Command) -> None:
     ctx = click.get_current_context()
     ctx.invoke(command)

app/utils/exercises.py

@@ -0,0 +1,76 @@
+from typing import List, Optional
+
+from app.utils.click import get_tag_cache, get_verbose
+from app.utils.command import run
+from app.utils.version import Version
+
+
+def get_exercises_tags() -> List[str]:
+    tag_cache = get_tag_cache()
+    if len(tag_cache) != 0:
+        if get_verbose():
+            print("Fetching tags from cache")
+        # Use the in-memory, per command cache for tags to avoid re-querying
+        return tag_cache
+
+    result = run(
+        [
+            "git",
+            "ls-remote",
+            "--tags",
+            "--refs",
+            "https://github.com/git-mastery/exercises",
+        ]
+    )
+    versions = []
+    if result.is_success() and result.stdout:
+        lines = result.stdout.split("\n")
+        for line in lines:
+            tag_raw = line.split()[1]
+            if tag_raw.startswith("refs/tags/"):
+                tag = tag_raw[len("refs/tags/") :]
+                versions.append(tag)
+    tag_cache = versions
+    if get_verbose():
+        print("Queried for tags to store in cache")
+    return versions
+
+
+def get_all_exercise_tags() -> List[Version]:
+    tags = get_exercises_tags()
+    return list(sorted([Version.parse_version_string(t) for t in tags], reverse=True))
+
+
+def get_latest_release_exercise_version() -> Optional[Version]:
+    all_tags = get_all_exercise_tags()
+    if len(all_tags) == 0:
+        # Although this should not be happening, we will let the callsite handle this
+        return None
+
+    # These should always ignore the development versions, just focus on the release
+    # versions
+    for tag in all_tags:
+        if tag.prerelease is None:
+            return tag
+
+    return None
+
+
+def get_latest_development_exercise_version() -> Optional[Version]:
+    all_tags = get_all_exercise_tags()
+    if len(all_tags) == 0:
+        # Although this should not be happening, we will let the callsite handle this
+        return None
+
+    for tag in all_tags:
+        if tag.build is None:
+            return tag
+    return None
+
+
+def get_latest_exercise_version_within_pin(pin_version: Version) -> Optional[Version]:
+    all_tags = get_all_exercise_tags()
+    for tag in all_tags:
+        if tag.within_pin(pin_version):
+            return tag
+    return None

app/utils/git.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import List, Optional
 
 from app.utils.command import run
 

app/utils/gitmastery.py

@@ -2,7 +2,6 @@
 import os
 import sys
 import tempfile
-import urllib.parse
 from pathlib import Path
 from typing import Any, Dict, Optional, Tuple, TypeVar, Union
 
@@ -11,8 +10,20 @@
 
 from app.exercise_config import ExerciseConfig
 from app.gitmastery_config import GitMasteryConfig
-from app.utils.click import error, get_exercise_root_config, get_gitmastery_root_config
+from app.utils.click import (
+    error,
+    get_exercise_root_config,
+    get_gitmastery_root_config,
+    get_verbose,
+    get_web_cache,
+)
+from app.utils.exercises import (
+    get_latest_development_exercise_version,
+    get_latest_exercise_version_within_pin,
+    get_latest_release_exercise_version,
+)
 from app.utils.general import ensure_str
+from app.utils.version import Version
 
 GITMASTERY_CONFIG_NAME = ".gitmastery.json"
 GITMASTERY_EXERCISE_CONFIG_NAME = ".gitmastery-exercise.json"
@@ -21,6 +32,33 @@
 )
 
 
+def _construct_gitmastery_exercises_url(filepath: str, version: Version) -> str:
+    if version.release:
+        latest_release = get_latest_release_exercise_version()
+        if latest_release is None:
+            raise ValueError("This should not happen. Contact the Git-Mastery team.")
+        ref = f"tags/v{latest_release.to_version_string()}"
+    elif version.development:
+        latest_development = get_latest_development_exercise_version()
+        if latest_development is None:
+            raise ValueError("This should not happen. Contact the Git-Mastery team.")
+        ref = f"tags/v{latest_development.to_version_string()}"
+    elif not version.pinned:
+        ref = f"tags/v{version.to_version_string()}"
+    else:
+        # If pinned, we need to basically search for all the available tags within the
+        # range
+        latest_within_pin = get_latest_exercise_version_within_pin(version)
+        if latest_within_pin is None:
+            raise ValueError("This should not happen. Contact the Git-Mastery team.")
+        ref = f"tags/v{latest_within_pin.to_version_string()}"
+
+    url = (
+        f"https://raw.githubusercontent.com/git-mastery/exercises/refs/{ref}/{filepath}"
+    )
+    return url
+
+
 def _find_root(filename: str) -> Optional[Tuple[Path, int]]:
     current = Path.cwd()
     steps = 0
@@ -67,6 +105,9 @@ def read_gitmastery_config(gitmastery_config_path: Path, cds: int) -> GitMastery
         cds=cds,
         progress_local=raw_config.get("progress_local", False),
         progress_remote=raw_config.get("progress_remote", False),
+        exercises_version=Version.parse_version_string(
+            raw_config.get("exercises_version", "release")
+        ),
     )
 
 
@@ -120,32 +161,56 @@ def generate_cds_string(cds: int) -> str:
 
 
 def get_gitmastery_file_path(path: str):
-    return urllib.parse.urljoin(GITMASTERY_EXERCISES_BASE_URL, path)
+    config = get_gitmastery_root_config()
+    if config is None:
+        version = Version.RELEASE
+    else:
+        version = config.exercises_version
+    return _construct_gitmastery_exercises_url(path, version)
 
 
 def fetch_file_contents(url: str, is_binary: bool) -> str | bytes:
+    web_cache = get_web_cache()
+    if url in web_cache:
+        if get_verbose():
+            print(f"Fetching {url} contents from WEB_CACHE")
+        return web_cache[url]
     response = requests.get(url)
 
+    if get_verbose():
+        print(f"Querying {url} for contents")
+
     if response.status_code == 200:
         if is_binary:
-            return response.content
-        return response.text
+            web_cache[url] = response.content
+        else:
+            web_cache[url] = response.text
+        return web_cache[url]
     else:
         error(
             f"Failed to fetch resource {click.style(url, bold=True, italic=True)}. Inform the Git-Mastery team."
         )
-    return ""
 
 
 def fetch_file_contents_or_none(
     url: str, is_binary: bool
 ) -> Optional[Union[str, bytes]]:
+    web_cache = get_web_cache()
+    if url in web_cache:
+        if get_verbose():
+            print(f"Fetching {url} contents from WEB_CACHE")
+        return web_cache[url]
     response = requests.get(url)
 
+    if get_verbose():
+        print(f"Querying {url} for contents")
+
     if response.status_code == 200:
         if is_binary:
-            return response.content
-        return response.text
+            web_cache[url] = response.content
+        else:
+            web_cache[url] = response.text
+        return web_cache[url]
     return None
 
 
@@ -177,10 +242,14 @@ def get_variable_from_url(
 
 def exercise_exists(exercise: str, timeout: int = 5) -> bool:
     try:
+        exercise_url = get_gitmastery_file_path(
+            f"{exercise.replace('-', '_')}/.gitmastery-exercise.json"
+        )
+        if get_verbose():
+            print(exercise_url)
+
         response = requests.head(
-            get_gitmastery_file_path(
-                f"{exercise.replace('-', '_')}/.gitmastery-exercise.json"
-            ),
+            exercise_url,
             allow_redirects=True,
             timeout=timeout,
         )
@@ -194,8 +263,14 @@ def hands_on_exists(hands_on: str, timeout: int = 5) -> bool:
         hands_on = hands_on[3:]
 
     try:
+        hands_on_url = get_gitmastery_file_path(
+            f"hands_on/{hands_on.replace('-', '_')}.py"
+        )
+        if get_verbose():
+            print(hands_on_url)
+
         response = requests.head(
-            get_gitmastery_file_path(f"hands_on/{hands_on.replace('-', '_')}.py"),
+            hands_on_url,
             allow_redirects=True,
             timeout=timeout,
         )