Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def patch_handler_registry(monkeypatch):
"""
Ensure Handler._registry is not mutated by these tests.
"""
monkeypatch.setattr(Handler, '_registry', [])
def test_command_with_aliases(self):
@command(aliases='mc')
def my_cmd():
"help for my command"
assert len(Handler._registry) == 2
# attempt to re-registor both the command and its alias
for handler in Handler._registry:
copy.deepcopy(handler).register()
assert len(Handler._registry) == 2
"""
return attach(self.func, params)
def attach(func, params):
"""
Given a function and a namespace of possible parameters,
bind any params matching the signature of the function
to that function.
"""
sig = inspect.signature(func)
params = Projection(sig.parameters.keys(), params)
return functools.partial(func, **params)
class ContainsHandler(Handler):
channels = ()
exclude = ()
rate = 1.0
"rate to invoke handler"
doc = None
class_priority = 1
def match(self, message, channel):
return (
self.name in message.lower()
and self._channel_match(channel)
and self._rate_match()
)
def _channel_match(self, channel):
return (
raise TypeError("when must be a date or time object")
def as_cmd(self):
factory = (
schedule.PeriodicCommandFixedDelay.daily_at
if isinstance(self.when, datetime.time)
else schedule.DelayedCommand.at_time
)
return factory(self.when, self)
class JoinHandler(Handler):
_registry = []
class LeaveHandler(Handler):
"""
Handles quits and parts.
"""
_registry = []
def contains(name, channels=(), exclude=(), rate=1.0, priority=1, doc=None, **kwargs):
return ContainsHandler(
name=name,
doc=doc,
channels=channels,
exclude=exclude,
rate=rate,
priority=priority,
**kwargs
def mk_entries():
handlers = (
handler
for handler in Handler._registry
if type(handler) is pmxbot.core.CommandHandler
)
handlers = sorted(handlers, key=operator.attrgetter('name'))
for handler in handlers:
res = "!" + handler.name
if handler.aliases:
alias_names = (alias.name for alias in handler.aliases)
res += " (%s)" % ', '.join(alias_names)
yield res
def process(self, message):
return self.pattern.search(message)
class ContentHandler(ContainsHandler):
"""
A custom handler that by default handles all messages.
"""
class_priority = 5
allow_chain = True
name = ''
class Scheduled(Handler):
_registry = []
class DelayHandler(Scheduled):
def as_cmd(self):
cls = schedule.PeriodicCommand if self.repeat else schedule.DelayedCommand
return cls.after(self.duration, self)
class AtHandler(Scheduled):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
date_types = datetime.date, datetime.datetime, datetime.time
if not isinstance(self.when, date_types):
raise TypeError("when must be a date or time object")
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
date_types = datetime.date, datetime.datetime, datetime.time
if not isinstance(self.when, date_types):
raise TypeError("when must be a date or time object")
def as_cmd(self):
factory = (
schedule.PeriodicCommandFixedDelay.daily_at
if isinstance(self.when, datetime.time)
else schedule.DelayedCommand.at_time
)
return factory(self.when, self)
class JoinHandler(Handler):
_registry = []
class LeaveHandler(Handler):
"""
Handles quits and parts.
"""
_registry = []
def contains(name, channels=(), exclude=(), rate=1.0, priority=1, doc=None, **kwargs):
return ContainsHandler(
name=name,
doc=doc,
channels=channels,
def get_context():
context = get_context()
commands = []
contains = []
by_name = operator.attrgetter('name')
for handler in sorted(pmxbot.core.Handler._registry, key=by_name):
if type(handler) is pmxbot.core.CommandHandler:
commands.append(handler)
elif isinstance(handler, pmxbot.core.ContainsHandler):
contains.append(handler)
context['commands'] = commands
context['contains'] = contains
return context
)
def _channel_match(self, channel):
return (
not self.channels
and not self.exclude
or channel in self.channels
or self.exclude
and channel not in self.exclude
)
def _rate_match(self):
return random.random() <= self.rate
class CommandHandler(Handler):
class_priority = 3
aliases = ()
def decorate(self, func):
self._set_doc(func)
for alias in self.aliases:
func = alias.decorate(func)
return super().decorate(func)
def _set_doc(self, func):
"""
If no doc was explicitly set, use the function's docstring, trimming
whitespace and replacing newlines with spaces.
"""
if not self.doc and func.__doc__:
self.doc = func.__doc__.strip().replace('\n', ' ')