entity-manager: add support for replacement while including
This commit enhances the feature by adding replacement of keyword
support while including json.
```
Let’s say, we have the following include json where we can see two
template parameters %%BOARD_NAME%% and %%SYSTEM_NAME%%.
// include/association_template.json
{
"Description": [
"This file has some keywords that needs to be replaced.",
"%%SYSTEM_NAME%%: the system name the mobo associated to.",
"%%BOARD_NAME%%: the board name the mobo and the system associated to.",
],
"xyz.openbmc_project.Association.Definitions": {
"Associations": [
{
"Forward": "contained_by",
"Path": "/xyz/openbmc_project/inventory/system/board/%%BOARD_NAME%%/%%SYSTEM_NAME%%",
"Reverse": "containing"
}
]
}
...
}
```
```
Normally in existing platforms we use `sed` to replace them and create a
new file. But since we are using include directive, we can pass the
value of these parameters and include feature will replace them while
generating the configuration. Here is an example of how we can use this
// mobo.json
{
"Description": [...],
"include": {
"path": "association_template.json",
"replace": [
{
"search": "%%BOARD_NAME%%",
"replace_with": "mobo_1"
},
{
"search": "%%SYSTEM_NAME%%",
"replace_with": "system1"
}
]
}
...
}
```
Best practices for templates:
* Always name the json file with `_template` suffix so that everyone knows it has a template.
* It is always good to have a similar kind of description in the include file to quickly understand what needs to be done to include this template.
* Keep consistent with the keywords throughout the configurations i.e use the same template parameter name %%SYSTEM_NAME%% in all templates instead of creating different template parameter names for the same purpose.
Fusion-Link:
platform11: https://fusion2.corp.google.com/25866add-0dbc-35a4-bade-7e590b173a04
platform5: https://fusion2.corp.google.com/f6fc2926-4463-3e83-a781-50012c9242db
platform15: https://fusion2.corp.google.com/a6b2d8da-c1d9-388a-a760-200ad80f73b9
platform17: https://fusion2.corp.google.com/5122a076-74fd-3cc2-a1b3-0d2eaf228c08
Tested: Unit test passed
Google-Bug-Id: 451442226
Change-Id: Ie3a5bc4be18ce58d19685c0217ad784b7299764e
Signed-off-by: Munawar Hussain <munawarhussain@google.com>
diff --git a/recipes-phosphor/configuration/entity-manager/scripts/config_preprocess.py b/recipes-phosphor/configuration/entity-manager/scripts/config_preprocess.py
index b33add0..30af316 100644
--- a/recipes-phosphor/configuration/entity-manager/scripts/config_preprocess.py
+++ b/recipes-phosphor/configuration/entity-manager/scripts/config_preprocess.py
@@ -9,7 +9,10 @@
import os
_INCLUDE_KEYWORD = "include"
-
+_INCLUDE_PATH_KEYWORD = "path"
+_INCLUDE_REPLACE_KEYWORD = "replace"
+_INCLUDE_REPLACE_SEARCH_KEYWORD = "search"
+_INCLUDE_REPLACE_REPLACE_WITH_KEYWORD = "replace_with"
# https://github.com/openbmc/entity-manager/blob/213b397d35168ef975a2bf5b6de7492dc00439c6/scripts/validate_configs.py#L34C1-L46C40
def remove_c_comments(string: str) -> str:
@@ -28,6 +31,74 @@
return regex.sub(_replacer, string)
+class IncludeConfig:
+ """Parses and holds configuration for an include statement.
+
+ This class handles both simple string includes (the path itself) and
+ more complex dict-based includes which can specify a path and
+ replacement rules.
+ """
+
+ def __init__(self, data: dict[str, Any] | str):
+ self.path = ""
+ self.replace = []
+ if isinstance(data, str):
+ self.path = data
+ return
+ if isinstance(data, dict):
+ if _INCLUDE_PATH_KEYWORD not in data:
+ print(
+ f"include property should have path if it is an object: {data}",
+ file=sys.stderr,
+ )
+ return
+ self.path = data[_INCLUDE_PATH_KEYWORD]
+ if _INCLUDE_REPLACE_KEYWORD in data:
+ if not isinstance(data[_INCLUDE_REPLACE_KEYWORD], list):
+ print(
+ f"replace property should have path if it is an object: {data}",
+ file=sys.stderr,
+ )
+ else:
+ self._load_replace(data[_INCLUDE_REPLACE_KEYWORD])
+
+ def _load_replace(self, data: list[dict[str, str]]):
+ """Loads replacement rules from the provided data.
+
+ Each entry in the data list should be a dictionary containing a 'search'
+ and a 'replace_with' key. Invalid entries are printed to stderr and skipped.
+
+ Args:
+ data: A list of dictionaries, where each dictionary specifies a search
+ string and its replacement.
+ """
+ for entry in data:
+ if not isinstance(entry, dict):
+ print(
+ f"entry in replace property should be object: {entry}",
+ file=sys.stderr,
+ )
+ continue
+ if (
+ _INCLUDE_REPLACE_SEARCH_KEYWORD not in entry
+ or _INCLUDE_REPLACE_REPLACE_WITH_KEYWORD not in entry
+ ):
+ print(
+ "entry in replace property should have replace_with and search"
+ f" keyword: {entry}",
+ file=sys.stderr,
+ )
+ continue
+ self.replace.append(entry)
+
+ def process_include_json_str(self, json_str: str) -> str:
+ resultant_json_str = json_str
+ for entry in self.replace:
+ resultant_json_str = resultant_json_str.replace(
+ entry[_INCLUDE_REPLACE_SEARCH_KEYWORD],
+ entry[_INCLUDE_REPLACE_REPLACE_WITH_KEYWORD],
+ )
+ return remove_c_comments(resultant_json_str)
def process_includes(
data: dict[str, Any] | list[Any],
@@ -41,8 +112,14 @@
visited_paths = set()
if isinstance(data, dict):
- if _INCLUDE_KEYWORD in data and isinstance(data[_INCLUDE_KEYWORD], str):
- include_path = include_folder / data[_INCLUDE_KEYWORD]
+ if _INCLUDE_KEYWORD in data and (
+ isinstance(data[_INCLUDE_KEYWORD], str)
+ or isinstance(data[_INCLUDE_KEYWORD], dict)
+ ):
+ include_config = IncludeConfig(data[_INCLUDE_KEYWORD])
+ if not include_config.path:
+ return
+ include_path = include_folder / include_config.path
if include_path in visited_paths:
print(
@@ -55,7 +132,9 @@
visited_paths.add(include_path)
try:
with open(include_path, "r") as include_file:
- include_data = json.loads(remove_c_comments(include_file.read()))
+ include_data = json.loads(
+ include_config.process_include_json_str(include_file.read())
+ )
del data[_INCLUDE_KEYWORD]
process_includes(include_data, include_folder, visited_paths)
if isinstance(include_data, list):
@@ -82,7 +161,11 @@
new_data = []
for item in data:
if isinstance(item, dict) and _INCLUDE_KEYWORD in item and len(item) == 1:
- include_path = include_folder / item[_INCLUDE_KEYWORD]
+ include_config = IncludeConfig(item[_INCLUDE_KEYWORD])
+ if not include_config.path:
+ continue
+
+ include_path = include_folder / include_config.path
if include_path in visited_paths:
print(
@@ -94,7 +177,9 @@
visited_paths.add(include_path)
try:
with open(include_path, "r") as include_file:
- include_data = json.loads(remove_c_comments(include_file.read()))
+ include_data = json.loads(
+ include_config.process_include_json_str(include_file.read())
+ )
if not isinstance(include_data, list):
print(
"JSON included into an array must be an array itself:",
diff --git a/recipes-phosphor/configuration/entity-manager/scripts/config_preprocess_test.py b/recipes-phosphor/configuration/entity-manager/scripts/config_preprocess_test.py
index b14b21e..8f7a7c6 100644
--- a/recipes-phosphor/configuration/entity-manager/scripts/config_preprocess_test.py
+++ b/recipes-phosphor/configuration/entity-manager/scripts/config_preprocess_test.py
@@ -156,6 +156,212 @@
output_data = json.load(f)
self.assertEqual(output_data, {"a": 1, "included": True})
+ def test_process_includes_with_replace(self):
+ """Tests process_includes with replace functionality."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ include_folder = pathlib.Path(tmpdir)
+ include_file = include_folder / "include.json"
+ with open(include_file, "w") as f:
+ f.write('{"key": "@@VALUE@@"}')
+
+ data = {
+ "include": {
+ "path": "include.json",
+ "replace": [{
+ "search": "@@VALUE@@",
+ "replace_with": "replaced_value"
+ }]
+ }
+ }
+ config_preprocess.process_includes(data, include_folder)
+ self.assertEqual(data, {"key": "replaced_value"})
+
+ def test_process_includes_with_replace_replace_not_list(self):
+ """Tests process_includes with replace functionality."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ include_folder = pathlib.Path(tmpdir)
+ include_file = include_folder / "include.json"
+ with open(include_file, "w") as f:
+ f.write('{"key": "@@VALUE@@"}')
+
+ data = {
+ "include": {
+ "path": "include.json",
+ "replace": {
+ "search": "@@VALUE@@",
+ "replace_with": "replaced_value",
+ },
+ }
+ }
+ with mock.patch("sys.stderr", new_callable=io.StringIO) as mock_stderr:
+ config_preprocess.process_includes(data, include_folder)
+ self.assertIn(
+ "replace property should have path if it is an object",
+ mock_stderr.getvalue(),
+ )
+
+ def test_process_includes_with_replace_entry_not_dict(self):
+ """Tests process_includes with replace functionality."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ include_folder = pathlib.Path(tmpdir)
+ include_file = include_folder / "include.json"
+ with open(include_file, "w") as f:
+ f.write('{"key": "@@VALUE@@"}')
+
+ data = {
+ "include": {
+ "path": "include.json",
+ "replace": ["not a dict"],
+ }
+ }
+ with mock.patch("sys.stderr", new_callable=io.StringIO) as mock_stderr:
+ config_preprocess.process_includes(data, include_folder)
+ self.assertIn(
+ "entry in replace property should be object",
+ mock_stderr.getvalue(),
+ )
+
+ def test_process_includes_with_replace_entry_missing_keyword(self):
+ """Tests process_includes with replace functionality."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ include_folder = pathlib.Path(tmpdir)
+ include_file = include_folder / "include.json"
+ with open(include_file, "w") as f:
+ f.write('{"key": "@@VALUE@@"}')
+
+ data = {
+ "include": {
+ "path": "include.json",
+ "replace": [{
+ "search": "@@VALUE@@",
+ }],
+ }
+ }
+ with mock.patch("sys.stderr", new_callable=io.StringIO) as mock_stderr:
+ config_preprocess.process_includes(data, include_folder)
+ self.assertIn(
+ "entry in replace property should have replace_with and "
+ "search keyword",
+ mock_stderr.getvalue(),
+ )
+
+ def test_process_includes_with_replace_entry_missing_search_keyword(self):
+ """Tests process_includes with replace functionality."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ include_folder = pathlib.Path(tmpdir)
+ include_file = include_folder / "include.json"
+ with open(include_file, "w") as f:
+ f.write('{"key": "@@VALUE@@"}')
+
+ data = {
+ "include": {
+ "path": "include.json",
+ "replace": [{
+ "replace_with": "replaced_value",
+ }],
+ }
+ }
+ with mock.patch("sys.stderr", new_callable=io.StringIO) as mock_stderr:
+ config_preprocess.process_includes(data, include_folder)
+ self.assertIn(
+ "entry in replace property should have replace_with and "
+ "search keyword",
+ mock_stderr.getvalue(),
+ )
+
+ def test_process_config_file_with_replace(self):
+ """Tests the process_config_file function with replace."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ tmpdir_path = pathlib.Path(tmpdir)
+ input_folder = tmpdir_path / "input"
+ include_folder = tmpdir_path / "include"
+ output_folder = tmpdir_path / "output"
+ input_folder.mkdir()
+ include_folder.mkdir()
+ output_folder.mkdir()
+
+ # Create include file
+ include_file = include_folder / "include.json"
+ with open(include_file, "w") as f:
+ f.write('{"key": "@@VALUE@@"}')
+
+ # Create input file
+ input_file = input_folder / "config.json"
+ with open(input_file, "w") as f:
+ json.dump(
+ {
+ "a": 1,
+ "include": {
+ "path": "include.json",
+ "replace": [{
+ "search": "@@VALUE@@",
+ "replace_with": "replaced_value",
+ }],
+ },
+ },
+ f,
+ )
+
+ args = argparse.Namespace(
+ include_folder=include_folder, output_folder=output_folder
+ )
+
+ config_preprocess.process_config_file(input_file, args)
+
+ output_file = output_folder / "config.json"
+ self.assertTrue(output_file.exists())
+ with open(output_file, "r") as f:
+ output_data = json.load(f)
+ self.assertEqual(output_data, {"a": 1, "key": "replaced_value"})
+
+ def test_process_includes_list_with_replace(self):
+ """Tests process_includes with list includes and replace."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ include_folder = pathlib.Path(tmpdir)
+ include_file = include_folder / "include.json"
+ with open(include_file, "w") as f:
+ f.write('[{"item": "@@VALUE@@"}]')
+
+ data = [
+ {"item": 1},
+ {
+ "include": {
+ "path": "include.json",
+ "replace": [{
+ "search": "@@VALUE@@",
+ "replace_with": "replaced_value",
+ }],
+ }
+ },
+ {"item": 4},
+ ]
+ config_preprocess.process_includes(data, include_folder)
+ self.assertEqual(
+ data, [{"item": 1}, {"item": "replaced_value"}, {"item": 4}]
+ )
+
+ def test_process_includes_with_multiple_replaces(self):
+ """Tests process_includes with multiple replace operations."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ include_folder = pathlib.Path(tmpdir)
+ include_file = include_folder / "include.json"
+ with open(include_file, "w") as f:
+ f.write('{"key1": "@@VALUE1@@", "key2": "@@VALUE2@@"}')
+
+ data = {
+ "include": {
+ "path": "include.json",
+ "replace": [
+ {"search": "@@VALUE1@@", "replace_with": "replaced_value1"},
+ {"search": "@@VALUE2@@", "replace_with": "replaced_value2"},
+ ],
+ }
+ }
+ config_preprocess.process_includes(data, include_folder)
+ self.assertEqual(
+ data, {"key1": "replaced_value1", "key2": "replaced_value2"}
+ )
+
if __name__ == "__main__":
unittest.main()
