view sat_frontends/jp/cmd_blocking.py @ 3914:4cb38c8312a1

plugin XEP-0384, xml_tools: avoid `getItems` timeout + fix empty node crash + parsing: - use `max_items` in `getItems` calls for bundles, as otherwise some pubsub service may return full nodes, which may be huge is `max_items=1` is not set on the node, possibly resulting in timeouts. - the plugin was crashing when TWOMEMO devices list node has no items at all. This is not the case anymore. - a naive parsing method has been implemented in `xml_tools` to replace the serialisation/deserialisation method. This should be more efficient and will avoid annoying `ns0:` prefixes in XML logs.
author Goffi <goffi@goffi.org>
date Sat, 24 Sep 2022 16:37:46 +0200
parents 12317ba98d99
children 524856bd7b19
line wrap: on
line source

#!/usr/bin/env python3


# jp: a SàT command line tool
# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.


import json
import os
from sat.core.i18n import _
from sat.tools.common import data_format
from sat_frontends.jp import common
from sat_frontends.jp.constants import Const as C
from . import base

__commands__ = ["Blocking"]


class List(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "list",
            use_output=C.OUTPUT_LIST,
            help=_("list blocked entities"),
        )

    def add_parser_options(self):
        pass

    async def start(self):
        try:
            blocked_jids = await self.host.bridge.blockingList(
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't get blocked entities: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            await self.output(blocked_jids)
            self.host.quit(C.EXIT_OK)


class Block(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "block",
            help=_("block one or more entities"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "entities",
            nargs="+",
            metavar="JID",
            help=_("JIDs of entities to block"),
        )

    async def start(self):
        try:
            await self.host.bridge.blockingBlock(
                self.args.entities,
                self.profile
            )
        except Exception as e:
            self.disp(f"can't block entities: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.host.quit(C.EXIT_OK)


class Unblock(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "unblock",
            help=_("unblock one or more entities"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "entities",
            nargs="+",
            metavar="JID",
            help=_("JIDs of entities to unblock"),
        )
        self.parser.add_argument(
            "-f",
            "--force",
            action="store_true",
            help=_('when "all" is used, unblock all entities without confirmation'),
        )

    async def start(self):
        if self.args.entities == ["all"]:
            if not self.args.force:
                await self.host.confirmOrQuit(
                    _("All entities will be unblocked, are you sure"),
                    _("unblock cancelled")
                )
            self.args.entities.clear()
        elif self.args.force:
            self.parser.error(_('--force is only allowed when "all" is used as target'))

        try:
            await self.host.bridge.blockingUnblock(
                self.args.entities,
                self.profile
            )
        except Exception as e:
            self.disp(f"can't unblock entities: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.host.quit(C.EXIT_OK)


class Blocking(base.CommandBase):
    subcommands = (List, Block, Unblock)

    def __init__(self, host):
        super().__init__(
            host, "blocking", use_profile=False, help=_("entities blocking")
        )