mirror of https://github.com/progval/irctest.git
155 lines
3.9 KiB
Python
155 lines
3.9 KiB
Python
import shutil
|
|
import subprocess
|
|
from typing import Optional, Set, Type
|
|
|
|
from irctest.basecontrollers import (
|
|
BaseServerController,
|
|
DirectoryBasedController,
|
|
NotImplementedByController,
|
|
)
|
|
from irctest.cases import BaseServerTestCase
|
|
|
|
TEMPLATE_CONFIG = """
|
|
clients:
|
|
# ping_frequency - client ping frequency
|
|
ping_frequency:
|
|
minutes: 1000
|
|
|
|
# ping_timeout - ping timeout length
|
|
ping_timeout:
|
|
seconds: 10
|
|
data:
|
|
format: json
|
|
filename: {directory}/data.json
|
|
save_frequency:
|
|
minutes: 5
|
|
extensions:
|
|
- mammon.ext.rfc1459.42
|
|
- mammon.ext.rfc1459.ident
|
|
- mammon.ext.ircv3.account_notify
|
|
- mammon.ext.ircv3.server_time
|
|
- mammon.ext.ircv3.echo_message
|
|
- mammon.ext.ircv3.register
|
|
- mammon.ext.ircv3.sasl
|
|
- mammon.ext.misc.nopost
|
|
metadata:
|
|
restricted_keys: []
|
|
whitelist:
|
|
- display-name
|
|
- avatar
|
|
monitor:
|
|
limit: 20
|
|
motd:
|
|
- "Hi"
|
|
limits:
|
|
foo: bar
|
|
listeners:
|
|
- {{"host": "{hostname}", "port": {port}, "ssl": false}}
|
|
logs:
|
|
{{
|
|
}}
|
|
register:
|
|
enabled_callbacks:
|
|
- none
|
|
verify_timeout:
|
|
days: 1
|
|
roles:
|
|
"placeholder":
|
|
title: "Just a placeholder"
|
|
server:
|
|
name: MyLittleServer
|
|
network: MyLittleNetwork
|
|
recvq_len: 20
|
|
"""
|
|
|
|
|
|
def make_list(list_: Set[str]) -> str:
|
|
return "\n".join(map(" - {}".format, list_))
|
|
|
|
|
|
class MammonController(BaseServerController, DirectoryBasedController):
|
|
software_name = "Mammon"
|
|
supported_sasl_mechanisms = {"PLAIN", "ECDSA-NIST256P-CHALLENGE"}
|
|
|
|
def create_config(self) -> None:
|
|
super().create_config()
|
|
with self.open_file("server.conf"):
|
|
pass
|
|
|
|
def kill_proc(self) -> None:
|
|
# Mammon does not seem to handle SIGTERM very well
|
|
assert self.proc
|
|
self.proc.kill()
|
|
|
|
def run(
|
|
self,
|
|
hostname: str,
|
|
port: int,
|
|
*,
|
|
password: Optional[str],
|
|
ssl: bool,
|
|
run_services: bool,
|
|
faketime: Optional[str],
|
|
) -> None:
|
|
if password is not None:
|
|
raise NotImplementedByController("PASS command")
|
|
if ssl:
|
|
raise NotImplementedByController("SSL")
|
|
assert self.proc is None
|
|
self.port = port
|
|
self.create_config()
|
|
with self.open_file("server.yml") as fd:
|
|
fd.write(
|
|
TEMPLATE_CONFIG.format(
|
|
directory=self.directory,
|
|
hostname=hostname,
|
|
port=port,
|
|
)
|
|
)
|
|
# with self.open_file('server.yml', 'r') as fd:
|
|
# print(fd.read())
|
|
assert self.directory
|
|
|
|
if faketime and shutil.which("faketime"):
|
|
faketime_cmd = ["faketime", "-f", faketime]
|
|
self.faketime_enabled = True
|
|
else:
|
|
faketime_cmd = []
|
|
|
|
self.proc = subprocess.Popen(
|
|
[
|
|
*faketime_cmd,
|
|
"mammond",
|
|
"--nofork", # '--debug',
|
|
"--config",
|
|
self.directory / "server.yml",
|
|
]
|
|
)
|
|
|
|
def registerUser(
|
|
self,
|
|
case: BaseServerTestCase,
|
|
username: str,
|
|
password: Optional[str] = None,
|
|
) -> None:
|
|
# XXX: Move this somewhere else when
|
|
# https://github.com/ircv3/ircv3-specifications/pull/152 becomes
|
|
# part of the specification
|
|
client = case.addClient(show_io=False)
|
|
case.sendLine(client, "CAP LS 302")
|
|
case.sendLine(client, "NICK registration_user")
|
|
case.sendLine(client, "USER r e g :user")
|
|
case.sendLine(client, "CAP END")
|
|
while case.getRegistrationMessage(client).command != "001":
|
|
pass
|
|
list(case.getMessages(client))
|
|
case.sendLine(client, "REG CREATE {} passphrase {}".format(username, password))
|
|
msg = case.getMessage(client)
|
|
assert msg.command == "920", msg
|
|
list(case.getMessages(client))
|
|
case.removeClient(client)
|
|
|
|
|
|
def get_irctest_controller_class() -> Type[MammonController]:
|
|
return MammonController
|