Coverage for python/univention/config_registry/frontend.py: 99%
393 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-09-11 14:11 +0000
« prev ^ index » next coverage.py v7.6.0, created at 2026-09-11 14:11 +0000
1"""Univention Configuration Registry command line implementation."""
2# main configuration registry classes
3#
4# SPDX-FileCopyrightText: 2004-2026 Univention GmbH
5# SPDX-License-Identifier: AGPL-3.0-only
6#
7# API stability :pylint: disable-msg=W0613
8# Rewrite :pylint: disable-msg=R0912,R0914,R0915
10import os
11import re
12import sys
13import time
14from collections.abc import Callable, Iterator
15from typing import IO, Any, NoReturn
17import univention.config_registry_info as cri
18from univention.config_registry.backend import SCOPE, ConfigRegistry, exception_occured
19from univention.config_registry.filters import filter_keys_only, filter_shell, filter_sort
20from univention.config_registry.handler import ConfigHandlers, run_filter
21from univention.config_registry.misc import escape_value, validate_key
22from univention.config_registry.validation import Type
25__all__ = [
26 'REPLOG_FILE',
27 'UnknownKeyException',
28 'handler_commit',
29 'handler_dump',
30 'handler_filter',
31 'handler_get',
32 'handler_help',
33 'handler_info',
34 'handler_register',
35 'handler_search',
36 'handler_set',
37 'handler_unregister',
38 'handler_unset',
39 'handler_update',
40 'handler_version',
41 'main',
42]
44REPLOG_FILE = '/var/log/univention/config-registry.replog'
46_SHOW_EMPTY, _SHOW_DESCRIPTION, _SHOW_SCOPE, _SHOW_CATEGORIES, _SHOW_DEFAULT, _SHOW_TYPE = (1 << _ for _ in range(6))
49class UnknownKeyException(Exception):
50 """Query for unknown key: no info file nor set."""
52 def __init__(self, value):
53 Exception.__init__(self, value)
55 def __str__(self):
56 return 'W: Unknown key: "%s"' % self.args
59def replog(ucr: ConfigRegistry, var: str, old_value: str | None, value: str | None = None) -> None:
60 """
61 This function writes a new entry to replication logfile if
62 this feature has been enabled.
64 :param ucr: UCR instance.
65 :param var: UCR variable name.
66 :param old_value: Old UCR variable value.
67 :param value: New UCR variable value. `None` is now unset.
68 """
69 if ucr.is_true('ucr/replog/enabled', False):
70 if value is not None:
71 method = 'set'
72 varvalue = "%s=%s" % (var, escape_value(value))
73 else:
74 method = 'unset'
75 varvalue = "'%s'" % var
77 scope_arg = {
78 ConfigRegistry.LDAP: '--ldap-policy ',
79 ConfigRegistry.FORCED: '--force ',
80 ConfigRegistry.SCHEDULE: '--schedule ',
81 }.get(ucr.scope, '')
83 if old_value is None:
84 old_value = "[Previously undefined]"
86 log = '%s: %s %s%s old:%s\n' % (time.strftime("%Y-%m-%d %H:%M:%S"), method, scope_arg, varvalue, old_value)
87 try:
88 if not os.path.isfile(REPLOG_FILE): 88 ↛ 91line 88 didn't jump to line 91 because the condition on line 88 was always true
89 os.close(os.open(REPLOG_FILE, os.O_CREAT, 0o640))
91 with open(REPLOG_FILE, "a+", encoding='utf-8') as logfile:
92 logfile.write(log)
93 except OSError as ex:
94 print("E: exception occurred while writing to replication log: %s" % (ex,), file=sys.stderr)
95 exception_occured()
98def handler_set(args: list[str], opts: dict[str, Any] = {}, quiet: bool = False) -> None:
99 """
100 Set config registry variables in args.
101 Args is an array of strings 'key=value' or 'key?value'.
103 :param args: Command line arguments.
104 :param opts: Command line options.
105 :param quiet: Hide output.
106 """
107 ucr = ConfigRegistry()
108 ucr.load()
109 ignore_check = opts.get('ignore-check') or ucr.is_false('ucr/check/type')
111 ucr = _ucr_from_opts(opts)
112 info = _get_config_registry_info()
113 with ucr:
114 changes: dict[str, str | None] = {}
115 for arg in args:
116 sep_set = arg.find('=') # set
117 sep_def = arg.find('?') # set if not already set
118 if sep_set == -1 and sep_def == -1:
119 print("W: Missing value for config registry variable '%s'" % (arg,), file=sys.stderr)
120 continue
122 if sep_set > 0 and sep_def == -1:
123 sep = sep_set
124 elif sep_def > 0 and sep_set == -1:
125 sep = sep_def
126 else:
127 sep = min(sep_set, sep_def)
128 key = arg[0:sep]
129 value = arg[sep + 1:]
130 key_exists = key in ucr._layer
131 do_set_value = (not key_exists or sep == sep_set)
132 if do_set_value and validate_key(key):
133 if not quiet: 133 ↛ 138line 133 didn't jump to line 138 because the condition on line 133 was always true
134 if key_exists:
135 print('Setting %s' % key)
136 else:
137 print('Create %s' % key)
138 vinfo = info.get_variable(key) or info.match_pattern(key)
139 if vinfo: # Type checking can only be done if key is already configured
140 try:
141 validator = Type(vinfo)
142 except (TypeError, ValueError):
143 if ignore_check:
144 print('W: Invalid UCR type definition for type %r of %r, but set anyway' % (vinfo.get('type'), key), file=sys.stderr)
145 else:
146 print('E: Invalid UCR type definition for type %r of %r, value %r not set' % (vinfo.get('type'), key, value), file=sys.stderr)
147 opts['exit_code'] = 1
148 opts.setdefault('type_def_error', []).append((key, vinfo.get('type'), value))
149 continue # do not set value and continue with next element of for loop to be set
150 else:
151 if not validator.check(value):
152 if ignore_check:
153 print('W: Value %r incompatible for %r, but setting anyway' % (value, key), file=sys.stderr)
154 else:
155 print('E: Value %r incompatible for %r' % (value, key), file=sys.stderr)
156 opts['exit_code'] = 2
157 opts.setdefault('type_errors', []).append((key, value))
158 continue # do not set value and continue with next element of for loop to be set
159 changes[key] = value
160 else:
161 if do_set_value:
162 opts['exit_code'] = 1
163 if not quiet: 163 ↛ 115line 163 didn't jump to line 115 because the condition on line 163 was always true
164 if key_exists:
165 print('Not updating %s' % key)
166 else:
167 print('Not setting %s' % key)
168 changed = ucr.update(changes)
170 _run_changed(ucr, changed, "" if quiet else 'W: %s is overridden by scope "%s"')
173def handler_unset(args: list[str], opts: dict[str, Any] = {}) -> None:
174 """
175 Unset config registry variables in args.
177 :param args: Command line arguments.
178 :param opts: Command line options.
179 """
180 ucr = _ucr_from_opts(opts)
181 with ucr:
182 changes: dict[str, str | None] = {}
183 for arg in args:
184 if arg in ucr._layer:
185 print('Unsetting %s' % arg)
186 changes[arg] = None
187 else:
188 msg = "W: The config registry variable '%s' does not exist"
189 print(msg % (arg,), file=sys.stderr)
190 changed = ucr.update(changes)
192 _run_changed(ucr, changed, 'W: %s is still set in scope "%s"')
195def ucr_update(ucr: ConfigRegistry, changes: dict[str, str | None]) -> None:
196 """
197 Set or unset the given config registry variables.
199 :param ucr: UCR instance.
200 :param changes: Changed UCR variables.
201 """
202 with ucr:
203 changed = ucr.update(changes)
204 _run_changed(ucr, changed)
207def _run_changed(ucr: ConfigRegistry, changed: dict[str, tuple[str | None, str | None]], msg: str = "") -> None:
208 """
209 Run handlers for changed UCR variables.
211 :param ucr: UCR instance.
212 :param changed: Mapping from UCR variable name to 2-tuple (old-value, new-value).
213 :param msg: Message to be printed when change is shadowed by higher layer. Must contain 2 `%s` placeholders for `key` and `scope-name`.
214 """
215 visible: dict[str, tuple[str | None, str | None]] = {}
216 for key, (old_value, new_value) in changed.items():
217 replog(ucr, key, old_value, new_value)
219 reg = ucr.scope
220 while old_value is None and reg > 0:
221 reg -= 1
222 old_value = ucr._registry[reg].get(key)
224 reg, new_value = ucr.get(key, (0, None), getscope=True)
225 if reg > ucr.scope:
226 if msg:
227 print(msg % (key, SCOPE[reg]), file=sys.stderr)
228 else:
229 visible[key] = (old_value, new_value)
231 handlers = ConfigHandlers()
232 handlers.load()
233 handlers(list(visible), (ucr, visible))
236def _ucr_from_opts(opts: dict[str, Any]) -> ConfigRegistry:
237 """
238 Create :py:class:`ConfigRegistry` instance according to requested layer.
240 :param opts: Command line options.
241 :returns: A new UCR instance.
242 """
243 if opts.get('ldap-policy'):
244 scope = ConfigRegistry.LDAP
245 elif opts.get('force'):
246 scope = ConfigRegistry.FORCED
247 elif opts.get('schedule'):
248 scope = ConfigRegistry.SCHEDULE
249 else:
250 scope = ConfigRegistry.NORMAL
251 ucr = ConfigRegistry(write_registry=scope)
252 return ucr
255def handler_dump(args: list[str], opts: dict[str, Any] = {}) -> Iterator[str]:
256 """
257 Dump all variables.
259 :param args: Command line arguments.
260 :param opts: Command line options.
261 """
262 ucr = ConfigRegistry()
263 ucr.load()
264 yield from str(ucr).split('\n')
267def handler_update(args: list[str], opts: dict[str, Any] = {}) -> None:
268 """
269 Update handlers.
271 :param args: Command line arguments.
272 :param opts: Command line options.
273 """
274 handlers = ConfigHandlers()
275 cur = handlers.update()
276 handlers.update_divert(cur)
278 ucr = ConfigRegistry()
279 _register_variable_default_values(ucr)
282def handler_commit(args: list[str], opts: dict[str, Any] = {}) -> None:
283 """
284 Commit all registered templated files.
286 :param args: Command line arguments.
287 :param opts: Command line options.
288 """
289 ucr = ConfigRegistry()
290 ucr.load()
292 handlers = ConfigHandlers()
293 handlers.load()
294 handlers.commit(ucr, args)
297def handler_register(args: list[str], opts: dict[str, Any] = {}) -> None:
298 """
299 Register new `.info` file.
301 :param args: Command line arguments.
302 :param opts: Command line options.
303 """
304 ucr = ConfigRegistry()
305 ucr.load()
307 handlers = ConfigHandlers()
308 handlers.update() # cache must be current
309 # Bug #21263: by forcing an update here, the new .info file is already
310 # incorporated. Calling register for multifiles will increment the
311 # def_count a second time, which is not nice, but uncritical, since the
312 # diversion is (re-)done when >= 1.
314 _register_variable_default_values(ucr)
315 handlers.register(args[0], ucr)
316 # handlers.commit((ucr, {}))
319def handler_unregister(args: list[str], opts: dict[str, Any] = {}) -> None:
320 """
321 Unregister old `.info` file.
323 :param args: Command line arguments.
324 :param opts: Command line options.
325 """
326 ucr = ConfigRegistry()
327 ucr.load()
329 handlers = ConfigHandlers()
330 cur = handlers.update() # cache must be current
331 obsolete = handlers.unregister(args[0], ucr)
332 handlers.update_divert(cur - obsolete)
333 _register_variable_default_values(ucr)
336def handler_filter(args: list[str], opts: dict[str, Any] = {}) -> None:
337 """Run filter on STDIN to STDOUT."""
338 ucr = ConfigRegistry()
339 ucr.load()
340 sys.stdout.buffer.write(run_filter(sys.stdin.read(), ucr, opts=opts))
343def handler_search(args: list[str], opts: dict[str, Any] = {}) -> Iterator[str]:
344 """
345 Search for registry variable.
347 :param args: Command line arguments.
348 :param opts: Command line options.
349 """
350 search_keys = opts.get('key', False)
351 search_values = opts.get('value', False)
352 search_all = opts.get('all', False)
353 count_search = int(search_keys) + int(search_values) + int(search_all)
354 if count_search > 1:
355 print('E: at most one out of [--key|--value|--all] may be set', file=sys.stderr)
356 sys.exit(1)
357 elif count_search == 0:
358 search_keys = True
359 search_values |= search_all
360 search_keys |= search_all
362 if args:
363 try:
364 search = re.compile('|'.join('(?:%s)' % (_,) for _ in args)).search
365 except re.error as ex:
366 print('E: invalid regular expression: %s' % (ex,), file=sys.stderr)
367 sys.exit(1)
368 else:
369 def search(x): # type: ignore[misc]
370 return True
372 info = _get_config_registry_info()
374 category = opts.get('category')
375 if category and not info.get_category(category):
376 print('E: unknown category: "%s"' % (category,), file=sys.stderr)
377 sys.exit(1)
379 ucr = ConfigRegistry()
380 ucr.load()
382 details = _SHOW_EMPTY | _SHOW_DESCRIPTION
383 if opts.get('non-empty'):
384 details &= ~_SHOW_EMPTY
385 if opts.get('brief') or ucr.is_true('ucr/output/brief', False):
386 details &= ~_SHOW_DESCRIPTION
387 if ucr.is_true('ucr/output/scope', False):
388 details |= _SHOW_SCOPE
389 if opts.get('verbose'):
390 details |= _SHOW_CATEGORIES | _SHOW_DESCRIPTION
392 all_vars: dict[str, tuple[str | None, cri.Variable | None, int | None]] = {} # key: (value, vinfo, scope)
393 for key, var in info.get_variables(category).items():
394 all_vars[key] = (None, var, None)
395 for key, (scope, value) in ucr.items(getscope=True):
396 try:
397 all_vars[key] = (value, all_vars[key][1], scope)
398 except LookupError:
399 all_vars[key] = (value, None, scope)
401 for key, (value2, vinfo, _scope2) in all_vars.items():
402 if any((
403 search_keys and search(key),
404 search_values and value2 and search(value2),
405 search_all and vinfo and search(vinfo.get('description', '')),
406 )):
407 yield variable_info_string(key, value2, vinfo, details=details)
409 if _SHOW_EMPTY & details and not OPT_FILTERS['shell'][2]:
410 patterns: dict = {}
411 for arg in args or ('',):
412 patterns.update(info.describe_search_term(arg))
413 for pattern, vinfo in patterns.items():
414 yield variable_info_string(pattern, None, vinfo, details=details)
417def handler_get(args: list[str], opts: dict[str, Any] = {}) -> Iterator[str]:
418 """
419 Return config registry variable.
421 :param args: Command line arguments.
422 :param opts: Command line options.
423 """
424 ucr = ConfigRegistry()
425 ucr.load()
426 key = args[0]
427 value = ucr.get(key)
428 if value is None:
429 return
430 elif OPT_FILTERS['shell'][2]:
431 yield '%s: %s' % (key, value)
432 else:
433 yield value
436def variable_info_string(key: str, value: str | None, variable_info: cri.Variable | None, scope: int | None = None, details: int = _SHOW_DESCRIPTION) -> str:
437 """
438 Format UCR variable key, value, description, scope, categories and default value.
440 :param key: UCR variable name.
441 :param value: UCR variable value.
442 :param variable_info: Description object.
443 :param scope: UCS layer.
444 :param details: bit-field for detail-level.
445 :returns: formatted string
446 """
447 if value is None and not variable_info:
448 raise UnknownKeyException(key)
449 elif value in (None, '') and not _SHOW_EMPTY & details:
450 return ''
451 elif value is None:
452 # if not shell filter option is set
453 value_string = "<empty>" if not OPT_FILTERS["shell"][2] else ""
454 else:
455 value_string = '%s' % value
457 if scope is None or not 0 <= scope < len(SCOPE) or not _SHOW_SCOPE & details or OPT_FILTERS['shell'][2]: # Do not display scope in shell export
458 key_value = '%s: %s' % (key, value_string)
459 else:
460 key_value = '%s (%s): %s' % (key, SCOPE[scope], value_string)
462 info = [key_value]
463 if variable_info and _SHOW_DESCRIPTION & details:
464 # info.append(' ' + variable_info.get('description',
465 # 'no description available'))
466 # <https://forge.univention.org/bugzilla/show_bug.cgi?id=15556>
467 # Workaround:
468 description = variable_info.get('description')
469 if not description or not description.strip():
470 description = 'no description available'
471 info.append(' ' + description)
473 if variable_info and _SHOW_CATEGORIES & details:
474 info.append(' Categories: ' + variable_info.get('categories', 'none'))
476 if variable_info and _SHOW_DEFAULT & details:
477 info.append(' Default: ' + variable_info.get('default', '(not set)'))
479 if variable_info and _SHOW_TYPE & details:
480 try:
481 validator = Type(variable_info)
482 info.append(' Type: %s' % (validator,))
483 except ValueError as exc:
484 info.append(' Type: %s' % exc)
486 if (_SHOW_CATEGORIES | _SHOW_DESCRIPTION) & details:
487 info.append('')
489 return '\n'.join(info)
492def handler_info(args: list[str], opts: dict[str, Any] = {}) -> Iterator[str]:
493 """
494 Print variable info.
496 :param args: Command line arguments.
497 :param opts: Command line options.
498 """
499 ucr = ConfigRegistry()
500 ucr.load()
501 info = _get_config_registry_info()
503 for arg in args:
504 try:
505 yield variable_info_string(
506 arg, ucr.get(arg, None),
507 info.get_variable(arg),
508 details=_SHOW_EMPTY | _SHOW_DESCRIPTION | _SHOW_CATEGORIES | _SHOW_DEFAULT | _SHOW_TYPE)
509 except UnknownKeyException as ex:
510 print(ex, file=sys.stderr)
513def handler_version(args: list[str], opts: dict[str, Any] = {}) -> NoReturn:
514 """
515 Print version info.
517 :param args: Command line arguments.
518 :param opts: Command line options.
519 """
520 print('univention-config-registry @%@package_version@%@')
521 sys.exit(0)
524def handler_help(args: list[str], opts: dict[str, Any] = {}, out: IO = sys.stdout) -> None:
525 """
526 Print config registry command line usage.
528 :param args: Command line arguments.
529 :param opts: Command line options.
530 """
531 print('''
532univention-config-registry: base configuration for UCS
533copyright (c) 2001-2026 Univention GmbH, Germany
535Syntax:
536 univention-config-registry [options] <action> [options] [parameters]
538Options:
540 -h | --help | -?:
541 print this usage message and exit program
543 --version | -v:
544 print version information and exit program
546 --shell (valid actions: dump, search):
547 convert key/value pair into shell compatible format, e.g.
548 `version/version: 1.0` => `version_version="1.0"`
550 --keys-only (valid actions: dump, search):
551 print only the keys
553Actions:
554 set [--force|--schedule|--ldap-policy] [--ignore-check] <key>=<value> [... <key>=<value>]:
555 set one or more keys to specified values; if a key is non-existent
556 in the configuration registry it will be created
557 --ignore-check: ignore check if given value is compatible with type of the key
559 get <key>:
560 retrieve the value of the specified key from the configuration
561 database
563 unset [--force|--schedule|--ldap-policy] <key> [... <key>]:
564 remove one or more keys (and its associated values) from
565 configuration database
567 dump:
568 display all key/value pairs which are stored in the
569 configuration database
571 search [--key|--value|--all] [--category <category>] [--brief|-verbose] \\
572 [--non-empty] [... <regex>]:
573 displays all key/value pairs and their descriptions that match at
574 least one of the given regular expressions
575 --key: only search the keys (default)
576 --value: only search the values
577 --all: search keys, values and descriptions
578 --category: limit search to variables of <category>
579 --brief: don't print descriptions (default controlled via ucr/output/brief)
580 --verbose: also print category for each variable
581 --non-empty: only search in non-empty variables
582 no <regex> given: display all variables
584 info <key> [... <key>]:
585 display verbose information for the specified variable(s)
587 shell [key]:
588 convert key/value pair into shell compatible format, e.g.
589 `version/version: 1.0` => `version_version="1.0"`
590 (deprecated: use --shell dump instead)
592 commit [file1 ...]:
593 rebuild configuration file from univention template; if
594 no file is specified ALL configuration files are rebuilt
596 filter [file]:
597 evaluate a template file, expects Python inline code in UTF-8 or US-ASCII
599Description:
600 univention-config-registry is a tool to handle the basic configuration for
601 Univention Corporate Server (UCS)
602''', file=out)
603 sys.exit(0)
606def missing_parameter(action: str) -> NoReturn:
607 """Print missing parameter error."""
608 print('E: too few arguments for command [%s]' % (action,), file=sys.stderr)
609 print('try `univention-config-registry --help` for more information', file=sys.stderr)
610 sys.exit(1)
613def _get_config_registry_info() -> cri.ConfigRegistryInfo:
614 cri.set_language('en')
615 return cri.ConfigRegistryInfo(install_mode=False)
618def _register_variable_default_values(ucr: ConfigRegistry) -> None:
619 """Create base-default.conf layer containig all default values"""
620 info = _get_config_registry_info()
621 _ucr = ConfigRegistry(write_registry=ConfigRegistry.DEFAULTS)
622 _ucr.load()
623 defaults: dict[str, str | None] = {}
624 default_variables = info.get_variables()
625 for key, variable in default_variables.items():
626 value = variable.get('default')
627 if value: 627 ↛ 625line 627 didn't jump to line 625 because the condition on line 627 was always true
628 defaults[key] = value
629 for key in _ucr:
630 if key not in default_variables: 630 ↛ 629line 630 didn't jump to line 629 because the condition on line 630 was always true
631 defaults[key] = None
633 changed = {key: (old, new) for key, (old, new) in _ucr.update(defaults).items() if old != new}
634 _ucr.save()
635 ucr.load()
636 _run_changed(ucr, changed, 'I: %s will be set in scope "%s"')
639HANDLERS: dict[str, tuple[Callable[[list[str], dict[str, Any]], Iterator[str] | None], int]] = {
640 'set': (handler_set, 1),
641 'unset': (handler_unset, 1),
642 'dump': (handler_dump, 0),
643 'update': (handler_update, 0),
644 'commit': (handler_commit, 0),
645 'register': (handler_register, 1),
646 'unregister': (handler_unregister, 1),
647 'filter': (handler_filter, 0),
648 'search': (handler_search, 0),
649 'get': (handler_get, 1),
650 'info': (handler_info, 1),
651}
653# action options: each of these options perform an action
654OPT_ACTIONS: dict[str, list] = {
655 # name: [function, state, (alias list)]
656 'help': [handler_help, False, ('-h', '-?')],
657 'version': [handler_version, False, ('-v',)],
658 'debug': [lambda args: None, False, ()],
659}
661# filter options: these options define filter for the output
662OPT_FILTERS: dict[str, list] = {
663 # name: [prio, function, state, (valid actions)]
664 'keys-only': [0, filter_keys_only, False, ('dump', 'search')],
665 'sort': [10, filter_sort, False, ('dump', 'search', 'info')],
666 'shell': [99, filter_shell, False, ('dump', 'search', 'get')],
667}
669BOOL, STRING = range(2)
671OPT_COMMANDS: dict[str, dict[str, Any]] = {
672 'set': {
673 'force': [BOOL, False],
674 'ldap-policy': [BOOL, False],
675 'schedule': [BOOL, False],
676 'ignore-check': [BOOL, False],
677 },
678 'unset': {
679 'force': [BOOL, False],
680 'ldap-policy': [BOOL, False],
681 'schedule': [BOOL, False],
682 },
683 'search': {
684 'key': [BOOL, False],
685 'value': [BOOL, False],
686 'all': [BOOL, False],
687 'brief': [BOOL, False],
688 'category': [STRING, None],
689 'non-empty': [BOOL, False],
690 'verbose': [BOOL, False],
691 },
692 'filter': {
693 'encode-utf8': [BOOL, False],
694 'disallow-execution': [BOOL, False],
695 },
696}
699def main(args: list[str]) -> int:
700 """Run config registry."""
701 try:
702 # close your eyes ...
703 if not args:
704 args.append('--help')
705 # search for options in command line arguments
706 while args and args[0].startswith('-'):
707 arg = args.pop(0)
708 # is action option?
709 for key, opt in OPT_ACTIONS.items():
710 if arg[2:] == key or arg in opt[2]:
711 opt[1] = True
712 break
713 else:
714 # not an action option; is a filter option?
715 try:
716 OPT_FILTERS[arg[2:]][2] = True
717 except LookupError:
718 print('E: unknown option %s' % (arg,), file=sys.stderr)
719 sys.exit(1)
721 # is action already defined by global option?
722 for name, (func, state, _aliases) in OPT_ACTIONS.items():
723 if state:
724 func(args)
726 # find action
727 try:
728 action = args.pop(0)
729 except IndexError:
730 print('E: missing action, see --help', file=sys.stderr)
731 sys.exit(1)
732 # COMPAT: the 'shell' command is now an option and equivalent to
733 # --shell search
734 if action == 'shell':
735 action = 'search'
736 # activate shell option
737 OPT_FILTERS['shell'][2] = True
738 # switch to old, brief output
739 OPT_COMMANDS['search']['brief'][1] = True
741 tmp = []
742 if not args:
743 tmp.append('')
744 else:
745 for arg in args:
746 if not arg.startswith('--'):
747 tmp.append('^%s$' % arg)
748 else:
749 tmp.append(arg)
750 args = tmp
752 # set 'sort' option by default for dump and search
753 if action in ['dump', 'search', 'info']:
754 OPT_FILTERS['sort'][2] = True
756 # set brief option when generating shell output
757 if OPT_FILTERS['shell'][2]:
758 OPT_COMMANDS['search']['brief'][1] = True
760 # if a filter option is set: verify that a valid command is given
761 for name, (_prio, func, state, actions) in OPT_FILTERS.items():
762 if state and action not in actions:
763 print('E: invalid option --%s for command %s' % (name, action), file=sys.stderr)
764 sys.exit(1)
766 # check command options
767 cmd_opts = OPT_COMMANDS.get(action, {})
768 while args and args[0].startswith('--'):
769 arg = args.pop(0)
770 if action in ('set', 'unset') and arg == '--forced':
771 arg = '--force'
772 try:
773 cmd_opt_tuple = cmd_opts[arg[2:]]
774 except LookupError:
775 print('E: invalid option %s for command %s' % (arg, action), file=sys.stderr)
776 sys.exit(1)
777 else:
778 if cmd_opt_tuple[0] == BOOL:
779 cmd_opt_tuple[1] = True
780 else: # STRING
781 try:
782 cmd_opt_tuple[1] = args.pop(0)
783 except IndexError:
784 msg = 'E: option %s for command %s expects an argument'
785 print(msg % (arg, action), file=sys.stderr)
786 sys.exit(1)
788 # Drop type
789 cmd_opts = {key: value for key, (typ, value) in cmd_opts.items()}
791 # action!
792 try:
793 handler_func, min_args = HANDLERS[action]
794 except LookupError:
795 print('E: unknown action "%s", see --help' % (action,), file=sys.stderr)
796 sys.exit(1)
797 else:
798 # enough arguments?
799 if len(args) < min_args:
800 missing_parameter(action)
802 # if any filter option is set
803 cmd_opts['exit_code'] = 0
804 result = handler_func(args, cmd_opts)
805 if result is None:
806 return cmd_opts['exit_code']
808 results = result
809 # let the filter options do their job
810 for (_prio, filter_func, state, _actions) in sorted(OPT_FILTERS.values()):
811 if not state:
812 continue
813 results = filter_func(args, results)
815 for line in results:
816 print(line)
818 except (OSError, TypeError):
819 if OPT_ACTIONS['debug'][1]:
820 raise
821 exception_occured()
822 return cmd_opts['exit_code']