Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 51 additions & 5 deletions adminapi/cli.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Serveradmin - adminapi - Command Line Interface

Copyright (c) 2025 InnoGames GmbH
Copyright (c) 2026 InnoGames GmbH
"""

import json
Expand Down Expand Up @@ -57,11 +57,34 @@ def parse_args(args):
return parser.parse_args(args)


def _resolve_dot_notations(attribute_ids):
"""Resolve dotted attribute notations, merging shared prefixes.

["hostname"] -> ["hostname"]
["vms.hostname"] -> [{"vms": ["hostname"]}]
["loadbalancer.vms", "loadbalancer.route_network.hostname"]
-> [{"loadbalancer": ["vms", {"route_network": ["hostname"]}]}]
"""
result = []
joins = {}
for attr in attribute_ids:
parts = attr.split('.')
if len(parts) == 1:
result.append(parts[0])
else:
joins.setdefault(parts[0], []).append('.'.join(parts[1:]))

for key, suffixes in joins.items():
result.append({key: _resolve_dot_notations(suffixes)})

return result


def main():
args = parse_args(sys.argv[1:])

attribute_ids_to_print = args.attr if args.attr else ['hostname']
attribute_ids_to_fetch = list(attribute_ids_to_print)
attribute_ids_to_fetch = _resolve_dot_notations(attribute_ids_to_print)
if args.reset:
attribute_ids_to_fetch.extend(args.reset)
if args.update:
Expand Down Expand Up @@ -109,19 +132,42 @@ def apply_updates(server, attribute_values):
server.set(attribute_id, value)


def _resolve_value(obj, attribute_id):
"""Resolve a possibly dot-notated attribute to its value."""
first, _, rest = attribute_id.partition('.')
value = obj[first]
if not rest:
return value
if isinstance(value, MultiAttr):
collected = []
for related in value:
v = _resolve_value(related, rest)
if isinstance(v, MultiAttr):
collected.extend(v)
else:
collected.append(v)
return MultiAttr(collected, obj, first)
return _resolve_value(value, rest)


def print_server(query: Query, attribute_ids: list[str], output_format: str):
if output_format == 'json':
values = [{key: value for key, value in server.items() if key in attribute_ids} for server in query]
top_level_keys = {attr.split('.')[0] for attr in attribute_ids}
values = [
{k: v for k, v in server.items() if k in top_level_keys}
for server in query
]
print(json.dumps(values, indent=2, cls=ServeradminJSONEncoder))
else:
for server in query:
values = []
for attribute_id in attribute_ids:
if attribute_id not in server:
try:
value = _resolve_value(server, attribute_id)
except KeyError:
values.append('{N/A}')
continue

value = server[attribute_id]
if any(value is v for v in (None, True, False)):
value = '{{{}}}'.format(str(value).lower())

Expand Down
25 changes: 23 additions & 2 deletions adminapi/tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,8 @@
from argparse import ArgumentParser
from typing import Text, NoReturn

from adminapi.cli import parse_args
from adminapi.cli import parse_args, _resolve_value
from adminapi.dataset import DatasetObject, MultiAttr


# ArgumentParser exists on error, we cannot capture this so we override it
Expand Down Expand Up @@ -65,4 +66,24 @@ def test_update_argument(self, *args):
self.assertEqual(args.update, [('hostname', 'SomeNewHostname')])

args = parse_args(['project=adminapi', '--update', 'hostname=SomeNewHostname', '-u', 'state=maintenance'])
self.assertEqual(args.update, [('hostname', 'SomeNewHostname'), ('state', 'maintenance')])
self.assertEqual(args.update, [('hostname', 'SomeNewHostname'), ('state', 'maintenance')])


class TestResolveValue(unittest.TestCase):
def test_resolve_dotted_multi_attr(self):
vm1 = DatasetObject({'hostname': 'vm-1'}, object_id=1)
vm2 = DatasetObject({'hostname': 'vm-2'}, object_id=2)
server = DatasetObject({'hostname': 'hv-1', 'vms': [vm1, vm2]})

result = _resolve_value(server, 'vms.hostname')

self.assertIsInstance(result, MultiAttr)
self.assertEqual(sorted(result), ['vm-1', 'vm-2'])

def test_resolve_dotted_single_relation(self):
related = DatasetObject({'hostname': 'lb-1'}, object_id=1)
server = DatasetObject({'loadbalancer': related})

result = _resolve_value(server, 'loadbalancer.hostname')

self.assertEqual(result, 'lb-1')
Loading