22142354625

Committed 18 Feb 2026 01:48PM UTC coverage: 67.452% (+0.2%) from 67.288%

Build # 22142354625

Build Type

push

github

Committed by

web-flow

Commit Message

Fix #216: Implement JSON serialization for user settings (#228)

- Implement JSON-based serialization for complex data types in Users plugin
- Update ACL plugin to use JSON for superadmin and permissions management
- Add deserialize_setting helper for robust decoding across the bot
- Update tests to reflect JSON storage format

Coverage Stats

20 of 21 new or added lines in 2 files covered. (95.24%)

945 of 1401 relevant lines covered (67.45%)

4.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.9

/onebot/plugins/users.py

# -*- coding: utf8 -*-
"""
==============================================
:mod:`onebot.plugins.users` Users plugin
==============================================

Keeps track of the users in channels. Also provides an authorisation system.
This plugin uses WHOIS to figure out someones NickServ account and then links
that to an automatically created, in-bot account.
"""

from __future__ import unicode_literals, print_function

import asyncio
import json
import re
from typing import (
    Any,
    Awaitable,
    Callable,
    Dict,
    Iterable,
    Literal,
    Optional,
    Self,
    Set,
)

import irc3
from irc3.plugins.storage import Storage
from irc3.utils import IrcString


def deserialize_setting(value: Any) -> Any:
    """Safely deserialize a setting value"""
    if not isinstance(value, str):
        return value

    # Try to parse as JSON
    try:
        # Handles JSON lists, dicts, booleans (true/false), null, and numbers
        return json.loads(value)
    except (ValueError, json.JSONDecodeError):
        pass

    # Return as a plain string
    return value


class User(object):
    """User object"""

    def __init__(
        self,
        mask: IrcString,
        channels: Iterable[str],
        id_: Callable[[], Awaitable[str]],
        database=None,
    ):
        self.nick = mask.nick
        self.host = mask.host
        self.channels: Set[str] = set()
        self.id: Callable[[], Awaitable[str]] = id_
        self.database: Optional[Storage] = database
        try:
            if isinstance(channels, str):
                raise ValueError("You must specify a list of channels!")
            for c in iter(channels):
                self.channels.add(c)
        except TypeError:
            raise ValueError("You need to specify in which channel this user is!")

    @property
    def mask(self) -> IrcString:
        """Get the mask of this user"""
        return IrcString("{}!{}".format(self.nick, self.host))

    def _get_database(self) -> Storage:
        if self.database is None:
            raise Exception("No database set for this user.")
        return self.database

    def set_settings(self, settings) -> None:
        """Replaces the settings with the provided dictionary"""

        async def wrapper() -> None:
            id_ = await self.id()
            self._get_database()[id_] = settings

        asyncio.ensure_future(wrapper())

    def set_setting(self, setting: str, value: Any) -> None:
        """Set a specified setting to a value"""

        # Serialize non-string types to JSON for consistent storage across backends
        if not isinstance(value, str):
            if isinstance(value, set):
                value = list(value)
            value = json.dumps(value)

        async def wrapper():
            id_ = await self.id()
            self._get_database().set(id_, **{setting: value})

        asyncio.ensure_future(wrapper())

    async def get_settings(self) -> Dict[str, Any]:
        """Get this users settings"""
        id_ = await self.id()
        return self._get_database().get(id_, dict())  # type: ignore

    async def get_setting(self, setting, default=None) -> Any:
        """Gets a setting for the users. Can be any type."""
        settings = await self.get_settings()
        return deserialize_setting(settings.get(setting, default))

    def join(self, channel) -> None:
        """Register that the user joined a channel"""
        self.channels.add(channel)

    def part(self, channel) -> None:
        """Register that the user parted a channel"""
        self.channels.remove(channel)

    def still_in_channels(self) -> bool:
        """Is the user still in channels?"""
        return len(self.channels) > 0

    def __eq__(self, other: object) -> bool:
        """Compare users by nick

        Since nicks are unique this works for exactly one irc server.
        """
        if not isinstance(other, self.__class__):
            return False
        return self.nick == other.nick


def redact_nick(nick: str) -> str:
    """Inserts a middle dot after the first character of a nick"""
    if len(nick) <= 1:
        return nick
    return nick[0] + "·" + nick[1:]


@irc3.plugin
class UsersPlugin(object):
    """User management plugin for OneBot

    Doesn't do anything with NAMES because we can't get hosts through
    NAMES

    Configuration settings:
        - ``identify_by``: the identification method

    Identification methods available:
        - ``mask``: Use the hostmask
        - ``whatcd``: Get the what.cd username from the host mask
        - ``nickserv``: Parse nickserv info from ``WHOIS``.
    """

    requires = ["irc3.plugins.storage", "irc3.plugins.asynchronious"]

    def __init__(self, bot: irc3.IrcBot):
        """Initialises the plugin"""
        self.bot = bot
        config = bot.config.get(__name__, {})
        method = config.get("identify_by", "mask")
        if method not in ("mask", "nickserv", "whatcd"):
            raise Exception(
                "Invalid configuration: UsersPlugin.identifying_method invalid"
            )
        self.identifying_method: Literal["mask", "nickserv", "whatcd"] = method
        self.log = bot.log.getChild(__name__)
        self.connection_lost()

    @irc3.extend
    def get_user(self, nick: str):
        user = self.active_users.get(nick)
        if not user:
            self.log.warning("Couldn't find %s!", nick)
        return user

    @irc3.extend
    def deserialize_setting(self, value: Any) -> Any:
        return deserialize_setting(value)

    @irc3.extend
    def redact_nicks(self, message: str, target: Optional[str] = None) -> str:
        """Redacts all known nicks in the message.

        If target is provided, only redacts nicks that are in that channel.
        """
        if target:
            nicks = [n for n, u in self.active_users.items() if target in u.channels]
        else:
            nicks = list(self.active_users.keys())

        if not nicks:
            return message

        def replace(match):
            return redact_nick(match.group(0))

        # Sort by length descending to match longest possible nick first
        nicks.sort(key=len, reverse=True)
        escaped_nicks = [re.escape(n) for n in nicks if len(n) > 1]
        if not escaped_nicks:
            return message

        nick_chars = r"A-Za-z0-9_\\\[\]\{\}^`|-"
        pattern = re.compile(
            rf"(?<![{nick_chars}])({'|'.join(escaped_nicks)})(?![{nick_chars}])",
            re.IGNORECASE,
        )
        return pattern.sub(replace, message)

    @irc3.event(irc3.rfc.JOIN_PART_QUIT)
    def on_join_part_quit(self, mask: IrcString, **kwargs):
        event = kwargs["event"]
        self.log.debug("%s %sed", mask.nick, event.lower())
        getattr(self, event.lower())(mask.nick, mask, **kwargs)

    @irc3.event(irc3.rfc.KICK)
    def on_kick(self, mask: IrcString, target: IrcString, **kwargs):
        self.log.debug("%s kicked %s", mask.nick, target.nick)
        self.part(target.nick, target, **kwargs)

    @irc3.event(irc3.rfc.NEW_NICK)
    def on_new_nick(self, nick: IrcString, new_nick: IrcString, **kwargs):
        self.log.debug("%s renamed to %s", nick.nick, new_nick)
        if nick.nick in self.active_users:
            user = self.active_users[nick.nick]
            user.nick = new_nick
            del self.active_users[nick.nick]
            self.active_users[new_nick] = user

    @irc3.event(irc3.rfc.PRIVMSG)
    def on_privmsg(
        self,
        mask: IrcString,
        event: Literal["PRIVMSG", "NOTICE"],
        target: IrcString,
        data=None,
    ):
        if target not in self.channels:
            return
        if mask.is_nick and mask.nick not in self.active_users:
            self.log.debug("Found user %s via PRIVMSG", mask.nick)
            self.active_users[mask.nick] = self.create_user(mask, [target])
        else:
            self.active_users[mask.nick].join(target)

    def connection_lost(self):
        self.channels = set()
        self.active_users = dict()

    def join(self, nick: IrcString, mask: IrcString, channel: IrcString, **kwargs):
        self.log.debug("%s joined channel %s", nick, channel)
        # This can only be observed if we're in that channel
        self.channels.add(channel)
        if nick == self.bot.nick:
            self.bot.send("WHO {}".format(channel))

        if nick not in self.active_users:
            self.active_users[nick] = self.create_user(mask, [channel])

        self.active_users[nick].join(channel)

    def quit(self, nick, _mask, **kwargs):
        if nick == self.bot.nick:
            self.connection_lost()

        if nick in self.active_users:
            del self.active_users[nick]

    def part(self, nick, mask, channel=None, **kwargs):
        if nick == self.bot.nick:
            self.log.info("%s left %s by %s", nick, channel, kwargs["event"])
            for n, user in self.active_users.copy().items():
                user.part(channel)
                if not user.still_in_channels():
                    del self.active_users[n]
            # Remove channel from administration
            self.channels.remove(channel)

        if nick not in self.active_users:
            return

        self.active_users[nick].part(channel)
        if not self.active_users[nick].still_in_channels():
            self.log.debug("Lost %s out of sight", mask.nick)
            del self.active_users[nick]

    @irc3.event(irc3.rfc.RPL_NAMREPLY)
    def on_names(self, channel: IrcString, data: str, **kwargs):
        """Initialise channel list and channel.modes"""
        # possible modes from server
        statusmsg = self.bot.server_config["STATUSMSG"]
        nicknames = data.split(" ")
        if channel not in self.channels:
            self.log.warning("I got NAMES for a channel I'm not in: %", channel)
            return
        for item in nicknames:
            nick = item.strip(statusmsg)
            if nick not in self.active_users:
                # We don't have the mask here, so skip setting up the user
                continue
            self.active_users[nick].join(channel)

    @irc3.event(irc3.rfc.RPL_WHOREPLY)
    def on_who(
        self,
        channel: IrcString,
        nick: IrcString,
        username=None,
        host=None,
        server=None,
        **kwargs,
    ):
        """Process a WHO reply since it could contain new information.

        Should only be processed for channels we are currently in!
        """
        if channel not in self.channels:
            self.log.debug(
                "Got WHO for channel I'm not in: {chan}".format(chan=channel)
            )
            return

        self.log.debug("Got WHO for %s: %s (%s)", channel, nick, host)

        if nick not in self.active_users:
            mask = IrcString("{}!{}@{}".format(nick, username, host))
            self.active_users[nick] = self.create_user(mask, [channel])
        else:
            self.active_users[nick].join(channel)

    def create_user(self, mask: IrcString, channels: Iterable[str | IrcString]):
        """Return a User object"""
        if self.identifying_method == "mask":

            async def mask_id_func() -> str:
                assert mask.host is not None
                return mask.host

            return User(mask, channels, mask_id_func, self.bot.db)
        if self.identifying_method == "nickserv":

            async def get_account() -> str:
                assert mask.nick is not None
                user = self.get_user(mask.nick)
                if hasattr(user, "account"):
                    return user.account
                result = await self.bot.async_cmds.whois(mask.nick)
                if result["success"] and "account" in result:
                    user.account = str(result["account"])
                    return user.account
                else:
                    assert mask.host is not None
                    return mask.host

            return User(mask, channels, get_account, self.bot.db)
        if self.identifying_method == "whatcd":

            async def id_func():
                assert mask.host is not None
                match = re.match(r"^\d+@(.*)\.\w+\.what\.cd", mask.host.lower())
                if match:
                    return match.group(1)
                else:
                    self.log.debug(
                        "Failed to extract what.cd user namefrom {mask}".format(
                            mask=mask
                        )
                    )
                    return mask.host

            return User(mask, channels, id_func, self.bot.db)
        else:  # pragma: no cover
            raise ValueError("A valid identifying method should be configured")

    @classmethod
    def reload(cls, old: Self) -> Self:  # pragma: no cover
        users = old.active_users
        newinstance = cls(old.bot)
        for user in users.values():
            user.database = newinstance.bot.db
        newinstance.channels = old.channels
        newinstance.active_users = users
        return newinstance

1	# -- coding: utf8 --
2	"""
3	==============================================
4	:mod:`onebot.plugins.users` Users plugin
5	==============================================
6
7	Keeps track of the users in channels. Also provides an authorisation system.
8	This plugin uses WHOIS to figure out someones NickServ account and then links
9	that to an automatically created, in-bot account.
10	"""
11
12	from __future__ import unicode_literals, print_function	7✔
13
14	import asyncio	7✔
15	import json	7✔
16	import re	7✔
17	from typing import (	7✔
18	Any,
19	Awaitable,
20	Callable,
21	Dict,
22	Iterable,
23	Literal,
24	Optional,
25	Self,
26	Set,
27	)
28
29	import irc3	7✔
30	from irc3.plugins.storage import Storage	7✔
31	from irc3.utils import IrcString	7✔
32
33
34	def deserialize_setting(value: Any) -> Any:	7✔
35	"""Safely deserialize a setting value"""
36	if not isinstance(value, str):	7✔
37	return value	7✔
38
39	# Try to parse as JSON
40	try:	7✔
41	# Handles JSON lists, dicts, booleans (true/false), null, and numbers
42	return json.loads(value)	7✔
43	except (ValueError, json.JSONDecodeError):	7✔
44	pass	7✔
45
46	# Return as a plain string
47	return value	7✔
48
49
50	class User(object):	7✔
51	"""User object"""
52
53	def __init__(	7✔
54	self,
55	mask: IrcString,
56	channels: Iterable[str],
57	id_: Callable[[], Awaitable[str]],
58	database=None,
59	):
60	self.nick = mask.nick	7✔
61	self.host = mask.host	7✔
62	self.channels: Set[str] = set()	7✔
63	self.id: Callable[[], Awaitable[str]] = id_	7✔
64	self.database: Optional[Storage] = database	7✔
65	try:	7✔
66	if isinstance(channels, str):	7✔
67	raise ValueError("You must specify a list of channels!")	7✔
68	for c in iter(channels):	7✔
69	self.channels.add(c)	7✔
70	except TypeError:	7✔
71	raise ValueError("You need to specify in which channel this user is!")	7✔
72
73	@property	7✔
74	def mask(self) -> IrcString:	7✔
75	"""Get the mask of this user"""
76	return IrcString("{}!{}".format(self.nick, self.host))	7✔
77
78	def _get_database(self) -> Storage:	7✔
79	if self.database is None:	7✔
80	raise Exception("No database set for this user.")	×
81	return self.database	7✔
82
83	def set_settings(self, settings) -> None:	7✔
84	"""Replaces the settings with the provided dictionary"""
85
86	async def wrapper() -> None:	7✔
87	id_ = await self.id()	7✔
88	self._get_database()[id_] = settings	7✔
89
90	asyncio.ensure_future(wrapper())	7✔
91
92	def set_setting(self, setting: str, value: Any) -> None:	7✔
93	"""Set a specified setting to a value"""
94
95	# Serialize non-string types to JSON for consistent storage across backends
96	if not isinstance(value, str):	7✔
97	if isinstance(value, set):	7✔
NEW 98	value = list(value)	×
99	value = json.dumps(value)	7✔
100
101	async def wrapper():	7✔
102	id_ = await self.id()	7✔
103	self._get_database().set(id_, **{setting: value})	7✔
104
105	asyncio.ensure_future(wrapper())	7✔
106
107	async def get_settings(self) -> Dict[str, Any]:	7✔
108	"""Get this users settings"""
109	id_ = await self.id()	7✔
110	return self._get_database().get(id_, dict()) # type: ignore	7✔
111
112	async def get_setting(self, setting, default=None) -> Any:	7✔
113	"""Gets a setting for the users. Can be any type."""
114	settings = await self.get_settings()	7✔
115	return deserialize_setting(settings.get(setting, default))	7✔
116
117	def join(self, channel) -> None:	7✔
118	"""Register that the user joined a channel"""
119	self.channels.add(channel)	7✔
120
121	def part(self, channel) -> None:	7✔
122	"""Register that the user parted a channel"""
123	self.channels.remove(channel)	7✔
124
125	def still_in_channels(self) -> bool:	7✔
126	"""Is the user still in channels?"""
127	return len(self.channels) > 0	7✔
128
129	def __eq__(self, other: object) -> bool:	7✔
130	"""Compare users by nick
131
132	Since nicks are unique this works for exactly one irc server.
133	"""
134	if not isinstance(other, self.__class__):	7✔
135	return False	×
136	return self.nick == other.nick	7✔
137
138
139	def redact_nick(nick: str) -> str:	7✔
140	"""Inserts a middle dot after the first character of a nick"""
141	if len(nick) <= 1:	7✔
142	return nick	×
143	return nick[0] + "·" + nick[1:]	7✔
144
145
146	@irc3.plugin	7✔
147	class UsersPlugin(object):	7✔
148	"""User management plugin for OneBot
149
150	Doesn't do anything with NAMES because we can't get hosts through
151	NAMES
152
153	Configuration settings:
154	- ``identify_by``: the identification method
155
156	Identification methods available:
157	- ``mask``: Use the hostmask
158	- ``whatcd``: Get the what.cd username from the host mask
159	- ``nickserv``: Parse nickserv info from ``WHOIS``.
160	"""
161
162	requires = ["irc3.plugins.storage", "irc3.plugins.asynchronious"]	7✔
163
164	def __init__(self, bot: irc3.IrcBot):	7✔
165	"""Initialises the plugin"""
166	self.bot = bot	7✔
167	config = bot.config.get(__name__, {})	7✔
168	method = config.get("identify_by", "mask")	7✔
169	if method not in ("mask", "nickserv", "whatcd"):	7✔
170	raise Exception(	×
171	"Invalid configuration: UsersPlugin.identifying_method invalid"
172	)
173	self.identifying_method: Literal["mask", "nickserv", "whatcd"] = method	7✔
174	self.log = bot.log.getChild(__name__)	7✔
175	self.connection_lost()	7✔
176
177	@irc3.extend	7✔
178	def get_user(self, nick: str):	7✔
179	user = self.active_users.get(nick)	7✔
180	if not user:	7✔
181	self.log.warning("Couldn't find %s!", nick)	7✔
182	return user	7✔
183
184	@irc3.extend	7✔
185	def deserialize_setting(self, value: Any) -> Any:	7✔
186	return deserialize_setting(value)	7✔
187
188	@irc3.extend	7✔
189	def redact_nicks(self, message: str, target: Optional[str] = None) -> str:	7✔
190	"""Redacts all known nicks in the message.
191
192	If target is provided, only redacts nicks that are in that channel.
193	"""
194	if target:	7✔
195	nicks = [n for n, u in self.active_users.items() if target in u.channels]	7✔
196	else:
197	nicks = list(self.active_users.keys())	7✔
198
199	if not nicks:	7✔
200	return message	7✔
201
202	def replace(match):	7✔
203	return redact_nick(match.group(0))	7✔
204
205	# Sort by length descending to match longest possible nick first
206	nicks.sort(key=len, reverse=True)	7✔
207	escaped_nicks = [re.escape(n) for n in nicks if len(n) > 1]	7✔
208	if not escaped_nicks:	7✔
209	return message	×
210
211	nick_chars = r"A-Za-z0-9_\\\[\]\{\}^`\|-"	7✔
212	pattern = re.compile(	7✔
213	rf"(?<![{nick_chars}])({'\|'.join(escaped_nicks)})(?![{nick_chars}])",
214	re.IGNORECASE,
215	)
216	return pattern.sub(replace, message)	7✔
217
218	@irc3.event(irc3.rfc.JOIN_PART_QUIT)	7✔
219	def on_join_part_quit(self, mask: IrcString, **kwargs):	7✔
220	event = kwargs["event"]	7✔
221	self.log.debug("%s %sed", mask.nick, event.lower())	7✔
222	getattr(self, event.lower())(mask.nick, mask, **kwargs)	7✔
223
224	@irc3.event(irc3.rfc.KICK)	7✔
225	def on_kick(self, mask: IrcString, target: IrcString, **kwargs):	7✔
226	self.log.debug("%s kicked %s", mask.nick, target.nick)	7✔
227	self.part(target.nick, target, **kwargs)	7✔
228
229	@irc3.event(irc3.rfc.NEW_NICK)	7✔
230	def on_new_nick(self, nick: IrcString, new_nick: IrcString, **kwargs):	7✔
231	self.log.debug("%s renamed to %s", nick.nick, new_nick)	7✔
232	if nick.nick in self.active_users:	7✔
233	user = self.active_users[nick.nick]	7✔
234	user.nick = new_nick	7✔
235	del self.active_users[nick.nick]	7✔
236	self.active_users[new_nick] = user	7✔
237
238	@irc3.event(irc3.rfc.PRIVMSG)	7✔
239	def on_privmsg(	7✔
240	self,
241	mask: IrcString,
242	event: Literal["PRIVMSG", "NOTICE"],
243	target: IrcString,
244	data=None,
245	):
246	if target not in self.channels:	7✔
247	return	7✔
248	if mask.is_nick and mask.nick not in self.active_users:	7✔
249	self.log.debug("Found user %s via PRIVMSG", mask.nick)	7✔
250	self.active_users[mask.nick] = self.create_user(mask, [target])	7✔
251	else:
252	self.active_users[mask.nick].join(target)	7✔
253
254	def connection_lost(self):	7✔
255	self.channels = set()	7✔
256	self.active_users = dict()	7✔
257
258	def join(self, nick: IrcString, mask: IrcString, channel: IrcString, **kwargs):	7✔
259	self.log.debug("%s joined channel %s", nick, channel)	7✔
260	# This can only be observed if we're in that channel
261	self.channels.add(channel)	7✔
262	if nick == self.bot.nick:	7✔
263	self.bot.send("WHO {}".format(channel))	7✔
264
265	if nick not in self.active_users:	7✔
266	self.active_users[nick] = self.create_user(mask, [channel])	7✔
267
268	self.active_users[nick].join(channel)	7✔
269
270	def quit(self, nick, _mask, **kwargs):	7✔
271	if nick == self.bot.nick:	7✔
272	self.connection_lost()	7✔
273
274	if nick in self.active_users:	7✔
275	del self.active_users[nick]	7✔
276
277	def part(self, nick, mask, channel=None, **kwargs):	7✔
278	if nick == self.bot.nick:	7✔
279	self.log.info("%s left %s by %s", nick, channel, kwargs["event"])	7✔
280	for n, user in self.active_users.copy().items():	7✔
281	user.part(channel)	7✔
282	if not user.still_in_channels():	7✔
283	del self.active_users[n]	7✔
284	# Remove channel from administration
285	self.channels.remove(channel)	7✔
286
287	if nick not in self.active_users:	7✔
288	return	7✔
289
290	self.active_users[nick].part(channel)	7✔
291	if not self.active_users[nick].still_in_channels():	7✔
292	self.log.debug("Lost %s out of sight", mask.nick)	7✔
293	del self.active_users[nick]	7✔
294
295	@irc3.event(irc3.rfc.RPL_NAMREPLY)	7✔
296	def on_names(self, channel: IrcString, data: str, **kwargs):	7✔
297	"""Initialise channel list and channel.modes"""
298	# possible modes from server
299	statusmsg = self.bot.server_config["STATUSMSG"]	7✔
300	nicknames = data.split(" ")	7✔
301	if channel not in self.channels:	7✔
302	self.log.warning("I got NAMES for a channel I'm not in: %", channel)	×
303	return	×
304	for item in nicknames:	7✔
305	nick = item.strip(statusmsg)	7✔
306	if nick not in self.active_users:	7✔
307	# We don't have the mask here, so skip setting up the user
308	continue	7✔
309	self.active_users[nick].join(channel)	×
310
311	@irc3.event(irc3.rfc.RPL_WHOREPLY)	7✔
312	def on_who(	7✔
313	self,
314	channel: IrcString,
315	nick: IrcString,
316	username=None,
317	host=None,
318	server=None,
319	**kwargs,
320	):
321	"""Process a WHO reply since it could contain new information.
322
323	Should only be processed for channels we are currently in!
324	"""
325	if channel not in self.channels:	7✔
326	self.log.debug(	7✔
327	"Got WHO for channel I'm not in: {chan}".format(chan=channel)
328	)
329	return	7✔
330
331	self.log.debug("Got WHO for %s: %s (%s)", channel, nick, host)	7✔
332
333	if nick not in self.active_users:	7✔
334	mask = IrcString("{}!{}@{}".format(nick, username, host))	7✔
335	self.active_users[nick] = self.create_user(mask, [channel])	7✔
336	else:
337	self.active_users[nick].join(channel)	7✔
338
339	def create_user(self, mask: IrcString, channels: Iterable[str \| IrcString]):	7✔
340	"""Return a User object"""
341	if self.identifying_method == "mask":	7✔
342
343	async def mask_id_func() -> str:	7✔
344	assert mask.host is not None	7✔
345	return mask.host	7✔
346
347	return User(mask, channels, mask_id_func, self.bot.db)	7✔
348	if self.identifying_method == "nickserv":	7✔
349
350	async def get_account() -> str:	7✔
351	assert mask.nick is not None	7✔
352	user = self.get_user(mask.nick)	7✔
353	if hasattr(user, "account"):	7✔
354	return user.account	7✔
355	result = await self.bot.async_cmds.whois(mask.nick)	7✔
356	if result["success"] and "account" in result:	7✔
357	user.account = str(result["account"])	7✔
358	return user.account	7✔
359	else:
360	assert mask.host is not None	×
361	return mask.host	×
362
363	return User(mask, channels, get_account, self.bot.db)	7✔
364	if self.identifying_method == "whatcd":	7✔
365
366	async def id_func():	7✔
367	assert mask.host is not None	7✔
368	match = re.match(r"^\d+@(.*)\.\w+\.what\.cd", mask.host.lower())	7✔
369	if match:	7✔
370	return match.group(1)	7✔
371	else:
372	self.log.debug(	×
373	"Failed to extract what.cd user namefrom {mask}".format(
374	mask=mask
375	)
376	)
377	return mask.host	×
378
379	return User(mask, channels, id_func, self.bot.db)	7✔
380	else: # pragma: no cover
381	raise ValueError("A valid identifying method should be configured")
382
383	@classmethod
384	def reload(cls, old: Self) -> Self: # pragma: no cover
385	users = old.active_users
386	newinstance = cls(old.bot)
387	for user in users.values():
388	user.database = newinstance.bot.db
389	newinstance.channels = old.channels
390	newinstance.active_users = users
391	return newinstance

thomwiggers / onebot / 22142354625

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous