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