Handle override commands.
(self, line)
| 928 | print(json.dumps(status, indent=4)) |
| 929 | |
| 930 | def do_override(self, line): |
| 931 | """Handle override commands.""" |
| 932 | parser = argparse.ArgumentParser(exit_on_error=False) |
| 933 | parser.add_argument("action", choices=["clear", "set", "query"], help="Action to perform: clear, set, or, query.") |
| 934 | parser.add_argument("activity", type=str, nargs="?", default="", help="The activity to set/clear.") |
| 935 | parser.add_argument("--enable", action="store_true", default=None, help="Enable the specified activity.") |
| 936 | parser.add_argument("--disable", action="store_true", default=False, help="Disable the specified activity.") |
| 937 | try: |
| 938 | args = parser.parse_args(line.split()) |
| 939 | args.activity = args.activity[1:-1] if args.activity and args.activity.startswith('"') and args.activity.endswith('"') else args.activity |
| 940 | if args.action == "clear": |
| 941 | status = self.machine.override_clear(args.activity) |
| 942 | print(json.dumps(status, indent=4)) |
| 943 | elif args.action == "query": |
| 944 | status = self.machine.override_query(args.activity) |
| 945 | accepted = status.get('commandStatus', {}).get('accepted', False) |
| 946 | if accepted: |
| 947 | response = status.pop("response", None) |
| 948 | print(json.dumps(status, indent=4)) |
| 949 | if accepted and response: |
| 950 | print(json.dumps(response, indent=None)) |
| 951 | elif args.action == "set": |
| 952 | if args.enable is None: |
| 953 | enable = not args.disable |
| 954 | else: |
| 955 | enable = args.enable |
| 956 | status = self.machine.override_set(args.activity, enable) |
| 957 | print(json.dumps(status, indent=4)) |
| 958 | except argparse.ArgumentError as e: |
| 959 | print("ArgumentError:", e) |
| 960 | except SystemExit: |
| 961 | pass |
| 962 | |
| 963 | def do_quit(self, line): |
| 964 | """Exit the WorkflowMachine CLI.""" |
nothing calls this directly
no test coverage detected