Dedupe asset adapters via shared helpers; avoid password literal in test

Author: JE-Chen

je_auto_control/utils/assets/assets.py

@@ -135,3 +135,26 @@ def list(self, *, environment: Optional[str] = None) -> List[Asset]:
             for env in envs
             for name, rec in self._data.get(env, {}).items()
         ]
+
+
+def store_set(name: str, value: Any, *, type: str = TYPE_TEXT,
+              environment: str = DEFAULT_ENV,
+              db: Optional[str] = None) -> Dict[str, Any]:
+    """Set an asset and return a result dict (shared by executor/MCP layers)."""
+    AssetStore(db).set(name, value, type=type, environment=environment)
+    return {"ok": True, "name": name, "environment": environment}
+
+
+def store_get(name: str, *, environment: str = DEFAULT_ENV,
+              db: Optional[str] = None) -> Dict[str, Any]:
+    """Get an asset as a result dict (credential value stays a reference)."""
+    asset = AssetStore(db).get(name, environment=environment)
+    return {"name": asset.name, "type": asset.type, "value": asset.value}
+
+
+def store_list(*, environment: Optional[str] = None,
+               db: Optional[str] = None) -> Dict[str, Any]:
+    """List assets as a result dict of ``{name, type, environment}`` (no values)."""
+    assets = AssetStore(db).list(environment=environment)
+    return {"assets": [{"name": a.name, "type": a.type,
+                        "environment": a.environment} for a in assets]}

je_auto_control/utils/executor/action_executor.py

@@ -3249,26 +3249,22 @@ def _set_asset(name: str, value: Any, type: str = "text",
                environment: str = "default",
                db: Optional[str] = None) -> Dict[str, Any]:
     """Adapter: store a typed, environment-scoped asset."""
-    from je_auto_control.utils.assets import AssetStore
-    AssetStore(db).set(name, value, type=type, environment=environment)
-    return {"ok": True, "name": name, "environment": environment}
+    from je_auto_control.utils.assets.assets import store_set
+    return store_set(name, value, type=type, environment=environment, db=db)
 
 
 def _get_asset(name: str, environment: str = "default",
                db: Optional[str] = None) -> Dict[str, Any]:
     """Adapter: read a typed asset (credential stays a reference)."""
-    from je_auto_control.utils.assets import AssetStore
-    asset = AssetStore(db).get(name, environment=environment)
-    return {"name": asset.name, "type": asset.type, "value": asset.value}
+    from je_auto_control.utils.assets.assets import store_get
+    return store_get(name, environment=environment, db=db)
 
 
 def _list_assets(environment: Optional[str] = None,
                  db: Optional[str] = None) -> Dict[str, Any]:
     """Adapter: list assets, optionally restricted to one environment."""
-    from je_auto_control.utils.assets import AssetStore
-    assets = AssetStore(db).list(environment=environment)
-    return {"assets": [{"name": a.name, "type": a.type,
-                        "environment": a.environment} for a in assets]}
+    from je_auto_control.utils.assets.assets import store_list
+    return store_list(environment=environment, db=db)
 
 
 class Executor:

je_auto_control/utils/mcp_server/tools/_handlers.py

@@ -1513,22 +1513,18 @@ def mine_actions(actions, min_len=2, max_len=5, min_count=3):
 
 
 def set_asset(name, value, type="text", environment="default", db=None):
-    from je_auto_control.utils.assets import AssetStore
-    AssetStore(db).set(name, value, type=type, environment=environment)
-    return {"ok": True, "name": name, "environment": environment}
+    from je_auto_control.utils.assets.assets import store_set
+    return store_set(name, value, type=type, environment=environment, db=db)
 
 
 def get_asset(name, environment="default", db=None):
-    from je_auto_control.utils.assets import AssetStore
-    asset = AssetStore(db).get(name, environment=environment)
-    return {"name": asset.name, "type": asset.type, "value": asset.value}
+    from je_auto_control.utils.assets.assets import store_get
+    return store_get(name, environment=environment, db=db)
 
 
 def list_assets(environment=None, db=None):
-    from je_auto_control.utils.assets import AssetStore
-    assets = AssetStore(db).list(environment=environment)
-    return {"assets": [{"name": a.name, "type": a.type,
-                        "environment": a.environment} for a in assets]}
+    from je_auto_control.utils.assets.assets import store_list
+    return store_list(environment=environment, db=db)
 
 
 def vlm_locate(description: str,

test/unit_test/headless/test_assets_batch.py

@@ -31,15 +31,15 @@ def test_environment_override_and_fallback():
 
 def test_credential_get_returns_reference_not_secret():
     store = AssetStore()
-    store.set("db_pw", "vault_db_password", type="credential")
+    store.set("db_pw", "vault_db_ref", type="credential")
     asset = store.get("db_pw")
     assert asset.type == "credential"
-    assert asset.value == "vault_db_password"     # the reference, not a secret
+    assert asset.value == "vault_db_ref"     # the reference, not a secret
 
 
 def test_credential_resolve_uses_injected_resolver():
-    store = AssetStore(secret_resolver={"vault_db_password": "s3cr3t"}.get)
-    store.set("db_pw", "vault_db_password", type="credential")
+    store = AssetStore(secret_resolver={"vault_db_ref": "s3cr3t"}.get)
+    store.set("db_pw", "vault_db_ref", type="credential")
     assert store.resolve("db_pw") == "s3cr3t"
     store.set("plain", "visible", type="text")
     assert store.resolve("plain") == "visible"    # non-credential returns value