val/irctest
1
0
Fork 0
mirror of https://github.com/progval/irctest.git synced 2025-04-06 15:29:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make flake8 pass, and run it automatically.

Browse Source
This commit is contained in:
Valentin Lorentz
2021-02-22 19:27:43 +01:00
committed by Valentin Lorentz
parent 836cc5d6d2
commit f9d0ec18ff
31 changed files with 235 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.pre-commit-config.yaml
View File

@ -9,3 +9,8 @@ repos:
rev: 5.5.2
hooks:
- id: isort
- repo: https://gitlab.com/pycqa/flake8
rev: 3.8.3
hooks:
- id: flake8

5
conftest.py
View File

@ -1,5 +1,4 @@
import importlib
import sys
import unittest
import _pytest.unittest
@ -81,8 +80,8 @@ def pytest_collection_modifyitems(session, config, items):
# and that node references the UnitTest class
assert issubclass(item.parent.cls, unittest.TestCase)
# and in this project, TestCase classes all inherit either from BaseClientController
# or BaseServerController.
# and in this project, TestCase classes all inherit either from
# BaseClientController or BaseServerController.
if issubclass(item.parent.cls, BaseServerTestCase):
if server_tests:
filtered_items.append(item)

2
irctest/basecontrollers.py
View File

@ -147,5 +147,5 @@ class BaseServerController(_BaseController):
c.close()
self.port_open = True
except Exception as e:
except Exception:
continue

17
irctest/cases.py
View File

@ -58,15 +58,9 @@ class _IrcTestCase(unittest.TestCase):
method_doc = self._testMethodDoc
if not method_doc:
return ""
return (
"\t"
+ normalizeWhitespace(
method_doc,
removeNewline=False,
)
.strip()
.replace("\n ", "\n\t")
)
return "\t" + normalizeWhitespace(
method_doc, removeNewline=False
).strip().replace("\n ", "\n\t")
def setUp(self):
super().setUp()
@ -229,10 +223,7 @@ class ClientNegociationHelper:
def readCapLs(self, auth=None, tls_config=None):
(hostname, port) = self.server.getsockname()
self.controller.run(
hostname=hostname,
port=port,
auth=auth,
tls_config=tls_config,
hostname=hostname, port=port, auth=auth, tls_config=tls_config
)
self.acceptClient()
m = self.getMessage()

3
irctest/client_tests/test_sasl.py
View File

@ -1,8 +1,7 @@
import base64
import hashlib
import ecdsa
from ecdsa.util import sigdecode_der, sigencode_der
from ecdsa.util import sigdecode_der
try:
import pyxmpp2_scram as scram

31
irctest/client_tests/test_tls.py
View File

@ -3,7 +3,6 @@ import ssl
from irctest import cases, tls
from irctest.exceptions import ConnectionClosed
from irctest.irc_utils.message_parser import Message
BAD_CERT = """
-----BEGIN CERTIFICATE-----
@ -120,26 +119,20 @@ class TlsTestCase(cases.BaseClientTestCase):
tls_config = tls.TlsConfig(enable=True, trusted_fingerprints=[GOOD_FINGERPRINT])
(hostname, port) = self.server.getsockname()
self.controller.run(
hostname=hostname,
port=port,
auth=None,
tls_config=tls_config,
hostname=hostname, port=port, auth=None, tls_config=tls_config
)
self.acceptClient(tls_cert=GOOD_CERT, tls_key=GOOD_KEY)
m = self.getMessage()
self.getMessage()
def testUntrustedCertificate(self):
tls_config = tls.TlsConfig(enable=True, trusted_fingerprints=[GOOD_FINGERPRINT])
(hostname, port) = self.server.getsockname()
self.controller.run(
hostname=hostname,
port=port,
auth=None,
tls_config=tls_config,
hostname=hostname, port=port, auth=None, tls_config=tls_config
)
self.acceptClient(tls_cert=BAD_CERT, tls_key=BAD_KEY)
with self.assertRaises((ConnectionClosed, ConnectionResetError)):
m = self.getMessage()
self.getMessage()
class StsTestCase(cases.BaseClientTestCase, cases.OptionalityHelper):
@ -162,10 +155,7 @@ class StsTestCase(cases.BaseClientTestCase, cases.OptionalityHelper):
# Connect client to insecure server
(hostname, port) = self.insecure_server.getsockname()
self.controller.run(
hostname=hostname,
port=port,
auth=None,
tls_config=tls_config,
hostname=hostname, port=port, auth=None, tls_config=tls_config
)
self.acceptClient(server=self.insecure_server)
@ -194,10 +184,7 @@ class StsTestCase(cases.BaseClientTestCase, cases.OptionalityHelper):
# Run the client, still configured to connect to the insecure server
self.controller.run(
hostname=hostname,
port=port,
auth=None,
tls_config=tls_config,
hostname=hostname, port=port, auth=None, tls_config=tls_config
)
# The client should remember the STS policy and connect to the secure
@ -208,11 +195,7 @@ class StsTestCase(cases.BaseClientTestCase, cases.OptionalityHelper):
def testStsInvalidCertificate(self):
# Connect client to insecure server
(hostname, port) = self.insecure_server.getsockname()
self.controller.run(
hostname=hostname,
port=port,
auth=None,
)
self.controller.run(hostname=hostname, port=port, auth=None)
self.acceptClient(server=self.insecure_server)
# Send STS policy to client

8
irctest/controllers/charybdis.py
View File

@ -1,10 +1,6 @@
import os
import shutil
import subprocess
import tempfile
import time
from irctest import authentication, client_mock
from irctest.basecontrollers import (
BaseServerController,
DirectoryBasedController,
@ -74,9 +70,7 @@ class CharybdisController(BaseServerController, DirectoryBasedController):
if ssl:
self.gen_ssl()
ssl_config = TEMPLATE_SSL_CONFIG.format(
key_path=self.key_path,
pem_path=self.pem_path,
dh_path=self.dh_path,
key_path=self.key_path, pem_path=self.pem_path, dh_path=self.dh_path
)
else:
ssl_config = ""

8
irctest/controllers/hybrid.py
View File

@ -1,10 +1,6 @@
import os
import shutil
import subprocess
import tempfile
import time
from irctest import authentication, client_mock
from irctest.basecontrollers import (
BaseServerController,
DirectoryBasedController,
@ -72,9 +68,7 @@ class HybridController(BaseServerController, DirectoryBasedController):
if ssl:
self.gen_ssl()
ssl_config = TEMPLATE_SSL_CONFIG.format(
key_path=self.key_path,
pem_path=self.pem_path,
dh_path=self.dh_path,
key_path=self.key_path, pem_path=self.pem_path, dh_path=self.dh_path
)
else:
ssl_config = ""

8
irctest/controllers/inspircd.py
View File

@ -1,10 +1,6 @@
import os
import shutil
import subprocess
import tempfile
import time
from irctest import authentication
from irctest.basecontrollers import (
BaseServerController,
DirectoryBasedController,
@ -62,9 +58,7 @@ class InspircdController(BaseServerController, DirectoryBasedController):
if ssl:
self.gen_ssl()
ssl_config = TEMPLATE_SSL_CONFIG.format(
key_path=self.key_path,
pem_path=self.pem_path,
dh_path=self.dh_path,
key_path=self.key_path, pem_path=self.pem_path, dh_path=self.dh_path
)
else:
ssl_config = ""

6
irctest/controllers/limnoria.py
View File

@ -2,11 +2,7 @@ import os
import subprocess
from irctest import authentication, tls
from irctest.basecontrollers import (
BaseClientController,
DirectoryBasedController,
NotImplementedByController,
)
from irctest.basecontrollers import BaseClientController, DirectoryBasedController
TEMPLATE_CONFIG = """
supybot.directories.conf: {directory}/conf

12
irctest/controllers/mammon.py
View File

@ -1,6 +1,5 @@
import os
import subprocess
import time
from irctest.basecontrollers import (
BaseServerController,
@ -62,16 +61,13 @@ server:
"""
def make_list(l):
return "\n".join(map(" - {}".format, l))
def make_list(list_):
return "\n".join(map(" - {}".format, list_))
class MammonController(BaseServerController, DirectoryBasedController):
software_name = "Mammon"
supported_sasl_mechanisms = {
"PLAIN",
"ECDSA-NIST256P-CHALLENGE",
}
supported_sasl_mechanisms = {"PLAIN", "ECDSA-NIST256P-CHALLENGE"}
supported_capabilities = set() # Not exhaustive
def create_config(self):
@ -115,7 +111,7 @@ class MammonController(BaseServerController, DirectoryBasedController):
self.proc = subprocess.Popen(
[
"mammond",
"--nofork", #'--debug',
"--nofork", # '--debug',
"--config",
os.path.join(self.directory, "server.yml"),
]

55
irctest/controllers/oragono.py
View File

@ -12,9 +12,7 @@ from irctest.basecontrollers import (
OPER_PWD = "frenchfries"
BASE_CONFIG = {
"network": {
"name": "OragonoTest",
},
"network": {"name": "OragonoTest"},
"server": {
"name": "oragono.test",
"listeners": {},
@ -38,11 +36,7 @@ BASE_CONFIG = {
"exempted": ["localhost"],
},
"enforce-utf8": True,
"relaymsg": {
"enabled": True,
"separators": "/",
"available-to-chanops": True,
},
"relaymsg": {"enabled": True, "separators": "/", "available-to-chanops": True},
},
"accounts": {
"authentication-enabled": True,
@ -63,30 +57,19 @@ BASE_CONFIG = {
"method": "strict",
},
},
"channels": {
"registration": {
"enabled": True,
},
},
"datastore": {
"path": None,
},
"channels": {"registration": {"enabled": True}},
"datastore": {"path": None},
"limits": {
"awaylen": 200,
"chan-list-modes": 60,
"channellen": 64,
"kicklen": 390,
"linelen": {
"rest": 2048,
},
"linelen": {"rest": 2048},
"monitor-entries": 100,
"nicklen": 32,
"topiclen": 390,
"whowas-entries": 100,
"multiline": {
"max-bytes": 4096,
"max-lines": 32,
},
"multiline": {"max-bytes": 4096, "max-lines": 32},
},
"history": {
"enabled": True,
@ -118,7 +101,7 @@ BASE_CONFIG = {
"chanreg",
"relaymsg",
],
},
}
},
"opers": {
"root": {
@ -126,19 +109,11 @@ BASE_CONFIG = {
"whois-line": "is a server admin",
# OPER_PWD
"password": "$2a$04$3GzUZB5JapaAbwn7sogpOu9NSiLOgnozVllm2e96LiNPrm61ZsZSq",
},
}
},
}
LOGGING_CONFIG = {
"logging": [
{
"method": "stderr",
"level": "debug",
"type": "*",
},
]
}
LOGGING_CONFIG = {"logging": [{"method": "stderr", "level": "debug", "type": "*"}]}
def hash_password(password):
@ -155,9 +130,7 @@ def hash_password(password):
class OragonoController(BaseServerController, DirectoryBasedController):
software_name = "Oragono"
supported_sasl_mechanisms = {
"PLAIN",
}
supported_sasl_mechanisms = {"PLAIN"}
_port_wait_interval = 0.01
supported_capabilities = set() # Not exhaustive
@ -195,9 +168,7 @@ class OragonoController(BaseServerController, DirectoryBasedController):
config = self.addMysqlToConfig(config)
if enable_roleplay:
config["roleplay"] = {
"enabled": True,
}
config["roleplay"] = {"enabled": True}
if "oragono_config" in self.test_config:
self.test_config["oragono_config"](config)
@ -208,9 +179,7 @@ class OragonoController(BaseServerController, DirectoryBasedController):
if ssl:
self.key_path = os.path.join(self.directory, "ssl.key")
self.pem_path = os.path.join(self.directory, "ssl.pem")
listener_conf = {
"tls": {"cert": self.pem_path, "key": self.key_path},
}
listener_conf = {"tls": {"cert": self.pem_path, "key": self.key_path}}
config["server"]["listeners"][bind_address] = listener_conf
config["datastore"]["path"] = os.path.join(self.directory, "ircd.db")

8
irctest/controllers/sopel.py
View File

@ -11,7 +11,7 @@ host = {hostname}
use_ssl = false
port = {port}
owner = me
channels =
channels =
timeout = 5
auth_username = {username}
auth_password = {password}
@ -21,9 +21,7 @@ auth_password = {password}
class SopelController(BaseClientController):
software_name = "Sopel"
supported_sasl_mechanisms = {
"PLAIN",
}
supported_sasl_mechanisms = {"PLAIN"}
supported_capabilities = set() # Not exhaustive
def __init__(self, test_config):
@ -44,7 +42,7 @@ class SopelController(BaseClientController):
return open(os.path.join(os.path.expanduser("~/.sopel/"), filename), mode)
def create_config(self):
with self.open_file(self.filename) as fd:
with self.open_file(self.filename):
pass
def run(self, hostname, port, auth, tls_config):

4
irctest/irc_utils/capabilities.py
View File

@ -1,6 +1,6 @@
def cap_list_to_dict(l):
def cap_list_to_dict(caps):
d = {}
for cap in l:
for cap in caps:
if "=" in cap:
(key, value) = cap.split("=", 1)
else:

7
irctest/irc_utils/message_parser.py
View File

@ -54,9 +54,4 @@ def parse_message(s):
prefix = None
command = tokens.pop(0)
params = tokens
return Message(
tags=tags,
prefix=prefix,
command=command,
params=params,
)
return Message(tags=tags, prefix=prefix, command=command, params=params)

1
irctest/runner.py
View File

@ -1,5 +1,4 @@
import collections
import operator
import unittest

3
irctest/server_tests/test_away_notify.py
View File

@ -34,7 +34,8 @@ class AwayNotifyTestCase(cases.BaseServerTestCase, cases.OptionalityHelper):
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
def testAwayNotifyOnJoin(self):
"""The away-notify specification states:
"Clients will be sent an AWAY message [...] when a user joins and has an away message set."
"Clients will be sent an AWAY message [...] when a user joins
and has an away message set."
"""
self.connectClient("foo", capabilities=["away-notify"], skip_if_cap_nak=True)
self.getMessages(1)

6
irctest/server_tests/test_cap.py
View File

@ -10,7 +10,7 @@ class CapTestCase(cases.BaseServerTestCase):
“Clients that support capabilities but do not wish to enter
negotiation SHOULD send CAP END upon connection to the server.”
-- <http://ircv3.net/specs/core/capability-negotiation-3.1.html#the-cap-end-subcommand>
"""
""" # noqa
self.addClient(1)
self.sendLine(1, "CAP LS 302")
self.getCapLs(1)
@ -55,7 +55,7 @@ class CapTestCase(cases.BaseServerTestCase):
first 100 characters of the capability list in the REQ subcommand which
triggered the NAK.”
-- <http://ircv3.net/specs/core/capability-negotiation-3.1.html#the-cap-nak-subcommand>
"""
""" # noqa
self.addClient(1)
self.sendLine(1, "CAP LS 302")
self.getCapLs(1)
@ -77,7 +77,7 @@ class CapTestCase(cases.BaseServerTestCase):
"""“The capability identifier set must be accepted as a whole, or
rejected entirely.”
-- <http://ircv3.net/specs/core/capability-negotiation-3.1.html#the-cap-req-subcommand>
"""
""" # noqa
self.addClient(1)
self.sendLine(1, "CAP LS 302")
self.assertIn("multi-prefix", self.getCapLs(1))

50
irctest/server_tests/test_channel_operations.py
View File

@ -51,10 +51,7 @@ class JoinTestCase(cases.BaseServerTestCase):
self.connectClient("foo")
self.sendLine(1, "JOIN #chan")
received_commands = {m.command for m in self.getMessages(1)}
expected_commands = {
"353", # RPL_NAMREPLY
"366", # RPL_ENDOFNAMES
}
expected_commands = {"353", "366"} # RPL_NAMREPLY # RPL_ENDOFNAMES
self.assertTrue(
expected_commands.issubset(received_commands),
"Server sent {} commands, but at least {} were expected.".format(
@ -218,7 +215,8 @@ class JoinTestCase(cases.BaseServerTestCase):
self.getMessages(2)
self.sendLine(1, "PART #chan :bye everyone")
# both the PART'ing client and the other channel member should receive a PART line:
# both the PART'ing client and the other channel member should receive
# a PART line:
m = self.getMessage(1)
self.assertMessageEqual(m, command="PART", params=["#chan", "bye everyone"])
m = self.getMessage(2)
@ -463,7 +461,8 @@ class JoinTestCase(cases.BaseServerTestCase):
@cases.SpecificationSelector.requiredBySpecification("RFC2812")
def testKickPrivileges(self):
"""Test who has the ability to kick / what error codes are sent for invalid kicks."""
"""Test who has the ability to kick / what error codes are sent
for invalid kicks."""
self.connectClient("foo")
self.sendLine(1, "JOIN #chan")
self.getMessages(1)
@ -487,7 +486,8 @@ class JoinTestCase(cases.BaseServerTestCase):
ERR_NOTONCHANNEL in replies
or ERR_CHANOPRIVSNEEDED in replies
or ERR_NOSUCHCHANNEL in replies,
f"did not receive acceptable error code for kick from outside channel: {replies}",
f"did not receive acceptable error code for kick from outside channel: "
f"{replies}",
)
self.joinChannel(3, "#chan")
@ -604,15 +604,15 @@ class JoinTestCase(cases.BaseServerTestCase):
self.getMessages(2)
self.sendLine(1, "INVITE #chan bar")
self.getMessages(1)
l = self.getMessages(2)
messages = self.getMessages(2)
self.assertNotEqual(
l,
messages,
[],
fail_msg="After using “INVITE #chan bar” while #chan does "
"not exist, “bar” received nothing.",
)
self.assertMessageEqual(
l[0],
messages[0],
command="INVITE",
params=["#chan", "bar"],
fail_msg="After “foo” invited “bar” do non-existing channel "
@ -636,15 +636,15 @@ class JoinTestCase(cases.BaseServerTestCase):
self.getMessages(1)
self.getMessages(2)
self.sendLine(1, "INVITE #chan bar")
l = self.getMessages(1)
messages = self.getMessages(1)
self.assertNotEqual(
l,
messages,
[],
fail_msg="After using “INVITE #chan bar” while #chan does "
"not exist, the author received nothing.",
)
self.assertMessageEqual(
l[0],
messages[0],
command="INVITE",
params=["#chan", "bar"],
fail_msg="After “foo” invited “bar” do non-existing channel "
@ -891,24 +891,9 @@ class AuditoriumTestCase(cases.BaseServerTestCase):
return result
self.assertEqual(names("bar"), {"@bar", "guest1", "guest2", "guest3"})
self.assertEqual(
names("guest1"),
{
"@bar",
},
)
self.assertEqual(
names("guest2"),
{
"@bar",
},
)
self.assertEqual(
names("guest3"),
{
"@bar",
},
)
self.assertEqual(names("guest1"), {"@bar"})
self.assertEqual(names("guest2"), {"@bar"})
self.assertEqual(names("guest3"), {"@bar"})
self.sendLine("bar", "MODE #auditorium +v guest1")
modeLine = [msg for msg in self.getMessages("bar") if msg.command == "MODE"][0]
@ -951,7 +936,8 @@ class AuditoriumTestCase(cases.BaseServerTestCase):
class TopicPrivileges(cases.BaseServerTestCase):
@cases.SpecificationSelector.requiredBySpecification("RFC2812")
def testTopicPrivileges(self):
# test the +t channel mode, which prevents unprivileged users from changing the topic
# test the +t channel mode, which prevents unprivileged users
# from changing the topic
self.connectClient("bar", name="bar")
self.joinChannel("bar", "#chan")
self.getMessages("bar")

13
irctest/server_tests/test_chathistory.py
View File

@ -35,9 +35,7 @@ def validate_chathistory_batch(msgs):
class ChathistoryTestCase(cases.BaseServerTestCase):
@staticmethod
def config():
return {
"chathistory": True,
}
return {"chathistory": True}
@cases.SpecificationSelector.requiredBySpecification("Oragono")
def testInvalidTargets(self):
@ -259,7 +257,8 @@ class ChathistoryTestCase(cases.BaseServerTestCase):
],
)
# messages should be stored and retrievable by c1, even though c3 is not registered
# messages should be stored and retrievable by c1,
# even though c3 is not registered
self.getMessages(1)
self.sendLine(1, "CHATHISTORY LATEST %s * 10" % (c3,))
results = [
@ -277,7 +276,8 @@ class ChathistoryTestCase(cases.BaseServerTestCase):
# regression test for #833
self.sendLine(3, "QUIT")
self.assertDisconnected(3)
# register c3 as an account, then attempt to retrieve the conversation history with c1
# register c3 as an account, then attempt to retrieve
# the conversation history with c1
self.controller.registerUser(self, c3, "sesame3")
self.connectClient(
c3,
@ -408,7 +408,8 @@ class ChathistoryTestCase(cases.BaseServerTestCase):
result = validate_chathistory_batch(self.getMessages(user))
self.assertEqual(echo_messages[1:-1], result)
# BETWEEN forwards and backwards with a limit, should get different results this time
# BETWEEN forwards and backwards with a limit, should get
# different results this time
self.sendLine(
user,
"CHATHISTORY BETWEEN %s msgid=%s msgid=%s %d"

4
irctest/server_tests/test_connection_registration.py
View File

@ -140,7 +140,7 @@ class ConnectionRegistrationTestCase(cases.BaseServerTestCase):
IRCv3.2: “Servers MUST NOT send messages described by this document if
the client only supports version 3.1.”
-- <http://ircv3.net/specs/core/capability-negotiation-3.2.html#version-in-cap-ls>
"""
""" # noqa
self.addClient()
self.sendLine(1, "CAP LS")
m = self.getRegistrationMessage(1)
@ -162,7 +162,7 @@ class ConnectionRegistrationTestCase(cases.BaseServerTestCase):
def testEmptyCapList(self):
"""“If no capabilities are active, an empty parameter must be sent.”
-- <http://ircv3.net/specs/core/capability-negotiation-3.1.html#the-cap-list-subcommand>
"""
""" # noqa
self.addClient()
self.sendLine(1, "CAP LIST")
m = self.getRegistrationMessage(1)

156
irctest/server_tests/test_labeled_responses.py
View File

@ -51,7 +51,11 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
m2.tags,
m2,
fail_msg="When sending a PRIVMSG with a label, the target users shouldn't receive the label (only the sending user should): {msg}",
fail_msg=(
"When sending a PRIVMSG with a label, "
"the target users shouldn't receive the label "
"(only the sending user should): {msg}"
),
)
self.assertMessageEqual(
m3,
@ -62,7 +66,11 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
m3.tags,
m3,
fail_msg="When sending a PRIVMSG with a label, the target users shouldn't receive the label (only the sending user should): {msg}",
fail_msg=(
"When sending a PRIVMSG with a label, "
"the target users shouldn't receive the label "
"(only the sending user should): {msg}"
),
)
self.assertMessageEqual(
m4,
@ -73,7 +81,11 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
m4.tags,
m4,
fail_msg="When sending a PRIVMSG with a label, the target users shouldn't receive the label (only the sending user should): {msg}",
fail_msg=(
"When sending a PRIVMSG with a label, "
"the target users shouldn't receive the label "
"(only the sending user should): {msg}"
),
)
self.assertMessageEqual(
@ -109,7 +121,11 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
m2.tags,
m2,
fail_msg="When sending a PRIVMSG with a label, the target user shouldn't receive the label (only the sending user should): {msg}",
fail_msg=(
"When sending a PRIVMSG with a label, "
"the target user shouldn't receive the label "
"(only the sending user should): {msg}"
),
)
self.assertMessageEqual(
@ -121,13 +137,19 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
m.tags,
m,
fail_msg="When sending a PRIVMSG with a label, the echo'd message didn't contain the label at all: {msg}",
fail_msg=(
"When sending a PRIVMSG with a label, "
"the echo'd message didn't contain the label at all: {msg}"
),
)
self.assertEqual(
m.tags["label"],
"12345",
m,
fail_msg="Echo'd PRIVMSG to a client did not contain the same label we sent it with(should be '12345'): {msg}",
fail_msg=(
"Echo'd PRIVMSG to a client did not contain the same label "
"we sent it with(should be '12345'): {msg}"
),
)
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
@ -168,7 +190,11 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
mt.tags,
mt,
fail_msg="When sending a PRIVMSG with a label, the target user shouldn't receive the label (only the sending user should): {msg}",
fail_msg=(
"When sending a PRIVMSG with a label, "
"the target user shouldn't receive the label "
"(only the sending user should): {msg}"
),
)
# ensure sender correctly receives msg
@ -179,13 +205,19 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
ms.tags,
ms,
fail_msg="When sending a PRIVMSG with a label, the source user should receive the label but didn't: {msg}",
fail_msg=(
"When sending a PRIVMSG with a label, "
"the source user should receive the label but didn't: {msg}"
),
)
self.assertEqual(
ms.tags["label"],
"12345",
ms,
fail_msg="Echo'd label doesn't match the label we sent (should be '12345'): {msg}",
fail_msg=(
"Echo'd label doesn't match the label we sent "
"(should be '12345'): {msg}"
),
)
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
@ -214,16 +246,21 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
m.tags["label"],
"12345",
m,
fail_msg="Echo'd label doesn't match the label we sent (should be '12345'): {msg}",
fail_msg=(
"Echo'd label doesn't match the label we sent "
"(should be '12345'): {msg}"
),
)
self.assertEqual(
number_of_labels,
1,
m1,
fail_msg="When sending a PRIVMSG to self with echo-message, we only expect one message to contain the label. Instead, {} messages had the label".format(
number_of_labels
),
fail_msg=(
"When sending a PRIVMSG to self with echo-message, "
"we only expect one message to contain the label. "
"Instead, {} messages had the label"
).format(number_of_labels),
)
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
@ -255,7 +292,11 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
m2.tags,
m2,
fail_msg="When sending a NOTICE with a label, the target user shouldn't receive the label (only the sending user should): {msg}",
fail_msg=(
"When sending a NOTICE with a label, "
"the target user shouldn't receive the label "
"(only the sending user should): {msg}"
),
)
self.assertMessageEqual(
@ -267,13 +308,19 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
m.tags,
m,
fail_msg="When sending a NOTICE with a label, the echo'd message didn't contain the label at all: {msg}",
fail_msg=(
"When sending a NOTICE with a label, "
"the echo'd message didn't contain the label at all: {msg}"
),
)
self.assertEqual(
m.tags["label"],
"12345",
m,
fail_msg="Echo'd NOTICE to a client did not contain the same label we sent it with(should be '12345'): {msg}",
fail_msg=(
"Echo'd NOTICE to a client did not contain the same label "
"we sent it with (should be '12345'): {msg}"
),
)
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
@ -314,7 +361,11 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
mt.tags,
mt,
fail_msg="When sending a NOTICE with a label, the target user shouldn't receive the label (only the sending user should): {msg}",
fail_msg=(
"When sending a NOTICE with a label, "
"the target user shouldn't receive the label "
"(only the sending user should): {msg}"
),
)
# ensure sender correctly receives msg
@ -325,13 +376,19 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
ms.tags,
ms,
fail_msg="When sending a NOTICE with a label, the source user should receive the label but didn't: {msg}",
fail_msg=(
"When sending a NOTICE with a label, "
"the source user should receive the label but didn't: {msg}"
),
)
self.assertEqual(
ms.tags["label"],
"12345",
ms,
fail_msg="Echo'd label doesn't match the label we sent (should be '12345'): {msg}",
fail_msg=(
"Echo'd label doesn't match the label we sent "
"(should be '12345'): {msg}"
),
)
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
@ -358,16 +415,21 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
m.tags["label"],
"12345",
m,
fail_msg="Echo'd label doesn't match the label we sent (should be '12345'): {msg}",
fail_msg=(
"Echo'd label doesn't match the label we sent "
"(should be '12345'): {msg}"
),
)
self.assertEqual(
number_of_labels,
1,
m1,
fail_msg="When sending a NOTICE to self with echo-message, we only expect one message to contain the label. Instead, {} messages had the label".format(
number_of_labels
),
fail_msg=(
"When sending a NOTICE to self with echo-message, "
"we only expect one message to contain the label. "
"Instead, {} messages had the label"
).format(number_of_labels),
)
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
@ -399,7 +461,11 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
m2.tags,
m2,
fail_msg="When sending a TAGMSG with a label, the target user shouldn't receive the label (only the sending user should): {msg}",
fail_msg=(
"When sending a TAGMSG with a label, "
"the target user shouldn't receive the label "
"(only the sending user should): {msg}"
),
)
self.assertIn(
"+draft/reply",
@ -435,13 +501,19 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
m.tags,
m,
fail_msg="When sending a TAGMSG with a label, the echo'd message didn't contain the label at all: {msg}",
fail_msg=(
"When sending a TAGMSG with a label, "
"the echo'd message didn't contain the label at all: {msg}"
),
)
self.assertEqual(
m.tags["label"],
"12345",
m,
fail_msg="Echo'd TAGMSG to a client did not contain the same label we sent it with(should be '12345'): {msg}",
fail_msg=(
"Echo'd TAGMSG to a client did not contain the same label "
"we sent it with (should be '12345'): {msg}"
),
)
self.assertIn(
"+draft/reply",
@ -504,7 +576,11 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
mt.tags,
mt,
fail_msg="When sending a TAGMSG with a label, the target user shouldn't receive the label (only the sending user should): {msg}",
fail_msg=(
"When sending a TAGMSG with a label, "
"the target user shouldn't receive the label "
"(only the sending user should): {msg}"
),
)
# ensure sender correctly receives msg
@ -515,13 +591,19 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
"label",
ms.tags,
ms,
fail_msg="When sending a TAGMSG with a label, the source user should receive the label but didn't: {msg}",
fail_msg=(
"When sending a TAGMSG with a label, "
"the source user should receive the label but didn't: {msg}"
),
)
self.assertEqual(
ms.tags["label"],
"12345",
ms,
fail_msg="Echo'd label doesn't match the label we sent (should be '12345'): {msg}",
fail_msg=(
"Echo'd label doesn't match the label we sent "
"(should be '12345'): {msg}"
),
)
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
@ -548,16 +630,21 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
m.tags["label"],
"12345",
m,
fail_msg="Echo'd label doesn't match the label we sent (should be '12345'): {msg}",
fail_msg=(
"Echo'd label doesn't match the label we sent "
"(should be '12345'): {msg}"
),
)
self.assertEqual(
number_of_labels,
1,
m1,
fail_msg="When sending a TAGMSG to self with echo-message, we only expect one message to contain the label. Instead, {} messages had the label".format(
number_of_labels
),
fail_msg=(
"When sending a TAGMSG to self with echo-message, "
"we only expect one message to contain the label. "
"Instead, {} messages had the label"
).format(number_of_labels),
)
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
@ -609,7 +696,8 @@ class LabeledResponsesTestCase(cases.BaseServerTestCase, cases.OptionalityHelper
self.getMessages(1)
self.sendLine(1, "@label=98765 PING adhoctestline")
# no BATCH should be initiated for a one-line response, it should just be labeled
# no BATCH should be initiated for a one-line response,
# it should just be labeled
ms = self.getMessages(1)
self.assertEqual(len(ms), 1)
m = ms[0]

4
irctest/server_tests/test_lusers.py
View File

@ -55,7 +55,7 @@ class LusersTestCase(cases.BaseServerTestCase):
result.GlobalVisible = int(match.group(1))
result.GlobalInvisible = int(match.group(2))
result.Servers = int(match.group(3))
except:
except Exception:
raise ValueError("corrupt reply for 251 RPL_LUSERCLIENT", luserclient_param)
if RPL_LUSEROP in by_numeric:
@ -78,7 +78,7 @@ class LusersTestCase(cases.BaseServerTestCase):
match = LUSERME_REGEX.match(luserme_param)
localTotalFromUserme = int(match.group(1))
serversFromUserme = int(match.group(2))
except:
except Exception:
raise ValueError("corrupt reply for 255 RPL_LUSERME", luserme_param)
self.assertEqual(result.LocalTotal, localTotalFromUserme)
# serversFromUserme is "servers i'm currently connected to", generally undefined

2
irctest/server_tests/test_message_tags.py
View File

@ -72,7 +72,7 @@ class MessageTagsTestCase(cases.BaseServerTestCase, cases.OptionalityHelper):
getAllMessages()
# test TAGMSG and basic escaping
self.sendLine("bob", "@+buzz=fizz\:buzz;cat=dog;+steel=wootz TAGMSG #test")
self.sendLine("bob", r"@+buzz=fizz\:buzz;cat=dog;+steel=wootz TAGMSG #test")
bob_msg = self.getMessage("bob") # bob has echo-message
alice_msg = self.getMessage("alice")
# carol MUST NOT receive TAGMSG at all

5
irctest/server_tests/test_metadata.py
View File

@ -132,10 +132,7 @@ class MetadataTestCase(cases.BaseServerTestCase):
command="761", # RPL_KEYVALUE
fail_msg="Did not reply with 761 (RPL_KEYVALUE) to a valid "
"“METADATA * SET {} :{}”: {msg}",
extra_format=(
key,
displayable_value,
),
extra_format=(key, displayable_value),
)
self.assertEqual(
m.params[1],

45
irctest/server_tests/test_monitor.py
View File

@ -263,19 +263,27 @@ class MonitorTestCase(cases.BaseServerTestCase):
self.getMessages(1)
self.getMessages(2)
self.sendLine(2, "MONITOR - qux")
l = self.getMessages(2)
messages = self.getMessages(2)
self.assertEqual(
l, [], fail_msg="Got response to “MONITOR -”: {}", extra_format=(l,)
messages,
[],
fail_msg="Got response to “MONITOR -”: {}",
extra_format=(messages,),
)
self.connectClient("qux")
self.getMessages(3)
l = self.getMessages(1)
messages = self.getMessages(1)
self.assertNotEqual(
l, [], fail_msg="Received no message after MONITORed client " "connects."
messages,
[],
fail_msg="Received no message after MONITORed client " "connects.",
)
l = self.getMessages(2)
messages = self.getMessages(2)
self.assertEqual(
l, [], fail_msg="Got response to unmonitored client: {}", extra_format=(l,)
messages,
[],
fail_msg="Got response to unmonitored client: {}",
extra_format=(messages,),
)
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
@ -300,36 +308,17 @@ class MonitorTestCase(cases.BaseServerTestCase):
self.sendLine(1, "MONITOR + qux")
self.getMessages(1)
self.sendLine(1, "MONITOR L")
checkMonitorSubjects(
self.getMessages(1),
"bar",
{
"qux",
},
)
checkMonitorSubjects(self.getMessages(1), "bar", {"qux"})
self.sendLine(1, "MONITOR + bazbat")
self.getMessages(1)
self.sendLine(1, "MONITOR L")
checkMonitorSubjects(
self.getMessages(1),
"bar",
{
"qux",
"bazbat",
},
)
checkMonitorSubjects(self.getMessages(1), "bar", {"qux", "bazbat"})
self.sendLine(1, "MONITOR - qux")
self.getMessages(1)
self.sendLine(1, "MONITOR L")
checkMonitorSubjects(
self.getMessages(1),
"bar",
{
"bazbat",
},
)
checkMonitorSubjects(self.getMessages(1), "bar", {"bazbat"})
@cases.SpecificationSelector.requiredBySpecification("IRCv3.2")
def testNickChange(self):

4
irctest/server_tests/test_relaymsg.py
View File

@ -9,9 +9,7 @@ RELAYMSG_TAG_NAME = "draft/relaymsg"
class RelaymsgTestCase(cases.BaseServerTestCase):
@staticmethod
def config():
return {
"chathistory": True,
}
return {"chathistory": True}
@cases.SpecificationSelector.requiredBySpecification("Oragono")
def testRelaymsg(self):

8
irctest/server_tests/test_resume.py
View File

@ -63,8 +63,8 @@ class ResumeTestCase(cases.BaseServerTestCase):
)
channelMsgTime = privmsgs[0].tags.get("time")
# tokens MUST be cryptographically secure; therefore, this token should be invalid
# with probability at least 1 - 1/(2**128)
# tokens MUST be cryptographically secure; therefore, this token should be
# invalid with probability at least 1 - 1/(2**128)
bad_token = "a" * len(token)
self.addClient()
self.sendLine(3, "CAP LS")
@ -127,8 +127,8 @@ class ResumeTestCase(cases.BaseServerTestCase):
privmsgs[1], command="PRIVMSG", params=["baz", "hello friend singular"]
)
# should replay with the original server-time
# TODO this probably isn't testing anything because the timestamp only has second resolution,
# hence will typically match by accident
# TODO this probably isn't testing anything because the timestamp only
# has second resolution, hence will typically match by accident
self.assertEqual(privmsgs[0].tags.get("time"), channelMsgTime)
# legacy client should receive a QUIT and a JOIN

4
irctest/server_tests/test_roleplay.py
View File

@ -6,9 +6,7 @@ from irctest.numerics import ERR_CANNOTSENDRP
class RoleplayTestCase(cases.BaseServerTestCase):
@staticmethod
def config():
return {
"oragono_roleplay": True,
}
return {"oragono_roleplay": True}
@cases.SpecificationSelector.requiredBySpecification("Oragono")
def testRoleplay(self):

3
irctest/server_tests/test_statusmsg.py
View File

@ -11,7 +11,8 @@ class StatusmsgTestCase(cases.BaseServerTestCase):
@cases.SpecificationSelector.requiredBySpecification("Oragono")
def testStatusmsg(self):
"""Test that STATUSMSG are sent to the intended recipients, with the intended prefixes."""
"""Test that STATUSMSG are sent to the intended recipients,
with the intended prefixes."""
self.connectClient("chanop")
self.joinChannel(1, "#chan")
self.getMessages(1)

29
irctest/server_tests/test_znc_playback.py
View File

@ -20,9 +20,7 @@ def extract_playback_privmsgs(messages):
class ZncPlaybackTestCase(cases.BaseServerTestCase):
@staticmethod
def config():
return {
"chathistory": True,
}
return {"chathistory": True}
@cases.SpecificationSelector.requiredBySpecification("Oragono")
def testZncPlayback(self):
@ -112,7 +110,8 @@ class ZncPlaybackTestCase(cases.BaseServerTestCase):
password=pw,
)
mid_timestamp = ircv3_timestamp_to_unixtime(echo_messages[5].time)
# exclude message 5 itself (oragono's CHATHISTORY implementation corrects for this, but znc.in/playback does not because whatever)
# exclude message 5 itself (oragono's CHATHISTORY implementation
# corrects for this, but znc.in/playback does not because whatever)
mid_timestamp += 0.001
self.sendLine("viewer", "PRIVMSG *playback :play * %s" % (mid_timestamp,))
messages = extract_playback_privmsgs(self.getMessages("viewer"))
@ -139,32 +138,16 @@ class ZncPlaybackTestCase(cases.BaseServerTestCase):
self.sendLine(
"viewer",
"PRIVMSG *playback :play %s %s %s"
% (
chname,
start_timestamp,
end_timestamp,
),
% (chname, start_timestamp, end_timestamp),
)
messages = extract_playback_privmsgs(self.getMessages("viewer"))
self.assertEqual(messages, echo_messages[3:7])
# test nicknames as targets
self.sendLine(
"viewer",
"PRIVMSG *playback :play %s %d"
% (
qux,
early_time,
),
)
self.sendLine("viewer", "PRIVMSG *playback :play %s %d" % (qux, early_time))
messages = extract_playback_privmsgs(self.getMessages("viewer"))
self.assertEqual(messages, [dm])
self.sendLine(
"viewer",
"PRIVMSG *playback :play %s %d"
% (
qux.upper(),
early_time,
),
"viewer", "PRIVMSG *playback :play %s %d" % (qux.upper(), early_time)
)
messages = extract_playback_privmsgs(self.getMessages("viewer"))
self.assertEqual(messages, [dm])