feat: Add support to check validity of collection and job names

Author: Alan Christie

decoder/decoder.py

@@ -8,7 +8,7 @@
 import enum
 import os
 import re
-from typing import Any, Dict, List, Optional, Pattern, Set, Tuple
+from typing import Any, Dict, List, Optional, Set, Tuple
 
 import jsonschema
 import yaml
@@ -46,7 +46,11 @@
 REPO_TYPE_GITLAB: str = "gitlab"
 _REPO_TYPES: List[str] = [REPO_TYPE_GITHUB, REPO_TYPE_GITLAB]
 
-_GITHUB_REF_RE: Pattern[str] = re.compile(r"/([^/]+)/data-manager/")
+_GITHUB_REF_RE: re.Pattern = re.compile(r"/([^/]+)/data-manager/")
+
+# Patterns for Collection and Job names
+_COLLECTION_NAME_RE: re.Pattern = re.compile(r"([a-z]{1}[a-z0-9-]{0,79})")
+_JOB_NAME_RE: re.Pattern = re.compile(r"([a-z]{1}[a-z0-9-]{0,79})")
 
 _JOB_KEY_DELIMITER: str = "|"
 
@@ -57,6 +61,16 @@ class TextEncoding(enum.Enum):
     JINJA2_3_0 = 1  # Encoding that complies with Jinja2 v3.0.x
 
 
+def is_valid_collection_name(collection: str) -> bool:
+    """Returns True if the collection name is valid"""
+    return _COLLECTION_NAME_RE.fullmatch(collection) is not None
+
+
+def is_valid_job_name(job: str) -> bool:
+    """Returns True if the collection name is valid"""
+    return _JOB_NAME_RE.fullmatch(job) is not None
+
+
 def get_job_key(*, collection: str, job: str) -> str:
     """Returns the job Key, a string formed from "<collection>|<job>."""
     return f"{collection}{_JOB_KEY_DELIMITER}{job}"

tests/test_decoder.py

@@ -76,3 +76,43 @@ def test_jinja2_3_0_decode_without_variables():
     # Assert
     assert success
     assert rendered == expected_text
+
+
+def test_is_valid_collection_name_when_not_valid():
+    # Arrange
+
+    # Act
+    success = decoder.is_valid_collection_name("blob.yaml")
+
+    # Assert
+    assert not success
+
+
+def test_is_valid_collection_name_when_valid():
+    # Arrange
+
+    # Act
+    success = decoder.is_valid_collection_name("collection-1")
+
+    # Assert
+    assert success
+
+
+def test_is_valid_job_name_when_not_valid():
+    # Arrange
+
+    # Act
+    success = decoder.is_valid_job_name("blob.yaml")
+
+    # Assert
+    assert not success
+
+
+def test_is_valid_job_name_when_valid():
+    # Arrange
+
+    # Act
+    success = decoder.is_valid_job_name("job-1")
+
+    # Assert
+    assert success