Skip to content

Commit

Permalink
feat: Adds get_job_from_key
Browse files Browse the repository at this point in the history
  • Loading branch information
Alan Christie committed Mar 20, 2023
1 parent 1e0731f commit 1addfa3
Show file tree
Hide file tree
Showing 2 changed files with 36 additions and 3 deletions.
15 changes: 12 additions & 3 deletions decoder/decoder.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,16 +47,25 @@

_GITHUB_REF_RE: Pattern[str] = re.compile(r"/([^/]+)/data-manager/")

_JOB_KEY_DELIMITER: str = "|"


class TextEncoding(enum.Enum):
"""A general text encoding format, used initially for Job text fields."""

JINJA2_3_0 = 1 # Encoding that complies with Jinja2 v3.0.x


def get_job_key(collection: str, job: str) -> str:
def get_job_key(*, collection: str, job: str) -> str:
"""Returns the job Key, a string formed from "<collection>|<job>."""
return f"{collection}|{job}"
return f"{collection}{_JOB_KEY_DELIMITER}{job}"


def get_job_from_key(*, key: str) -> Tuple[str, str]:
"""Returns the job Key "<collection>" and "<job>"."""
parts = key.split(_JOB_KEY_DELIMITER)
assert len(parts) == 2
return parts[0], parts[1]


def validate_manifest_schema(manifest: Dict[str, Any]) -> Optional[str]:
Expand Down Expand Up @@ -315,7 +324,7 @@ def get_jobs_replaced(job_definition: Dict[str, Any]) -> Optional[List[str]]:
for replaces in replaces_list:
r_collection: str = replaces["collection"]
r_job: str = replaces["job"]
replaced.add(get_job_key(r_collection, r_job))
replaced.add(get_job_key(collection=r_collection, job=r_job))
return list(replaced)


Expand Down
24 changes: 24 additions & 0 deletions tests/test_decoder.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,30 @@
from decoder import decoder


def test_get_job_key():
# Arrange
j_collection: str = "c-1"
j_job: str = "j-1"

# Act
key = decoder.get_job_key(collection=j_collection, job=j_job)

# Assert
assert key == "c-1|j-1"


def test_get_job_from_key():
# Arrange
j_key: str = "c-1|j-1"

# Act
j_collection, j_job = decoder.get_job_from_key(key=j_key)

# Assert
assert j_collection == "c-1"
assert j_job == "j-1"


def test_jinja2_3_0_decode():
# Arrange
text: str = "foo={{ foo }}, bar={{ bar }}, baz={{ baz }}"
Expand Down

0 comments on commit 1addfa3

Please sign in to comment.