1 from collections import Counter
2 from collections.abc import Mapping
3 from functools import partial
4 from typing import Final
5
6 from pcs.cli.common.errors import SEE_MAN_CHANGES, CmdLineInputError
7 from pcs.common.str_tools import (
8 format_list,
9 format_list_custom_last_separator,
10 format_plural,
11 )
12 from pcs.common.tools import timeout_to_seconds
13 from pcs.common.types import StringCollection, StringIterable, StringSequence
14
15 # sys.argv always returns a list, we don't need StringSequence in here
16 Argv = list[str]
17 ModifierValueType = None | bool | str
18
19 _FUTURE_OPTION_STR: Final = "future"
20 FUTURE_OPTION: Final = f"--{_FUTURE_OPTION_STR}"
21 FUTURE_OPTION_DEPRECATION_WARNING: Final = (
22 "Option '--future' is deprecated and no longer has any effect, "
23 "since the behavior it enabled is now the default. It will be "
24 "removed in a future release."
25 )
26 _OUTPUT_FORMAT_OPTION_STR: Final = "output-format"
27 OUTPUT_FORMAT_OPTION: Final = f"--{_OUTPUT_FORMAT_OPTION_STR}"
28 OUTPUT_FORMAT_VALUE_CMD: Final = "cmd"
29 OUTPUT_FORMAT_VALUE_JSON: Final = "json"
30 OUTPUT_FORMAT_VALUE_TEXT: Final = "text"
31 OUTPUT_FORMAT_VALUES: Final = frozenset(
32 (
33 OUTPUT_FORMAT_VALUE_CMD,
34 OUTPUT_FORMAT_VALUE_JSON,
35 OUTPUT_FORMAT_VALUE_TEXT,
36 )
37 )
38
39 MODIFIER_OPTIONS_BOOL: Final[frozenset[str]] = frozenset(
40 (
41 "--all",
42 "--agent-validation",
43 # TODO remove
44 # used only in acl commands and it is deprecated there
45 "--autodelete",
46 "--brief",
47 "--config",
48 "--corosync",
49 "--debug",
50 "--defaults",
51 "--disabled",
52 "--enable",
53 "--expired",
54 "--force",
55 "--full",
56 "--quiet",
57 FUTURE_OPTION,
58 # TODO remove
59 # used only in deprecated 'pcs resource|stonith show'
60 "--groups",
61 "--hide-inactive",
62 "--local",
63 "--monitor",
64 "--no-default-ops",
65 "--nodesc",
66 "--no-expire-check",
67 "--no-cluster-uuid",
68 "--no-keys-sync",
69 "--no-stop",
70 "--no-strict",
71 "--no-watchdog-validation",
72 "--off",
73 "--overwrite",
74 "--pacemaker",
75 "--promoted",
76 "--safe",
77 "--show-secrets",
78 "--simulate",
79 "--skip-offline",
80 "--start",
81 "--strict",
82 "--yes",
83 )
84 )
85
86 MODIFIER_OPTIONS_VAL: Final[frozenset[str]] = frozenset(
87 (
88 "--after",
89 "--before",
90 "--booth-conf",
91 "--booth-key",
92 "--corosync_conf",
93 "--from",
94 # TODO remove
95 # used in resource create and stonith create, deprecated in both
96 "--group",
97 "--name",
98 "--node",
99 "--request-timeout",
100 "--to",
101 "--token",
102 "-f",
103 "-p",
104 "-u",
105 )
106 )
107
108 ARG_TYPE_DELIMITER: Final = "%"
109
110 # h = help, f = file,
111 # p = password (cluster auth), u = user (cluster auth),
112 PCS_SHORT_OPTIONS: Final = "hf:p:u:"
113 PCS_LONG_OPTIONS: Final = [
114 "debug",
115 "version",
116 "help",
117 "fullhelp",
118 "force",
119 "skip-offline",
120 "autodelete",
121 "simulate",
122 "all",
123 "full",
124 "local",
125 "wait",
126 "config",
127 "start",
128 "enable",
129 "disabled",
130 "off",
131 "request-timeout=",
132 "brief",
133 _FUTURE_OPTION_STR,
134 # resource (safe-)disable
135 "safe",
136 "no-strict",
137 # resource cleanup | refresh
138 "strict",
139 "pacemaker",
140 "corosync",
141 "no-default-ops",
142 "defaults",
143 "nodesc",
144 "promoted",
145 "name=",
146 "group=",
147 "node=",
148 "from=",
149 "to=",
150 "after=",
151 "before=",
152 "corosync_conf=",
153 "booth-conf=",
154 "booth-key=",
155 # do not stop resources in resource delete command
156 "no-stop",
157 "no-watchdog-validation",
158 # pcs cluster setup
159 "no-cluster-uuid",
160 "no-keys-sync",
161 # in pcs status - do not display resource status on inactive node
162 "hide-inactive",
163 # pcs resource (un)manage - enable or disable monitor operations
164 "monitor",
165 # TODO remove
166 # used only in deprecated 'pcs resource|stonith show'
167 "groups",
168 # "pcs resource clear --expired" - only clear expired moves and bans
169 "expired",
170 # disable evaluating whether rules are expired
171 "no-expire-check",
172 # allow overwriting existing files, currently meant for / used in CLI only
173 "overwrite",
174 # output format of commands, e.g: json, cmd, text, ...
175 f"{_OUTPUT_FORMAT_OPTION_STR}=",
176 # auth token
177 "token=",
178 # enable agent self validation
179 "agent-validation",
180 # disable text output in query commands
181 "quiet",
182 # proceed with dangerous actions, meant for / used in CLI only
183 "yes",
184 # retrieve and display cibsecret values / used in CLI only
185 "show-secrets",
186 ]
187
188
189 def split_list(arg_list: Argv, separator: str) -> list[Argv]:
190 """
191 split a list of arguments to several lists using separator as a delimiter
192
193 arg_list -- list of command line arguments to split
194 separator -- delimiter
195 """
196 separator_indexes = [i for i, x in enumerate(arg_list) if x == separator]
197 bounds = zip(
198 [0] + [i + 1 for i in separator_indexes],
199 separator_indexes + [None],
200 strict=False,
201 )
202 return [arg_list[i:j] for i, j in bounds]
203
204
205 def split_list_by_any_keywords(
206 arg_list: Argv, keyword_label: str
207 ) -> dict[str, Argv]:
208 """
209 split a list of arguments using any argument not containing = as a delimiter
210
211 arg_list -- list of command line arguments to split
212 keyword_label -- description of all keywords
213 """
214 groups: dict[str, Argv] = {}
215 if not arg_list:
216 return groups
217
218 if "=" in arg_list[0]:
219 raise CmdLineInputError(
220 f"Invalid character '=' in {keyword_label} '{arg_list[0]}'"
221 )
222
223 current_keyword = arg_list[0]
224 groups[current_keyword] = []
225 for arg in arg_list[1:]:
226 if "=" in arg:
227 groups[current_keyword].append(arg)
228 else:
229 current_keyword = arg
230 if current_keyword in groups:
231 raise CmdLineInputError(
232 "{} '{}' defined multiple times".format(
233 keyword_label.capitalize(), current_keyword
234 )
235 )
236 groups[current_keyword] = []
237 return groups
238
239
240 def split_option(arg: str, allow_empty_value: bool = True) -> tuple[str, str]:
241 """
242 Get (key, value) from a key=value commandline argument.
243
244 Split the argument by the first = and return resulting parts. Raise
245 CmdLineInputError if the argument cannot be split.
246
247 arg -- commandline argument to split
248 allow_empty_value -- if False, raise CmdLineInputError on empty value
249 """
250 if "=" not in arg:
251 raise CmdLineInputError(f"missing value of '{arg}' option")
252 if arg.startswith("="):
253 raise CmdLineInputError(f"missing key in '{arg}' option")
254 key, value = arg.split("=", 1)
255 if not (value or allow_empty_value):
256 raise CmdLineInputError(f"value of '{key}' option is empty")
257 return key, value
258
259
260 def ensure_unique_args(cmdline_args: Argv) -> None:
261 """
262 Raises in case there are duplicate args
263 """
264 duplicities = [
265 item for item, count in Counter(cmdline_args).items() if count > 1
266 ]
267 if duplicities:
268 argument_pl = format_plural(duplicities, "argument")
269 duplicities_list = format_list(duplicities)
270 raise CmdLineInputError(f"duplicate {argument_pl}: {duplicities_list}")
271
272
273 class KeyValueParser:
274 """
275 Parse and check key=value options
276 """
277
278 def __init__(self, arg_list: Argv, repeatable: StringCollection = ()):
279 """
280 arg_list -- commandline arguments to be parsed
281 repeatable -- keys that are allowed to be specified several times
282 """
283 self._repeatable_keys = repeatable
284 self._key_value_map: dict[str, list[str]] = {}
285 for arg in arg_list:
286 name, value = split_option(arg)
287 if name not in self._key_value_map:
288 self._key_value_map[name] = [value]
289 else:
290 self._key_value_map[name].append(value)
291
292 def check_allowed_keys(self, allowed_keys: StringCollection) -> None:
293 """
294 Check that only allowed keys were specified
295
296 allowed_keys -- list of allowed keys
297 """
298 unknown_options = set(self._key_value_map.keys()) - set(allowed_keys)
299 if unknown_options:
300 raise CmdLineInputError(
301 "Unknown option{s} '{options}'".format(
302 s=("s" if len(unknown_options) > 1 else ""),
303 options="', '".join(sorted(unknown_options)),
304 )
305 )
306
307 def get_unique(self) -> dict[str, str]:
308 """
309 Get all non-repeatable keys and their values; raise if a key has more values
310 """
311 result: dict[str, str] = {}
312 for key, values in self._key_value_map.items():
313 if key in self._repeatable_keys:
314 continue
315 values_uniq = set(values)
316 if len(values_uniq) > 1:
317 raise CmdLineInputError(
318 f"duplicate option '{key}' with different values "
319 f"{format_list_custom_last_separator(values_uniq, ' and ')}"
320 )
321 result[key] = values[0]
322 return result
323
324 def get_repeatable(self) -> dict[str, list[str]]:
325 """
326 Get all repeatable keys and their values
327 """
328 return {
329 key: self._key_value_map[key]
330 for key in self._repeatable_keys
331 if key in self._key_value_map
332 }
333
334
335 class ArgsByKeywords:
336 def __init__(self, groups: Mapping[str, list[Argv]]):
337 self._groups = groups
338 self._flat_cache: dict[str, Argv] = {}
339
340 def allow_repetition_only_for(self, keyword_set: StringCollection) -> None:
341 """
342 Raise CmdLineInputError if a keyword has been repeated when not allowed
343
344 keyword_set -- repetition is allowed for these keywords
345 """
346 for keyword, arg_groups in self._groups.items():
347 if len(arg_groups) > 1 and keyword not in keyword_set:
348 raise CmdLineInputError(
349 f"'{keyword}' cannot be used more than once"
350 )
351
352 def ensure_unique_keywords(self) -> None:
353 """
354 Raise CmdLineInputError if any keyword has been repeated
355 """
356 return self.allow_repetition_only_for(set())
357
358 def is_empty(self) -> bool:
359 """
360 Check if any args have been specified
361 """
362 return not self._groups
363
364 def has_keyword(self, keyword: str) -> bool:
365 """
366 Check if a keyword has been specified
367
368 keyword -- a keyword to check
369 """
370 return keyword in self._groups
371
372 def has_empty_keyword(self, keyword: str) -> bool:
373 """
374 Check if a keyword has been specified without any following args
375
376 keyword -- a keyword to check
377 """
378 return self.has_keyword(keyword) and not self.get_args_flat(keyword)
379
380 def get_args_flat(self, keyword: str) -> Argv:
381 """
382 Get arguments of a keyword in one sequence
383 """
384 if keyword in self._groups:
385 if keyword not in self._flat_cache:
386 self._flat_cache[keyword] = [
387 arg
388 for one_group in self._groups[keyword]
389 for arg in one_group
390 ]
391 return self._flat_cache[keyword]
392 return []
393
394 def get_args_groups(self, keyword: str) -> list[Argv]:
395 """
396 Get arguments of a keyword, one group for each keyword occurrence
397 """
398 if keyword in self._groups:
399 return self._groups[keyword]
400 return []
401
402
403 def group_by_keywords(
404 arg_list: Argv,
405 keyword_set: StringCollection,
406 implicit_first_keyword: str | None = None,
407 ) -> ArgsByKeywords:
408 """
409 Separate argv into groups delimited by specified keywords
410
411 arg_list -- commandline arguments containing keywords
412 keyword_set -- all expected keywords
413 implicit_first_keyword -- key for capturing args before the first keyword
414 """
415 args_by_keywords: dict[str, list[Argv]] = {}
416
417 def new_keyword(keyword: str) -> None:
418 if keyword not in args_by_keywords:
419 args_by_keywords[keyword] = []
420 args_by_keywords[keyword].append([])
421
422 if arg_list:
423 if arg_list[0] not in keyword_set:
424 if not implicit_first_keyword:
425 raise CmdLineInputError()
426 current_keyword = implicit_first_keyword
427 new_keyword(current_keyword)
428
429 for arg in arg_list:
430 if arg in keyword_set:
431 current_keyword = arg
432 new_keyword(current_keyword)
433 else:
434 args_by_keywords[current_keyword][-1].append(arg)
435
436 return ArgsByKeywords(args_by_keywords)
437
438
439 def parse_typed_arg(
440 arg: str, allowed_types: StringSequence, default_type: str
441 ) -> tuple[str, str]:
442 """
443 Get (type, value) from a typed commandline argument.
444
445 Split the argument by the type separator and return the type and the value.
446 Raise CmdLineInputError in the argument format or type is not valid.
447 string arg -- commandline argument
448 Iterable allowed_types -- list of allowed argument types
449 string default_type -- type to return if the argument doesn't specify a type
450 """
451 if ARG_TYPE_DELIMITER not in arg:
452 return default_type, arg
453 arg_type, arg_value = arg.split(ARG_TYPE_DELIMITER, 1)
454 if not arg_type:
455 return default_type, arg_value
456 if arg_type not in allowed_types:
457 raise CmdLineInputError(
458 (
459 "'{arg_type}' is not an allowed type for '{arg_full}', use "
460 "{hint}"
461 ).format(
462 arg_type=arg_type,
463 arg_full=arg,
464 hint=", ".join(sorted(allowed_types)),
465 )
466 )
467 return arg_type, arg_value
468
469
470 def is_short_option_expecting_value(arg: str) -> bool:
471 return len(arg) == 2 and arg[0] == "-" and f"{arg[1]}:" in PCS_SHORT_OPTIONS
472
473
474 def is_long_option_expecting_value(arg: str) -> bool:
475 return (
476 len(arg) > 2 and arg[0:2] == "--" and f"{arg[2:]}=" in PCS_LONG_OPTIONS
477 )
478
479
480 def is_option_expecting_value(arg: str) -> bool:
481 return is_short_option_expecting_value(
482 arg
483 ) or is_long_option_expecting_value(arg)
484
485
486 def wait_to_timeout(wait: bool | str | None) -> int:
487 if wait is False:
488 return -1
489 if wait is None:
490 return 0
491 timeout = timeout_to_seconds(wait)
492 if timeout is None:
493 raise CmdLineInputError(f"'{wait}' is not a valid interval value")
494 return timeout
495
496
497 class InputModifiers:
498 def __init__(self, options: Mapping[str, ModifierValueType]):
499 self._defined_options = set(options.keys())
500 self._options = dict(options)
501 self._options.update(
502 {opt: opt in options for opt in MODIFIER_OPTIONS_BOOL}
503 )
504 self._options.update(
505 {opt: options.get(opt, None) for opt in MODIFIER_OPTIONS_VAL}
506 )
507 self._options.update(
508 {
509 OUTPUT_FORMAT_OPTION: options.get(
510 OUTPUT_FORMAT_OPTION, OUTPUT_FORMAT_VALUE_TEXT
511 ),
512 "--wait": options.get("--wait", False),
513 }
514 )
515
516 def get_subset(
517 self, *options: str, **custom_options: ModifierValueType
518 ) -> "InputModifiers":
519 opt_dict = {
520 opt: self.get(opt) for opt in options if self.is_specified(opt)
521 }
522 opt_dict.update(custom_options)
523 return InputModifiers(opt_dict)
524
525 def ensure_only_supported(
526 self,
527 *supported_options: str,
528 hint_syntax_changed: str | None = None,
529 output_format_supported: bool = False,
530 ) -> None:
531 # --debug is supported in all commands
532 supported_options_set = set(supported_options) | {"--debug"}
533 if output_format_supported:
534 supported_options_set.add(OUTPUT_FORMAT_OPTION)
535 unsupported_options = self._defined_options - supported_options_set
536 if unsupported_options:
537 pluralize = partial(format_plural, unsupported_options)
538 raise CmdLineInputError(
539 "Specified {option} {option_list} {_is} not supported in this "
540 "command".format(
541 option=pluralize("option"),
542 option_list=format_list(sorted(unsupported_options)),
543 _is=pluralize("is"),
544 ),
545 hint=(
546 "Syntax has changed from previous version. {}".format(
547 SEE_MAN_CHANGES.format(hint_syntax_changed)
548 )
549 if hint_syntax_changed
550 else None
551 ),
552 )
553
554 def ensure_not_mutually_exclusive(self, *mutually_exclusive: str) -> None:
555 """
556 Raise CmdLineInputError if several exclusive options were specified
557
558 mutually_exclusive -- mutually exclusive options
559 """
560 options_to_report = self._defined_options & set(mutually_exclusive)
561 if len(options_to_report) > 1:
562 raise CmdLineInputError(
563 "Only one of {} can be used".format(
564 format_list(sorted(options_to_report))
565 )
566 )
567
568 def ensure_not_incompatible(
569 self, checked: str, incompatible: StringCollection
570 ) -> None:
571 """
572 Raise CmdLineInputError if both the checked and an incompatible option
573 were specified
574
575 checked -- option incompatible with any of incompatible options
576 incompatible -- set of options incompatible with checked
577 """
578 if checked not in self._defined_options:
579 return
580 disallowed = self._defined_options & set(incompatible)
581 if disallowed:
582 raise CmdLineInputError(
583 "'{}' cannot be used with {}".format(
584 checked, format_list(sorted(disallowed))
585 )
586 )
587
588 def ensure_dependency_satisfied(
589 self, main_option: str, dependent_options: StringCollection
590 ) -> None:
591 """
592 Raise CmdLineInputError if any of dependent_options is present and
593 main_option is not present.
594
595 main_option -- option on which dependent_options depend
596 dependent_options -- none of these options can be specified if
597 main_option is not specified
598 """
599 if main_option in self._defined_options:
600 return
601 disallowed = self._defined_options & set(dependent_options)
602 if disallowed:
603 raise CmdLineInputError(
604 "{} cannot be used without '{}'".format(
605 format_list(sorted(disallowed)), main_option
606 )
607 )
608
609 def is_specified(self, option: str) -> bool:
610 return option in self._defined_options
611
612 def is_specified_any(self, option_list: StringIterable) -> bool:
613 return any(self.is_specified(option) for option in option_list)
614
615 def get(
616 self, option: str, default: ModifierValueType = None
617 ) -> ModifierValueType:
618 if option in self._defined_options:
|
CID (unavailable; MK=489eaaa87afbb746a3d63c1bcfb03702) (#1 of 1): Copy-paste error (COPY_PASTE_ERROR): |
|
(2) Event copy_paste_error: |
"_options" in "self._options[option]" looks like a copy-paste error. |
|
(3) Event remediation: |
Should it say "_defined_options" instead? |
| Also see events: |
[original] |
619 return self._options[option]
620 if default is not None:
621 return default
622 if option in self._options:
623 return self._options[option]
624 raise AssertionError(f"Non existing default value for '{option}'")
625
626 def get_output_format(
627 self,
628 supported_formats: StringCollection = OUTPUT_FORMAT_VALUES,
629 ) -> str:
630 output_format = self.get(OUTPUT_FORMAT_OPTION)
631 if output_format in supported_formats:
632 return str(output_format)
633 raise CmdLineInputError(
634 (
635 "Unknown value '{value}' for '{option}' option. Supported "
636 "{value_pl} {is_pl}: {supported}"
637 ).format(
638 value=output_format,
639 option=OUTPUT_FORMAT_OPTION,
640 value_pl=format_plural(supported_formats, "value"),
641 is_pl=format_plural(supported_formats, "is"),
642 supported=format_list(list(supported_formats)),
643 )
644 )
645
646
647 def get_rule_str(argv: Argv) -> str | None:
648 if argv:
649 if len(argv) > 1:
650 raise CmdLineInputError()
651 return argv[0]
652 return None
653