test(transforms): add tests for AWS transforms

.github/workflows/test_and_version.yml

@@ -27,12 +27,19 @@ jobs:
           ruff check .
   test:
     runs-on: ubuntu-latest
+    permissions:
+      id-token: write
     strategy:
       fail-fast: false
       matrix:
         python-version: ["3.9", "3.10", "3.11", "3.12"]
     steps:
       - uses: actions/checkout@v3
+      - name: Configure AWS Credentials
+        uses: aws-actions/configure-aws-credentials@v4
+        with:
+          aws-region: eu-west-1
+          role-to-assume: arn:aws:iam::826306421341:role/pycs-github-actions
       - name: Set up Python ${{ matrix.python-version }}
         uses: actions/setup-python@v3
         with:

pycs/transforms.py

@@ -10,6 +10,8 @@
 import yaml
 
 if TYPE_CHECKING:
+    import boto3
+
     from .node import CN
 
 
@@ -92,33 +94,51 @@ def get_updates(self, _) -> dict[str, Any] | None:
 class LoadFromAWSAppConfig(TransformBase):
     key: str
     required = False
+    session: boto3.Session | None = None
 
     def get_updates(self, cfg: CN) -> dict[str, Any] | None:
         try:
-            from appconfig_helper import AppConfigHelper
+            import boto3
         except ModuleNotFoundError as e:
             raise ImportError("Please install with aws extra: pip install pycs[aws]") from e
         if self.key not in cfg:
             raise ValueError(f"Can't find AppConfig key '{self.key}' in cfg")
         ac_cfg = cfg[self.key]
+
         required_keys = ["APP", "ENV", "PROFILE"]
         for key in required_keys:
             if key not in ac_cfg:
                 raise ValueError(f"Specified key ({self.key}) must contain {required_keys} subkeys, missing {key}")
+
         if not ac_cfg.APP:
             if self.required:
                 raise ValueError("Got empty APP for AppConfig")
             return None
-        appconfig = AppConfigHelper(ac_cfg.APP, ac_cfg.ENV, ac_cfg.PROFILE, fetch_on_read=True, max_config_age=600)
-        if not isinstance(appconfig.config, dict):
-            raise TypeError("Got invalid config from AppConfig")
-        return appconfig.config
+
+        client = (self.session or boto3).client("appconfigdata")
+        # Maybe actually create a proper client for this
+        config_token = client.start_configuration_session(
+            ApplicationIdentifier=ac_cfg.APP,
+            ConfigurationProfileIdentifier=ac_cfg.PROFILE,
+            EnvironmentIdentifier=ac_cfg.ENV,
+            RequiredMinimumPollIntervalInSeconds=15,
+        )["InitialConfigurationToken"]
+        response = client.get_latest_configuration(ConfigurationToken=config_token)
+
+        content_type = response["ContentType"]
+        content: bytes = response["Configuration"].read()
+        if content_type == "application/x-yaml":
+            return yaml.safe_load(content)
+        if content_type == "application/json":
+            return json.loads(content.decode("utf-8"))
+        raise ValueError(f"Got config in invalid type: {content_type}")
 
 
 @dataclass
 class LoadFromAWSSecretsManager(TransformBase):
     key: str
     required = False
+    session: boto3.Session | None = None
 
     def get_updates(self, cfg: CN) -> dict[str, Any] | None:
         try:
@@ -128,19 +148,14 @@ def get_updates(self, cfg: CN) -> dict[str, Any] | None:
         if self.key not in cfg:
             raise ValueError(f"Can't find SecretsManager key '{self.key}' in cfg")
         sm_cfg = cfg[self.key]
-        required_keys = ["NAME", "MAP"]
-        for key in required_keys:
-            if key not in sm_cfg:
-                raise ValueError(f"Specified key ({self.key}) must contain {required_keys} subkeys, missing {key}")
+
+        if "NAME" not in sm_cfg:
+            raise ValueError(f"Specified key ({self.key}) must contain Name subkey")
+
         if not sm_cfg.NAME:
             if self.required:
                 raise ValueError("Got empty NAME for SecretsManager")
             return None
 
-        secrets_manager = boto3.client("secretsmanager")
-        secrets = json.loads(secrets_manager.get_secret_value(SecretId=sm_cfg.NAME)["SecretString"])
-
-        changes = {}
-        for sm_key, target_key in sm_cfg.MAP.items():
-            changes[target_key] = secrets[sm_key]
-        return _flat_to_structured(changes)
+        secrets_manager = (self.session or boto3).client("secretsmanager")
+        return json.loads(secrets_manager.get_secret_value(SecretId=sm_cfg.NAME)["SecretString"])

pyproject.toml

@@ -19,9 +19,9 @@ cfg-output-dir = "pycs.helpers:_get_output_dir_cli"
 Home = "https://github.com/Rizhiy/pycs"
 
 [project.optional-dependencies]
-aws = ["sample-helper-aws-appconfig"]
-test = ["pytest", "pytest-cov", "types-pyyaml"]
-dev = ["black", "pre-commit", "pycs[test,aws]", "ruff"]
+aws = ["boto3"]
+test = ["pycs[aws]", "pytest", "pytest-cov", "types-pyyaml"]
+dev = ["black", "pre-commit", "pycs[test]", "ruff"]
 
 [tool.flit.sdist]
 include = ["README.md"]

tests/test_transforms.py

@@ -3,8 +3,16 @@
 import os
 from pathlib import Path
 
+import pytest
+
 from pycs import CN
-from pycs.transforms import LoadFromEnvVars, LoadFromFile, LoadFromKeyValue
+from pycs.transforms import (
+    LoadFromAWSAppConfig,
+    LoadFromAWSSecretsManager,
+    LoadFromEnvVars,
+    LoadFromFile,
+    LoadFromKeyValue,
+)
 
 DATA_DIR = Path(__file__).parent / "data" / "transforms"
 
@@ -52,3 +60,32 @@ def test_load_from_env_vars(monkeypatch):
     assert cfg_base.BOOL is False
     assert cfg.BOOL is True
     assert cfg.DICT.X == ""
+
+
+@pytest.mark.parametrize("type_", ["yaml", "json"])
+def test_load_from_aws_appconfig(type_: str):
+    from .data.base_cfg import schema as cfg_base
+
+    cfg = cfg_base.inherit()
+    cfg.APP_CONFIG = CN()
+    cfg.APP_CONFIG.APP = "pycs-test"
+    cfg.APP_CONFIG.PROFILE = type_
+    cfg.APP_CONFIG.ENV = "default"
+    cfg.add_transform(LoadFromAWSAppConfig("APP_CONFIG"))
+    cfg.freeze_schema()
+    cfg.transform()
+    assert cfg.NAME == "AppConfig"
+    assert type_ == cfg.STR
+
+
+def test_load_from_aws_secrets_manager():
+    from .data.base_cfg import schema as cfg_base
+
+    cfg = cfg_base.inherit()
+    cfg.SECRETS_MANAGER = CN()
+    cfg.SECRETS_MANAGER.NAME = "pycs-test"
+    cfg.add_transform(LoadFromAWSSecretsManager("SECRETS_MANAGER"))
+    cfg.freeze_schema()
+    cfg.transform()
+    assert cfg.NAME == "SecretsManager"
+    assert cfg.STR == "secret"