Add option to specify templates_branch on a primary group, to use that

branch when generating config for devices in that primary group

docs/reporef/index.rst

@@ -153,6 +153,10 @@ and that key contains a dictionary with two keys:
   have the highest priority when determining the primary group for a device.
   Higher value means higher priority. Defaults to 0, value of 1 is reserved
   for builtin group DEFAULT.
+- templates_branch: Optional string that specifies an alternative git branch
+  in the templates repository to use for devices in this primary group. Make
+  sure the branch exists in the templates repository and that the templates
+  repository is refreshed before setting this value or you will get an error.
 
 There will always exist a group called DEFAULT with group_priority 1 even
 if it's not specified in groups.yml.
@@ -179,6 +183,7 @@ All devices that matches the regex will be included in the group.
          name: 'E1'
          regex: 'eosdist1$'
          group_priority: 100
+         templates_branch: "new_dist_features"
      - group:
          name: 'E'
          regex: 'eosdist.*'

src/cnaas_nms/db/git.py

@@ -11,6 +11,7 @@
 from cnaas_nms.app_settings import app_settings
 from cnaas_nms.db.device import Device, DeviceType
 from cnaas_nms.db.exceptions import ConfigException, RepoStructureException
+from cnaas_nms.db.git_worktrees import WorktreeError, clean_templates_worktree
 from cnaas_nms.db.job import Job, JobStatus
 from cnaas_nms.db.joblock import Joblock, JoblockError
 from cnaas_nms.db.session import redis_session, sqla_session
@@ -259,6 +260,10 @@ def _refresh_repo_task(repo_type: RepoType = RepoType.TEMPLATES, job_id: Optiona
             if repo_chekout_working(repo_type):
                 rebuild_settings_cache()
             raise e
+        except WorktreeError as e:
+            if repo_chekout_working(repo_type):
+                rebuild_settings_cache()
+            raise e
         else:
             try:
                 repo_save_working_commit(repo_type, local_repo.head.commit.hexsha)
@@ -297,6 +302,7 @@ def _refresh_repo_task(repo_type: RepoType = RepoType.TEMPLATES, job_id: Optiona
             devtype: DeviceType
             for devtype, platform in updated_devtypes:
                 Device.set_devtype_syncstatus(session, devtype, ret, "templates", platform, job_id)
+        clean_templates_worktree()
 
     return ret
 

src/cnaas_nms/db/git_worktrees.py

@@ -0,0 +1,50 @@
+import os
+import shutil
+from typing import Optional
+
+import git.exc
+from cnaas_nms.app_settings import app_settings
+from cnaas_nms.tools.log import get_logger
+from git import Repo
+
+
+class WorktreeError(Exception):
+    pass
+
+
+def clean_templates_worktree():
+    if os.path.isdir("/tmp/worktrees"):
+        for subdir in os.listdir("/tmp/worktrees"):
+            shutil.rmtree("/tmp/worktrees/" + subdir, ignore_errors=True)
+
+    local_repo = Repo(app_settings.TEMPLATES_LOCAL)
+    local_repo.git.worktree("prune")
+
+
+def get_branch_folder(branch: str) -> str:
+    return os.path.join("/tmp/worktrees/", branch.replace("/", "__"))
+
+
+def refresh_templates_worktree(branch: str):
+    """Add worktree for specified branch in separate folder"""
+    logger = get_logger()
+    branch_folder = get_branch_folder(branch)
+    if os.path.isdir(branch_folder):
+        return
+    local_repo = Repo(app_settings.TEMPLATES_LOCAL)
+    if not os.path.isdir("/tmp/worktrees"):
+        os.mkdir("/tmp/worktrees")
+    logger.debug("Adding worktree for templates branch {} in folder {}".format(branch, branch_folder))
+    try:
+        local_repo.git.worktree("add", branch_folder, branch)
+    except git.exc.GitCommandError as e:
+        logger.error("Error adding worktree for templates branch {}: {}".format(branch, e.stderr.strip()))
+        raise WorktreeError(e.stderr.strip())
+
+
+def find_templates_worktree_path(branch: str) -> Optional[str]:
+    branch_folter = get_branch_folder(branch)
+    if os.path.isdir(branch_folter):
+        return branch_folter
+    else:
+        return None

src/cnaas_nms/db/settings.py

@@ -2,7 +2,7 @@
 import json
 import os
 import re
-from typing import Dict, List, Optional, Set, Tuple, Union
+from typing import Any, Dict, List, Optional, Set, Tuple, Union
 
 import pkg_resources
 import yaml
@@ -12,6 +12,7 @@
 
 from cnaas_nms.app_settings import api_settings, app_settings
 from cnaas_nms.db.device import Device, DeviceState, DeviceType
+from cnaas_nms.db.git_worktrees import refresh_templates_worktree
 from cnaas_nms.db.mgmtdomain import Mgmtdomain
 from cnaas_nms.db.session import redis_session, sqla_session
 from cnaas_nms.db.settings_fields import f_groups
@@ -612,8 +613,12 @@ def get_settings(
         )
         settings = get_downstream_dependencies(hostname, settings)
         # 5. Get settings repo group specific settings
-        if hostname in get_device_primary_groups():
-            primary_group = get_device_primary_groups()[hostname]
+        primary_group = get_device_primary_groups().get(hostname)
+        if primary_group:
+            # add templates worktree
+            templates_branch = get_group_templates_branch(primary_group)
+            if templates_branch:
+                refresh_templates_worktree(templates_branch)
             if os.path.isdir(os.path.join(local_repo_path, "groups", primary_group)):
                 settings, settings_origin = read_settings(
                     local_repo_path,
@@ -752,18 +757,30 @@ def get_groups(hostname: Optional[str] = None) -> List[str]:
 def get_group_regex(group_name: str) -> Optional[str]:
     """Returns a string containing the regex defining the specified
     group name if it's found."""
-    settings, origin = get_group_settings()
+    return get_group_settings_asdict().get(group_name, {}).get("regex")
+
+
+def get_group_templates_branch(group_name: str) -> Optional[str]:
+    """Returns a string containing the regex defining the specified
+    group name if it's found."""
+    return get_group_settings_asdict().get(group_name, {}).get("templates_branch")
+
+
+@redis_lru_cache
+def get_group_settings_asdict() -> Dict[str, Dict[str, Any]]:
+    """Returns a dict with group name as key and other parameters as values"""
+    settings, _ = get_group_settings()
     if not settings:
-        return None
-    if not settings.get("groups", None):
-        return None
+        return {}
+    if not settings.get("groups"):
+        return {}
+    group_dict: Dict[str, Dict[str, Any]] = {}
     for group in settings["groups"]:
         if "name" not in group["group"]:
             continue
-        if "regex" not in group["group"]:
-            continue
-        if group_name == group["group"]["name"]:
-            return group["group"]["regex"]
+        group_dict[group["group"]["name"]] = group["group"]
+        del group_dict[group["group"]["name"]]["name"]
+    return group_dict
 
 
 def get_groups_priorities(hostname: Optional[str] = None, settings: Optional[dict] = None) -> Dict[str, int]:

src/cnaas_nms/db/settings_fields.py

@@ -386,6 +386,7 @@ class f_group_item(BaseModel):
     name: str = group_name
     regex: str = ""
     group_priority: int = group_priority_schema
+    templates_branch: Optional[str] = None
 
     @field_validator("group_priority")
     @classmethod
@@ -394,6 +395,13 @@ def reserved_priority(cls, v: int, info: FieldValidationInfo):
             raise ValueError("group_priority 1 is reserved for built-in group DEFAULT")
         return v
 
+    @field_validator("templates_branch")
+    @classmethod
+    def templates_branch_primary_group_only(cls, v: str, info: FieldValidationInfo):
+        if v and info.data["group_priority"] <= 1:
+            raise ValueError("templates_branch can only be specified on primary groups")
+        return v
+
 
 class f_group(BaseModel):
     group: Optional[f_group_item] = None

src/cnaas_nms/db/tests/test_settings.py

@@ -4,6 +4,7 @@
 import pkg_resources
 import pytest
 import yaml
+from pydantic import ValidationError
 
 from cnaas_nms.db.device import DeviceType
 from cnaas_nms.db.settings import (
@@ -17,6 +18,7 @@
     get_settings,
     verify_dir_structure,
 )
+from cnaas_nms.db.settings_fields import f_groups
 
 
 class SettingsTests(unittest.TestCase):
@@ -208,6 +210,33 @@ def test_groups_priorities_collission(self):
         del group_settings_dict["groups"][2]
         self.assertIsNone(check_group_priority_collisions(group_settings_dict))
 
+    def test_groups_templates_braches(self):
+        group_settings_dict = {
+            "groups": [
+                {
+                    "group": {"name": "DEFAULT", "group_priority": 1},
+                },
+                {
+                    "group": {
+                        "name": "TEMPLATE1",
+                        "regex": "eosdist1$",
+                        "group_priority": 100,
+                        "templates_branch": "test1",
+                    }
+                },
+                {"group": {"name": "NOT_PRIMARY_GROUP", "templates_branch": "test2"}},
+            ]
+        }
+        with self.assertRaises(
+            ValidationError,
+            msg="Group with template_branch set but no group_priority value should raise ValidationError",
+        ):
+            f_groups(**group_settings_dict).model_dump()
+
+        # Remove bad entry
+        del group_settings_dict["groups"][2]
+        f_groups(**group_settings_dict).model_dump()
+
 
 if __name__ == "__main__":
     unittest.main()

src/cnaas_nms/devicehandler/sync_devices.py

@@ -19,11 +19,12 @@
 from cnaas_nms.db.device import Device, DeviceState, DeviceType
 from cnaas_nms.db.device_vars import expand_interface_settings
 from cnaas_nms.db.git import RepoStructureException
+from cnaas_nms.db.git_worktrees import find_templates_worktree_path
 from cnaas_nms.db.interface import Interface
 from cnaas_nms.db.job import Job
 from cnaas_nms.db.joblock import Joblock, JoblockError
 from cnaas_nms.db.session import redis_session, sqla_session
-from cnaas_nms.db.settings import get_settings
+from cnaas_nms.db.settings import get_device_primary_groups, get_group_templates_branch, get_settings
 from cnaas_nms.devicehandler.changescore import calculate_score
 from cnaas_nms.devicehandler.get import calc_config_hash
 from cnaas_nms.devicehandler.nornir_helper import NornirJobResult, cnaas_init, get_jinja_env, inventory_selector
@@ -518,6 +519,15 @@ def push_sync_device(
 
     local_repo_path = app_settings.TEMPLATES_LOCAL
 
+    # override template path if primary group template path is set
+    primary_group = get_device_primary_groups().get(hostname)
+    if primary_group:
+        templates_branch = get_group_templates_branch(primary_group)
+        if templates_branch:
+            primary_group_template_path = find_templates_worktree_path(templates_branch)
+            if primary_group_template_path:
+                local_repo_path = primary_group_template_path
+
     mapfile = os.path.join(local_repo_path, platform, "mapping.yml")
     if not os.path.isfile(mapfile):
         raise RepoStructureException("File {} not found in template repo".format(mapfile))