added new funcs to so-yaml.py to support gemini tests

This commit is contained in:
Matthew Wright
2026-02-19 16:20:44 -05:00
parent 93f52453b4
commit 7fa01f5fd5
2 changed files with 464 additions and 11 deletions

View File

@@ -9,6 +9,7 @@ import os
import sys
import time
import yaml
import json
lockFile = "/tmp/so-yaml.lock"
@@ -16,19 +17,24 @@ lockFile = "/tmp/so-yaml.lock"
def showUsage(args):
print('Usage: {} <COMMAND> <YAML_FILE> [ARGS...]'.format(sys.argv[0]), file=sys.stderr)
print(' General commands:', file=sys.stderr)
print(' append - Append a list item to a yaml key, if it exists and is a list. Requires KEY and LISTITEM args.', file=sys.stderr)
print(' removelistitem - Remove a list item from a yaml key, if it exists and is a list. Requires KEY and LISTITEM args.', file=sys.stderr)
print(' add - Add a new key and set its value. Fails if key already exists. Requires KEY and VALUE args.', file=sys.stderr)
print(' get - Displays (to stdout) the value stored in the given key. Requires KEY arg.', file=sys.stderr)
print(' remove - Removes a yaml key, if it exists. Requires KEY arg.', file=sys.stderr)
print(' replace - Replaces (or adds) a new key and set its value. Requires KEY and VALUE args.', file=sys.stderr)
print(' help - Prints this usage information.', file=sys.stderr)
print(' append - Append a list item to a yaml key, if it exists and is a list. Requires KEY and LISTITEM args.', file=sys.stderr)
print(' appendlistobject - Append an object to a yaml list key. Requires KEY and JSON_OBJECT args.', file=sys.stderr)
print(' removelistitem - Remove a list item from a yaml key, if it exists and is a list. Requires KEY and LISTITEM args.', file=sys.stderr)
print(' replacelistobject - Replace a list object based on a condition. Requires KEY, CONDITION_FIELD, CONDITION_VALUE, and JSON_OBJECT args.', file=sys.stderr)
print(' add - Add a new key and set its value. Fails if key already exists. Requires KEY and VALUE args.', file=sys.stderr)
print(' get - Displays (to stdout) the value stored in the given key. Requires KEY arg.', file=sys.stderr)
print(' remove - Removes a yaml key, if it exists. Requires KEY arg.', file=sys.stderr)
print(' replace - Replaces (or adds) a new key and set its value. Requires KEY and VALUE args.', file=sys.stderr)
print(' help - Prints this usage information.', file=sys.stderr)
print('', file=sys.stderr)
print(' Where:', file=sys.stderr)
print(' YAML_FILE - Path to the file that will be modified. Ex: /opt/so/conf/service/conf.yaml', file=sys.stderr)
print(' KEY - YAML key, does not support \' or " characters at this time. Ex: level1.level2', file=sys.stderr)
print(' VALUE - Value to set for a given key. Can be a literal value or file:<path> to load from a YAML file.', file=sys.stderr)
print(' LISTITEM - Item to append to a given key\'s list value. Can be a literal value or file:<path> to load from a YAML file.', file=sys.stderr)
print(' YAML_FILE - Path to the file that will be modified. Ex: /opt/so/conf/service/conf.yaml', file=sys.stderr)
print(' KEY - YAML key, does not support \' or " characters at this time. Ex: level1.level2', file=sys.stderr)
print(' VALUE - Value to set for a given key. Can be a literal value or file:<path> to load from a YAML file.', file=sys.stderr)
print(' LISTITEM - Item to append to a given key\'s list value. Can be a literal value or file:<path> to load from a YAML file.', file=sys.stderr)
print(' JSON_OBJECT - JSON string representing an object to append to a list.', file=sys.stderr)
print(' CONDITION_FIELD - Field name to match in list items (e.g., "name").', file=sys.stderr)
print(' CONDITION_VALUE - Value to match for the condition field.', file=sys.stderr)
sys.exit(1)
@@ -122,6 +128,52 @@ def append(args):
return 0
def appendListObjectItem(content, key, listObject):
pieces = key.split(".", 1)
if len(pieces) > 1:
appendListObjectItem(content[pieces[0]], pieces[1], listObject)
else:
try:
if not isinstance(content[key], list):
raise AttributeError("Value is not a list")
content[key].append(listObject)
except AttributeError:
print("The existing value for the given key is not a list. No action was taken on the file.", file=sys.stderr)
return 1
except KeyError:
print("The key provided does not exist. No action was taken on the file.", file=sys.stderr)
return 1
def appendlistobject(args):
if len(args) != 3:
print('Missing filename, key arg, or JSON object to append', file=sys.stderr)
showUsage(None)
return 1
filename = args[0]
key = args[1]
jsonString = args[2]
try:
# Parse the JSON string into a Python dictionary
listObject = json.loads(jsonString)
except json.JSONDecodeError as e:
print(f'Invalid JSON string: {e}', file=sys.stderr)
return 1
# Verify that the parsed content is a dictionary (object)
if not isinstance(listObject, dict):
print('The JSON string must represent an object (dictionary), not an array or primitive value.', file=sys.stderr)
return 1
content = loadYaml(filename)
appendListObjectItem(content, key, listObject)
writeYaml(filename, content)
return 0
def removelistitem(args):
if len(args) != 3:
print('Missing filename, key arg, or list item to remove', file=sys.stderr)
@@ -139,6 +191,68 @@ def removelistitem(args):
return 0
def replaceListObjectByCondition(content, key, conditionField, conditionValue, newObject):
pieces = key.split(".", 1)
if len(pieces) > 1:
replaceListObjectByCondition(content[pieces[0]], pieces[1], conditionField, conditionValue, newObject)
else:
try:
if not isinstance(content[key], list):
raise AttributeError("Value is not a list")
# Find and replace the item that matches the condition
found = False
for i, item in enumerate(content[key]):
if isinstance(item, dict) and item.get(conditionField) == conditionValue:
content[key][i] = newObject
found = True
break
if not found:
print(f"No list item found with {conditionField}={conditionValue}. No action was taken on the file.", file=sys.stderr)
return 1
except AttributeError:
print("The existing value for the given key is not a list. No action was taken on the file.", file=sys.stderr)
return 1
except KeyError:
print("The key provided does not exist. No action was taken on the file.", file=sys.stderr)
return 1
def replacelistobject(args):
if len(args) != 5:
print('Missing filename, key arg, condition field, condition value, or JSON object', file=sys.stderr)
showUsage(None)
return 1
filename = args[0]
key = args[1]
conditionField = args[2]
conditionValue = args[3]
jsonString = args[4]
try:
# Parse the JSON string into a Python dictionary
newObject = json.loads(jsonString)
except json.JSONDecodeError as e:
print(f'Invalid JSON string: {e}', file=sys.stderr)
return 1
# Verify that the parsed content is a dictionary (object)
if not isinstance(newObject, dict):
print('The JSON string must represent an object (dictionary), not an array or primitive value.', file=sys.stderr)
return 1
content = loadYaml(filename)
result = replaceListObjectByCondition(content, key, conditionField, conditionValue, newObject)
if result != 1:
writeYaml(filename, content)
return result if result is not None else 0
def addKey(content, key, value):
pieces = key.split(".", 1)
if len(pieces) > 1:
@@ -247,7 +361,9 @@ def main():
"help": showUsage,
"add": add,
"append": append,
"appendlistobject": appendlistobject,
"removelistitem": removelistitem,
"replacelistobject": replacelistobject,
"get": get,
"remove": remove,
"replace": replace,

View File

@@ -580,3 +580,340 @@ class TestRemoveListItem(unittest.TestCase):
soyaml.main()
sysmock.assert_called()
self.assertEqual("The existing value for the given key is not a list. No action was taken on the file.\n", mock_stderr.getvalue())
class TestAppendListObject(unittest.TestCase):
def test_appendlistobject_missing_arg(self):
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "help"]
soyaml.appendlistobject(["file", "key"])
sysmock.assert_called()
self.assertIn("Missing filename, key arg, or JSON object to append", mock_stderr.getvalue())
def test_appendlistobject(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: 123 }, key2: [{name: item1, value: 10}]}")
file.close()
json_obj = '{"name": "item2", "value": 20}'
soyaml.appendlistobject([filename, "key2", json_obj])
file = open(filename, "r")
actual = file.read()
file.close()
expected = "key1:\n child1: 123\nkey2:\n- name: item1\n value: 10\n- name: item2\n value: 20\n"
self.assertEqual(actual, expected)
def test_appendlistobject_nested(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: [{name: a, id: 1}], child2: abc }, key2: false}")
file.close()
json_obj = '{"name": "b", "id": 2}'
soyaml.appendlistobject([filename, "key1.child1", json_obj])
file = open(filename, "r")
actual = file.read()
file.close()
# YAML doesn't guarantee key order in dictionaries, so check for content
self.assertIn("child1:", actual)
self.assertIn("name: a", actual)
self.assertIn("id: 1", actual)
self.assertIn("name: b", actual)
self.assertIn("id: 2", actual)
self.assertIn("child2: abc", actual)
self.assertIn("key2: false", actual)
def test_appendlistobject_nested_deep(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: 123, child2: { deep1: 45, deep2: [{x: 1}] } }, key2: false}")
file.close()
json_obj = '{"x": 2, "y": 3}'
soyaml.appendlistobject([filename, "key1.child2.deep2", json_obj])
file = open(filename, "r")
actual = file.read()
file.close()
expected = "key1:\n child1: 123\n child2:\n deep1: 45\n deep2:\n - x: 1\n - x: 2\n y: 3\nkey2: false\n"
self.assertEqual(actual, expected)
def test_appendlistobject_invalid_json(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1}]}")
file.close()
with patch('sys.stderr', new=StringIO()) as mock_stderr:
result = soyaml.appendlistobject([filename, "key1", "{invalid json"])
self.assertEqual(result, 1)
self.assertIn("Invalid JSON string:", mock_stderr.getvalue())
def test_appendlistobject_not_dict(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1}]}")
file.close()
with patch('sys.stderr', new=StringIO()) as mock_stderr:
# Try to append an array instead of an object
result = soyaml.appendlistobject([filename, "key1", "[1, 2, 3]"])
self.assertEqual(result, 1)
self.assertIn("The JSON string must represent an object (dictionary)", mock_stderr.getvalue())
def test_appendlistobject_not_dict_primitive(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1}]}")
file.close()
with patch('sys.stderr', new=StringIO()) as mock_stderr:
# Try to append a primitive value
result = soyaml.appendlistobject([filename, "key1", "123"])
self.assertEqual(result, 1)
self.assertIn("The JSON string must represent an object (dictionary)", mock_stderr.getvalue())
def test_appendlistobject_key_noexist(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1}]}")
file.close()
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "appendlistobject", filename, "key2", '{"name": "item2"}']
soyaml.main()
sysmock.assert_called()
self.assertEqual("The key provided does not exist. No action was taken on the file.\n", mock_stderr.getvalue())
def test_appendlistobject_key_noexist_deep(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: [{name: a}] }}")
file.close()
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "appendlistobject", filename, "key1.child2", '{"name": "b"}']
soyaml.main()
sysmock.assert_called()
self.assertEqual("The key provided does not exist. No action was taken on the file.\n", mock_stderr.getvalue())
def test_appendlistobject_key_nonlist(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: 123 }}")
file.close()
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "appendlistobject", filename, "key1", '{"name": "item"}']
soyaml.main()
sysmock.assert_called()
self.assertEqual("The existing value for the given key is not a list. No action was taken on the file.\n", mock_stderr.getvalue())
def test_appendlistobject_key_nonlist_deep(self):
filename = "/tmp/so-yaml_test-appendlistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: 123, child2: { deep1: 45 } }}")
file.close()
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "appendlistobject", filename, "key1.child2.deep1", '{"name": "item"}']
soyaml.main()
sysmock.assert_called()
self.assertEqual("The existing value for the given key is not a list. No action was taken on the file.\n", mock_stderr.getvalue())
class TestReplaceListObject(unittest.TestCase):
def test_replacelistobject_missing_arg(self):
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "help"]
soyaml.replacelistobject(["file", "key", "field"])
sysmock.assert_called()
self.assertIn("Missing filename, key arg, condition field, condition value, or JSON object", mock_stderr.getvalue())
def test_replacelistobject(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1, value: 10}, {name: item2, value: 20}]}")
file.close()
json_obj = '{"name": "item2", "value": 25, "extra": "field"}'
soyaml.replacelistobject([filename, "key1", "name", "item2", json_obj])
file = open(filename, "r")
actual = file.read()
file.close()
expected = "key1:\n- name: item1\n value: 10\n- extra: field\n name: item2\n value: 25\n"
self.assertEqual(actual, expected)
def test_replacelistobject_nested(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: [{id: '1', status: active}, {id: '2', status: inactive}] }}")
file.close()
json_obj = '{"id": "2", "status": "active", "updated": true}'
soyaml.replacelistobject([filename, "key1.child1", "id", "2", json_obj])
file = open(filename, "r")
actual = file.read()
file.close()
expected = "key1:\n child1:\n - id: '1'\n status: active\n - id: '2'\n status: active\n updated: true\n"
self.assertEqual(actual, expected)
def test_replacelistobject_nested_deep(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: 123, child2: { deep1: 45, deep2: [{name: a, val: 1}, {name: b, val: 2}] } }}")
file.close()
json_obj = '{"name": "b", "val": 99}'
soyaml.replacelistobject([filename, "key1.child2.deep2", "name", "b", json_obj])
file = open(filename, "r")
actual = file.read()
file.close()
expected = "key1:\n child1: 123\n child2:\n deep1: 45\n deep2:\n - name: a\n val: 1\n - name: b\n val: 99\n"
self.assertEqual(actual, expected)
def test_replacelistobject_invalid_json(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1}]}")
file.close()
with patch('sys.stderr', new=StringIO()) as mock_stderr:
result = soyaml.replacelistobject([filename, "key1", "name", "item1", "{invalid json"])
self.assertEqual(result, 1)
self.assertIn("Invalid JSON string:", mock_stderr.getvalue())
def test_replacelistobject_not_dict(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1}]}")
file.close()
with patch('sys.stderr', new=StringIO()) as mock_stderr:
result = soyaml.replacelistobject([filename, "key1", "name", "item1", "[1, 2, 3]"])
self.assertEqual(result, 1)
self.assertIn("The JSON string must represent an object (dictionary)", mock_stderr.getvalue())
def test_replacelistobject_condition_not_found(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1, value: 10}, {name: item2, value: 20}]}")
file.close()
with patch('sys.stderr', new=StringIO()) as mock_stderr:
json_obj = '{"name": "item3", "value": 30}'
result = soyaml.replacelistobject([filename, "key1", "name", "item3", json_obj])
self.assertEqual(result, 1)
self.assertIn("No list item found with name=item3", mock_stderr.getvalue())
# Verify file was not modified
file = open(filename, "r")
actual = file.read()
file.close()
self.assertIn("item1", actual)
self.assertIn("item2", actual)
self.assertNotIn("item3", actual)
def test_replacelistobject_key_noexist(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1}]}")
file.close()
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "replacelistobject", filename, "key2", "name", "item1", '{"name": "item2"}']
soyaml.main()
sysmock.assert_called()
self.assertEqual("The key provided does not exist. No action was taken on the file.\n", mock_stderr.getvalue())
def test_replacelistobject_key_noexist_deep(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: [{name: a}] }}")
file.close()
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "replacelistobject", filename, "key1.child2", "name", "a", '{"name": "b"}']
soyaml.main()
sysmock.assert_called()
self.assertEqual("The key provided does not exist. No action was taken on the file.\n", mock_stderr.getvalue())
def test_replacelistobject_key_nonlist(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: 123 }}")
file.close()
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "replacelistobject", filename, "key1", "name", "item", '{"name": "item"}']
soyaml.main()
sysmock.assert_called()
self.assertEqual("The existing value for the given key is not a list. No action was taken on the file.\n", mock_stderr.getvalue())
def test_replacelistobject_key_nonlist_deep(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: { child1: 123, child2: { deep1: 45 } }}")
file.close()
with patch('sys.exit', new=MagicMock()) as sysmock:
with patch('sys.stderr', new=StringIO()) as mock_stderr:
sys.argv = ["cmd", "replacelistobject", filename, "key1.child2.deep1", "name", "item", '{"name": "item"}']
soyaml.main()
sysmock.assert_called()
self.assertEqual("The existing value for the given key is not a list. No action was taken on the file.\n", mock_stderr.getvalue())
def test_replacelistobject_string_condition_value(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{name: item1, value: 10}, {name: item2, value: 20}]}")
file.close()
json_obj = '{"name": "item1", "value": 15}'
soyaml.replacelistobject([filename, "key1", "name", "item1", json_obj])
file = open(filename, "r")
actual = file.read()
file.close()
expected = "key1:\n- name: item1\n value: 15\n- name: item2\n value: 20\n"
self.assertEqual(actual, expected)
def test_replacelistobject_numeric_condition_value(self):
filename = "/tmp/so-yaml_test-replacelistobject.yaml"
file = open(filename, "w")
file.write("{key1: [{id: '1', status: active}, {id: '2', status: inactive}]}")
file.close()
json_obj = '{"id": "1", "status": "updated"}'
soyaml.replacelistobject([filename, "key1", "id", "1", json_obj])
file = open(filename, "r")
actual = file.read()
file.close()
expected = "key1:\n- id: '1'\n status: updated\n- id: '2'\n status: inactive\n"
self.assertEqual(actual, expected)