Reworked utility funcs for reading and writing base64 files

kupydo/internal/kube_models/namespaced/configmap.py

@@ -9,10 +9,12 @@
 # SPDX-License-Identifier: MIT
 #
 from __future__ import annotations as anno
+from pathlib import Path
 from dotmap import DotMap
 from kubernetes_asyncio import client
 from kupydo.internal.types import *
 from kupydo.internal.base import *
+from kupydo.internal import errors
 from kupydo.internal import utils
 
 
@@ -21,7 +23,7 @@
 
 class ConfigMapValues(KupydoBaseValues):
  data: OptionalDictStr
- binary_data: OptionalListStr
+ binary_data: OptionalDictStr
  immutable: OptionalBool
 
 
@@ -72,7 +74,11 @@ def _api(self, session: client.ApiClient) -> KupydoApiActions:
  def _read_binary_files(files: list[str] | None) -> dict[str, str] | None:
  if files:
  encoded_files = dict()
- for file_name in files:
- data = utils.read_encode_rel_file(file_name)
- encoded_files[file_name] = data
+ for file_path in files:
+ if utils.is_path_absolute(file_path):
+ raise errors.InvalidPathTypeError(file_path, "relative")
+ ext_file = utils.first_external_caller()[0]
+ bin_file = (ext_file.parent / file_path).resolve()
+ data = utils.read_encode_b64_file(bin_file)
+ encoded_files[Path(file_path).name] = data
  return encoded_files

kupydo/internal/kube_models/namespaced/secret.py

@@ -90,15 +90,19 @@ def _resolve_secret(keyword: str, value: str, from_file: bool = False) -> str:
  secret = GlobalRegistry.get_secret(enc_tag)
  return secret.secret_value
 
- file_path, from_line = utils.first_external_caller()
- lines = utils.read_cached_file_lines(file_path)
+ ext_file, from_line = utils.first_external_caller()
+ lines = utils.read_cached_file_lines(ext_file)
  lineno = sec_ops.find_kwarg_line(lines, from_line, keyword, value)
- secret = utils.read_encode_rel_file(value) if from_file else value
+
+ secret = value
+ if from_file:
+ sec_file = (ext_file.parent / value).resolve()
+ secret = utils.read_encode_b64_file(sec_file)
 
  if GlobalRegistry.is_enabled():
  sfd = sec_ops.SecretFieldDetails(
  enc_tag=sec_ops.generate_enc_tag(),
- file_path=file_path,
+ file_path=ext_file,
  line_number=lineno,
  field_keyword=keyword,
  field_value=value,

kupydo/internal/utils/__init__.py

@@ -17,7 +17,8 @@
 
 __all__ = [
  "match_dict_structure",
- "read_encode_rel_file",
+ "read_encode_b64_file",
+ "write_decode_b64_file",
  "read_cached_file_lines",
  "generate_name",
  "deep_merge",

kupydo/internal/utils/file_utils.py

@@ -12,25 +12,34 @@
 import functools
 from pathlib import Path
 from kupydo.internal import errors
-from .trace_utils import first_external_caller
 from .path_utils import is_path_absolute
 
 
 __all__ = [
- "read_encode_rel_file",
+ "read_encode_b64_file",
+ "write_decode_b64_file",
  "read_cached_file_lines"
 ]
 
 
-def read_encode_rel_file(rel_fp: str) -> str:
- if is_path_absolute(rel_fp):
- raise errors.PathNotRelativeError(rel_fp)
+def read_encode_b64_file(abs_fp: Path | str) -> str:
+ abs_fp = Path(abs_fp)
+ if not is_path_absolute(abs_fp):
+ raise errors.InvalidPathTypeError(
+ abs_fp.as_posix(), "absolute"
+ )
+ with abs_fp.open('rb') as file:
+ return base64.b64encode(file.read()).decode()
 
- caller_path = first_external_caller()[0]
- target = (caller_path.parent / rel_fp).resolve()
 
- with target.open('rb') as file:
- return base64.b64encode(file.read()).decode()
+def write_decode_b64_file(abs_fp: Path | str, b64_data: str) -> None:
+ abs_fp = Path(abs_fp)
+ if not is_path_absolute(abs_fp):
+ raise errors.InvalidPathTypeError(
+ abs_fp.as_posix(), "absolute"
+ )
+ with abs_fp.open('wb') as file:
+ file.write(base64.b64decode(b64_data.encode()))
 
 
 @functools.cache

tests/test_sec_ops/test_file_tools/test_write_secret_files.py

@@ -19,12 +19,13 @@ def test_with_valid_data(mocker, tmp_path):
  rel_path = 'clusters/staging/Heart.py'
  sfd_list = [
  sec_ops.SecretFieldDetails(
+ enc_tag="id1",
  file_path=heart_path,
  line_number=1,
  field_keyword="key",
  field_value="value",
  secret_value="secret",
- enc_tag="id1"
+ from_file=False
  )
  ]
  mocked_rel_path = mocker.patch.object(

tests/test_utils/test_file_utils.py

@@ -8,26 +8,49 @@
 #
 # SPDX-License-Identifier: MIT
 #
-import site
-import base64
 import pytest
-import textwrap
-import subprocess
 from pathlib import Path
+from typing import Callable
 from kupydo.internal import errors
 from kupydo.internal import utils
 
 
 @pytest.fixture(name="create_tmp_file")
 def fixture_create_tmp_file(tmp_path: Path):
- def create_file(content, filename="temp_file.txt"):
- file_path = tmp_path / filename
- file_path.write_text(content)
+ def create_file(content: str = None):
+ file_path = tmp_path / "tmp_file.txt"
+ file_path.touch()
+ if content:
+ file_path.write_text(content)
  return file_path
  return create_file
 
 
-def test_file_lines(create_tmp_file):
+def test_read_encode_b64_file(create_tmp_file: Callable):
+ path = create_tmp_file("Hello World")
+ result = utils.read_encode_b64_file(path)
+ assert result == "SGVsbG8gV29ybGQ="
+
+
+def test_write_decode_b64_file(create_tmp_file: Callable):
+ path = create_tmp_file()
+ utils.write_decode_b64_file(path, "SGVsbG8gV29ybGQ=")
+ assert path.read_text() == "Hello World"
+
+
+def test_read_encode_b64_file_invalid_rel_path():
+ for inv_path in ["./home/user/file", "home/user/file"]:
+ with pytest.raises(errors.InvalidPathTypeError):
+ utils.read_encode_b64_file(inv_path)
+
+
+def test_write_decode_b64_file_invalid_rel_path():
+ for inv_path in ["./home/user/file", "home/user/file"]:
+ with pytest.raises(errors.InvalidPathTypeError):
+ utils.write_decode_b64_file(inv_path, '')
+
+
+def test_read_cached_file_lines(create_tmp_file):
  content = "Line 1\nLine 2\nLine 3"
  file_path = create_tmp_file(content)
  expected_output = ["Line 1\n", "Line 2\n", "Line 3"]
@@ -36,7 +59,7 @@ def test_file_lines(create_tmp_file):
  utils.read_cached_file_lines.cache_clear()
 
 
-def test_same_file(create_tmp_file):
+def test_read_cached_same_file(create_tmp_file):
  content = "Line 1\nLine 2\nLine 3"
  file_path = create_tmp_file(content)
 
@@ -49,7 +72,7 @@ def test_same_file(create_tmp_file):
  utils.read_cached_file_lines.cache_clear()
 
 
-def test_after_file_change(create_tmp_file):
+def test_read_cached_after_file_change(create_tmp_file):
  content = "Line 1\nLine 2\nLine 3"
  file_path = create_tmp_file(content)
 
@@ -63,42 +86,3 @@ def test_after_file_change(create_tmp_file):
  assert initial_result == updated_result, \
  "Function should not detect file content change due to caching"
  utils.read_cached_file_lines.cache_clear()
-
-
-def test_relative_file_path(tmp_path: Path):
- pkgs = site.getsitepackages()
- lib = Path(__file__).parents[2].as_posix()
- pkgs.append(lib)
-
- data_path = tmp_path / "test_data.py"
- script_path = tmp_path / "test_script.py"
-
- script = textwrap.dedent(f'''
- import sys
- for path in {pkgs}:
- sys.path.append(path)
-
- from kupydo.internal import utils
-
- def caller():
- encoded = utils.read_encode_rel_file("test_data.py")
- print(encoded)
-
- caller()
- ''')
- data_path.write_text("Hello World")
- script_path.write_text(script)
- result = subprocess.run(
- args=['python', script_path],
- capture_output=True,
- text=True
- )
- expected_result = base64.b64encode("Hello World".encode()).decode()
- assert result.stdout.strip() == expected_result, \
- "The output should be the base64 encoded string of 'Hello World'"
-
-
-def test_absolute_file_path():
- with pytest.raises(errors.PathNotRelativeError):
- abs_path = Path(__file__).resolve().as_posix()
- utils.read_encode_rel_file(abs_path)