mirror of https://github.com/progval/irctest.git
239 lines
8.7 KiB
Python
239 lines
8.7 KiB
Python
"""
|
|
The JOIN command (`RFC 1459
|
|
<https://datatracker.ietf.org/doc/html/rfc1459#section-4.2.1>`__,
|
|
`RFC 2812 <https://datatracker.ietf.org/doc/html/rfc2812#section-3.2.1>`__,
|
|
`Modern <https://modern.ircdocs.horse/#join-message>`__)
|
|
"""
|
|
|
|
from irctest import cases, runner
|
|
from irctest.irc_utils import ambiguities
|
|
from irctest.numerics import (
|
|
ERR_BADCHANMASK,
|
|
ERR_FORBIDDENCHANNEL,
|
|
ERR_NOSUCHCHANNEL,
|
|
RPL_ENDOFNAMES,
|
|
RPL_NAMREPLY,
|
|
)
|
|
from irctest.patma import ANYSTR, StrRe
|
|
|
|
ERR_BADCHANNAME = "479" # Hybrid only, and conflicts with others
|
|
|
|
|
|
JOIN_ERROR_NUMERICS = {
|
|
ERR_BADCHANMASK,
|
|
ERR_NOSUCHCHANNEL,
|
|
ERR_FORBIDDENCHANNEL,
|
|
ERR_BADCHANNAME,
|
|
}
|
|
|
|
|
|
class JoinTestCase(cases.BaseServerTestCase):
|
|
@cases.mark_specifications("RFC1459", "RFC2812", strict=True)
|
|
def testJoinAllMessages(self):
|
|
"""“If a JOIN is successful, the user receives a JOIN message as
|
|
confirmation and is then sent the channel's topic (using RPL_TOPIC) and
|
|
the list of users who are on the channel (using RPL_NAMREPLY), which
|
|
MUST include the user joining.”
|
|
-- <https://tools.ietf.org/html/rfc2812#section-3.2.1>
|
|
|
|
“If a JOIN is successful, the user is then sent the channel's topic
|
|
(using RPL_TOPIC) and the list of users who are on the channel (using
|
|
RPL_NAMREPLY), which must include the user joining.”
|
|
-- <https://tools.ietf.org/html/rfc1459#section-4.2.1>
|
|
"""
|
|
self.connectClient("foo")
|
|
self.sendLine(1, "JOIN #chan")
|
|
received_commands = {m.command for m in self.getMessages(1)}
|
|
expected_commands = {RPL_NAMREPLY, RPL_ENDOFNAMES, "JOIN"}
|
|
acceptable_commands = expected_commands | {"MODE"}
|
|
self.assertLessEqual( # set inclusion
|
|
expected_commands,
|
|
received_commands,
|
|
"Server sent {} commands, but at least {} were expected.".format(
|
|
received_commands, expected_commands
|
|
),
|
|
)
|
|
self.assertLessEqual( # ditto
|
|
received_commands,
|
|
acceptable_commands,
|
|
"Server sent {} commands, but only {} were expected.".format(
|
|
received_commands, acceptable_commands
|
|
),
|
|
)
|
|
|
|
@cases.mark_specifications("RFC2812")
|
|
def testJoinNamreply(self):
|
|
"""“353 RPL_NAMREPLY
|
|
"( "=" / "*" / "@" ) <channel>
|
|
:[ "@" / "+" ] <nick> *( " " [ "@" / "+" ] <nick> )”
|
|
-- <https://tools.ietf.org/html/rfc2812#section-5.2>
|
|
|
|
This test makes a user join and check what is sent to them.
|
|
"""
|
|
self.connectClient("foo")
|
|
self.sendLine(1, "JOIN #chan")
|
|
|
|
for m in self.getMessages(1):
|
|
if m.command == "353":
|
|
self.assertIn(
|
|
len(m.params),
|
|
(3, 4),
|
|
m,
|
|
fail_msg="RPL_NAM_REPLY with number of arguments "
|
|
"<3 or >4: {msg}",
|
|
)
|
|
params = ambiguities.normalize_namreply_params(m.params)
|
|
self.assertIn(
|
|
params[1],
|
|
"=*@",
|
|
m,
|
|
fail_msg="Bad channel prefix: {item} not in {list}: {msg}",
|
|
)
|
|
self.assertEqual(
|
|
params[2],
|
|
"#chan",
|
|
m,
|
|
fail_msg="Bad channel name: {got} instead of " "{expects}: {msg}",
|
|
)
|
|
self.assertIn(
|
|
params[3],
|
|
{"foo", "@foo", "+foo"},
|
|
m,
|
|
fail_msg="Bad user list: should contain only user "
|
|
'"foo" with an optional "+" or "@" prefix, but got: '
|
|
"{msg}",
|
|
)
|
|
|
|
def testJoinTwice(self):
|
|
self.connectClient("foo")
|
|
self.sendLine(1, "JOIN #chan")
|
|
m = self.getMessage(1)
|
|
self.assertMessageMatch(m, command="JOIN", params=["#chan"])
|
|
self.getMessages(1)
|
|
self.sendLine(1, "JOIN #chan")
|
|
# Note that there may be no message. Both RFCs require replies only
|
|
# if the join is successful, or has an error among the given set.
|
|
for m in self.getMessages(1):
|
|
if m.command == "353":
|
|
self.assertIn(
|
|
len(m.params),
|
|
(3, 4),
|
|
m,
|
|
fail_msg="RPL_NAM_REPLY with number of arguments "
|
|
"<3 or >4: {msg}",
|
|
)
|
|
params = ambiguities.normalize_namreply_params(m.params)
|
|
self.assertIn(
|
|
params[1],
|
|
"=*@",
|
|
m,
|
|
fail_msg="Bad channel prefix: {item} not in {list}: {msg}",
|
|
)
|
|
self.assertEqual(
|
|
params[2],
|
|
"#chan",
|
|
m,
|
|
fail_msg="Bad channel name: {got} instead of " "{expects}: {msg}",
|
|
)
|
|
self.assertIn(
|
|
params[3],
|
|
{"foo", "@foo", "+foo"},
|
|
m,
|
|
fail_msg='Bad user list after user "foo" joined twice '
|
|
"the same channel: should contain only user "
|
|
'"foo" with an optional "+" or "@" prefix, but got: '
|
|
"{msg}",
|
|
)
|
|
|
|
def testJoinPartiallyInvalid(self):
|
|
"""TODO: specify this in Modern"""
|
|
self.connectClient("foo")
|
|
if int(self.targmax.get("JOIN") or "4") < 2:
|
|
raise runner.OptionalExtensionNotSupported("multi-channel JOIN")
|
|
|
|
self.sendLine(1, "JOIN #valid,inv@lid")
|
|
messages = self.getMessages(1)
|
|
received_commands = {m.command for m in messages}
|
|
expected_commands = {RPL_NAMREPLY, RPL_ENDOFNAMES, "JOIN"}
|
|
acceptable_commands = expected_commands | JOIN_ERROR_NUMERICS | {"MODE"}
|
|
self.assertLessEqual(
|
|
expected_commands,
|
|
received_commands,
|
|
"Server sent {} commands, but at least {} were expected.".format(
|
|
received_commands, expected_commands
|
|
),
|
|
)
|
|
self.assertLessEqual(
|
|
received_commands,
|
|
acceptable_commands,
|
|
"Server sent {} commands, but only {} were expected.".format(
|
|
received_commands, acceptable_commands
|
|
),
|
|
)
|
|
|
|
nb_errors = 0
|
|
for m in messages:
|
|
if m.command in JOIN_ERROR_NUMERICS:
|
|
nb_errors += 1
|
|
self.assertMessageMatch(m, params=["foo", "inv@lid", ANYSTR])
|
|
|
|
self.assertEqual(
|
|
nb_errors,
|
|
1,
|
|
fail_msg="Expected 1 error when joining channels '#valid' and 'inv@lid', "
|
|
"got {got}",
|
|
)
|
|
|
|
@cases.mark_capabilities("batch", "labeled-response")
|
|
def testJoinPartiallyInvalidLabeledResponse(self):
|
|
"""TODO: specify this in Modern"""
|
|
self.connectClient(
|
|
"foo", capabilities=["batch", "labeled-response"], skip_if_cap_nak=True
|
|
)
|
|
if int(self.targmax.get("JOIN") or "4") < 2:
|
|
raise runner.OptionalExtensionNotSupported("multi-channel JOIN")
|
|
|
|
self.sendLine(1, "@label=label1 JOIN #valid,inv@lid")
|
|
messages = self.getMessages(1)
|
|
|
|
first_msg = messages.pop(0)
|
|
last_msg = messages.pop(-1)
|
|
self.assertMessageMatch(
|
|
first_msg, command="BATCH", params=[StrRe(r"\+.*"), "labeled-response"]
|
|
)
|
|
batch_id = first_msg.params[0][1:]
|
|
self.assertMessageMatch(last_msg, command="BATCH", params=["-" + batch_id])
|
|
|
|
received_commands = {m.command for m in messages}
|
|
expected_commands = {RPL_NAMREPLY, RPL_ENDOFNAMES, "JOIN"}
|
|
acceptable_commands = expected_commands | JOIN_ERROR_NUMERICS | {"MODE"}
|
|
self.assertLessEqual(
|
|
expected_commands,
|
|
received_commands,
|
|
"Server sent {} commands, but at least {} were expected.".format(
|
|
received_commands, expected_commands
|
|
),
|
|
)
|
|
self.assertLessEqual(
|
|
received_commands,
|
|
acceptable_commands,
|
|
"Server sent {} commands, but only {} were expected.".format(
|
|
received_commands, acceptable_commands
|
|
),
|
|
)
|
|
|
|
nb_errors = 0
|
|
for m in messages:
|
|
self.assertIn("batch", m.tags)
|
|
self.assertEqual(m.tags["batch"], batch_id)
|
|
if m.command in JOIN_ERROR_NUMERICS:
|
|
nb_errors += 1
|
|
self.assertMessageMatch(m, params=["foo", "inv@lid", ANYSTR])
|
|
|
|
self.assertEqual(
|
|
nb_errors,
|
|
1,
|
|
fail_msg="Expected 1 error when joining channels '#valid' and 'inv@lid', "
|
|
"got {got}",
|
|
)
|