diff --git a/evennia/server/is_typing.py b/evennia/server/is_typing.py index f6c69b62d9..aca2371d7a 100644 --- a/evennia/server/is_typing.py +++ b/evennia/server/is_typing.py @@ -54,7 +54,7 @@ def is_typing_setup(session, *args, **kwargs): nick.value[2] # this is the 'nick' for nick in session.puppet.nicks.all() if nick.value[1].split(" ")[0] in live_report_commands # spaced keywords - or nick[1][0] in live_report_commands # handle things like ', ", : + or nick.value[1][0] in live_report_commands # handle things like ', ", : ] session.msg( diff --git a/evennia/server/portal/telnet.py b/evennia/server/portal/telnet.py index 4bc8af1fba..46d1de0baa 100644 --- a/evennia/server/portal/telnet.py +++ b/evennia/server/portal/telnet.py @@ -10,12 +10,6 @@ sessions etc. import re from django.conf import settings -from evennia.server.portal import mssp, naws, suppress_ga, telnet_oob, ttype -from evennia.server.portal.mccp import MCCP, Mccp, mccp_compress -from evennia.server.portal.mxp import Mxp, mxp_parse -from evennia.server.portal.naws import NAWS -from evennia.utils import ansi -from evennia.utils.utils import class_from_module, to_bytes from twisted.conch.telnet import ( ECHO, GA, @@ -34,6 +28,13 @@ from twisted.conch.telnet import ( from twisted.internet import protocol from twisted.internet.task import LoopingCall +from evennia.server.portal import mssp, naws, suppress_ga, telnet_oob, ttype +from evennia.server.portal.mccp import MCCP, Mccp, mccp_compress +from evennia.server.portal.mxp import Mxp, mxp_parse +from evennia.server.portal.naws import NAWS +from evennia.utils import ansi +from evennia.utils.utils import class_from_module, to_bytes + _RE_N = re.compile(r"\|n$") _RE_LEND = re.compile(rb"\n$|\r$|\r\n$|\r\x00$|", re.MULTILINE) _RE_LINEBREAK = re.compile(rb"\n\r|\r\n|\n|\r", re.DOTALL + re.MULTILINE) @@ -44,7 +45,7 @@ _IDLE_COMMAND = str.encode(settings.IDLE_COMMAND + "\n") # identify HTTP indata _HTTP_REGEX = re.compile( - rb"(GET|HEAD|POST|PUT|DELETE|TRACE|OPTIONS|CONNECT|PATCH) (.*? HTTP/[0-9]\.[0-9])", re.I + b"(GET|HEAD|POST|PUT|DELETE|TRACE|OPTIONS|CONNECT|PATCH) (.*? HTTP/[0-9]\.[0-9])", re.I ) _HTTP_WARNING = bytes( @@ -305,8 +306,6 @@ class TelnetProtocol(Telnet, StatefulTelnetProtocol, _BASE_SESSION_CLASS): """ self.sessionhandler.disconnect(self) - if self.nop_keep_alive and self.nop_keep_alive.running: - self.toggle_nop_keepalive() self.transport.loseConnection() def applicationDataReceived(self, data): @@ -330,6 +329,7 @@ class TelnetProtocol(Telnet, StatefulTelnetProtocol, _BASE_SESSION_CLASS): data = [_IDLE_COMMAND] else: data = _RE_LINEBREAK.split(data) + if len(data) > 2 and _HTTP_REGEX.match(data[0]): # guard against HTTP request on the Telnet port; we # block and kill the connection. diff --git a/evennia/server/tests/test_is_typing.py b/evennia/server/tests/test_is_typing.py new file mode 100644 index 0000000000..d2759a684f --- /dev/null +++ b/evennia/server/tests/test_is_typing.py @@ -0,0 +1,292 @@ +""" +Tests for the is_typing module. + +Covers: +- is_typing_get_audience: audience filtering by location +- is_typing_setup: setup message with commands, aliases, and nicks +- is_typing_state: state broadcast with per-session ISTYPING flag +""" + +from unittest.mock import MagicMock, patch + +from evennia.commands.command import Command +from evennia.server.is_typing import is_typing_get_audience, is_typing_setup, is_typing_state +from evennia.utils.test_resources import BaseEvenniaTest + + +# --------------------------------------------------------------------------- +# Minimal test commands with client_live_report_typing +# --------------------------------------------------------------------------- + + +class _SayCmd(Command): + """Test stand-in for CmdSay.""" + + key = "say" + aliases = ['"', "'"] + locks = "cmd:all()" + client_live_report_typing = True + + def func(self): + pass + + +class _PoseCmd(Command): + """Test stand-in for CmdPose.""" + + key = "pose" + aliases = [":", "emote"] + locks = "cmd:all()" + client_live_report_typing = True + + def func(self): + pass + + +class _NormalCmd(Command): + """A command without live_report_typing -- should be excluded.""" + + key = "look" + aliases = ["l"] + locks = "cmd:all()" + + def func(self): + pass + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_mock_puppet_session(istyping=True): + """Return a mock session suitable for use as a puppet's session.""" + sess = MagicMock() + sess.protocol_flags = {"ISTYPING": istyping} + return sess + + +# --------------------------------------------------------------------------- +# is_typing_get_audience +# --------------------------------------------------------------------------- + + +class TestIsTypingGetAudience(BaseEvenniaTest): + """Tests for is_typing_get_audience.""" + + def setUp(self): + super().setUp() + self.session.puppet = self.char1 + + def test_excludes_typer_from_audience(self): + """Other characters in the same room are returned; the typer is not.""" + audience = is_typing_get_audience(self.session) + self.assertIn(self.char2, audience) + self.assertNotIn(self.char1, audience) + + def test_excludes_non_character_objects(self): + """Non-character objects in the room are not included.""" + audience = is_typing_get_audience(self.session) + self.assertNotIn(self.obj1, audience) + self.assertNotIn(self.obj2, audience) + + def test_empty_when_typer_is_alone(self): + """Audience is empty when no other characters share the location.""" + self.char2.location = self.room2 + audience = is_typing_get_audience(self.session) + self.assertEqual(audience, []) + + def test_returns_multiple_others(self): + """All other characters in the same room are included.""" + from evennia.utils import create + + char3 = create.create_object( + self.character_typeclass, key="Char3", location=self.room1, home=self.room1 + ) + try: + audience = is_typing_get_audience(self.session) + self.assertIn(self.char2, audience) + self.assertIn(char3, audience) + self.assertEqual(len(audience), 2) + finally: + char3.delete() + + +# --------------------------------------------------------------------------- +# is_typing_setup +# --------------------------------------------------------------------------- + + +class TestIsTypingSetup(BaseEvenniaTest): + """Tests for is_typing_setup.""" + + def setUp(self): + super().setUp() + self.session.puppet = self.char1 + self.session.msg = MagicMock() + # Provide a controlled cmdset with known commands. + self.say_cmd = _SayCmd() + self.pose_cmd = _PoseCmd() + self.normal_cmd = _NormalCmd() + self.char1.cmdset.current = [self.say_cmd, self.pose_cmd, self.normal_cmd] + + def _get_setup_payload(self): + """Return the payload dict from the first is_typing msg call.""" + self.session.msg.assert_called_once() + kwargs = self.session.msg.call_args.kwargs + self.assertIn("is_typing", kwargs) + return kwargs["is_typing"] + + def test_sends_setup_message(self): + """A setup message is sent to the session when ISTYPING is enabled.""" + is_typing_setup(self.session) + payload = self._get_setup_payload() + self.assertEqual(payload["type"], "setup") + self.assertIn("live_report_keywords", payload["payload"]) + self.assertIn("typing_timeout", payload["payload"]) + + def test_istyping_false_sends_nothing(self): + """No message is sent when ISTYPING is explicitly disabled.""" + self.session.protocol_flags["ISTYPING"] = False + is_typing_setup(self.session) + self.session.msg.assert_not_called() + + def test_istyping_defaults_to_enabled(self): + """ISTYPING defaults to True when absent from protocol_flags.""" + self.session.protocol_flags.pop("ISTYPING", None) + is_typing_setup(self.session) + self.session.msg.assert_called_once() + + def test_includes_command_key(self): + """The key of a live-reporting command is in the keywords.""" + is_typing_setup(self.session) + keywords = self._get_setup_payload()["payload"]["live_report_keywords"] + self.assertIn("say", keywords) + self.assertIn("pose", keywords) + + def test_includes_command_aliases(self): + """Aliases of live-reporting commands are in the keywords.""" + is_typing_setup(self.session) + keywords = self._get_setup_payload()["payload"]["live_report_keywords"] + # CmdSay aliases + self.assertIn('"', keywords) + self.assertIn("'", keywords) + # CmdPose aliases + self.assertIn(":", keywords) + self.assertIn("emote", keywords) + + def test_excludes_non_live_command(self): + """Commands without client_live_report_typing are not in the keywords.""" + is_typing_setup(self.session) + keywords = self._get_setup_payload()["payload"]["live_report_keywords"] + self.assertNotIn("look", keywords) + self.assertNotIn("l", keywords) + + def test_includes_nick_with_spaced_live_command(self): + """A nick whose replacement starts with a live command is included.""" + # "s $1" -> "say $1": first word of replacement is "say" -> match + self.char1.nicks.add("s $1", "say $1") + is_typing_setup(self.session) + keywords = self._get_setup_payload()["payload"]["live_report_keywords"] + self.assertIn("s $1", keywords) + + def test_excludes_nick_not_mapped_to_live_command(self): + """A nick that maps to a non-live command is not included.""" + # "go $1" -> "look $1": "look" is not a live command + self.char1.nicks.add("go $1", "look $1") + is_typing_setup(self.session) + keywords = self._get_setup_payload()["payload"]["live_report_keywords"] + self.assertNotIn("go $1", keywords) + + def test_includes_nick_with_single_char_command_alias(self): + """A nick whose replacement starts with a single-char command alias is included.""" + # "greet" -> "'hello friend": first char of replacement is "'", an alias for say + self.char1.nicks.add("greet", "'hello friend") + is_typing_setup(self.session) + keywords = self._get_setup_payload()["payload"]["live_report_keywords"] + self.assertIn("greet", keywords) + + def test_typing_timeout_value(self): + """The timeout sent matches the module constant.""" + from evennia.server.is_typing import _IS_TYPING_TIMEOUT + + is_typing_setup(self.session) + timeout = self._get_setup_payload()["payload"]["typing_timeout"] + self.assertEqual(timeout, _IS_TYPING_TIMEOUT) + + +# --------------------------------------------------------------------------- +# is_typing_state +# --------------------------------------------------------------------------- + + +class TestIsTypingState(BaseEvenniaTest): + """Tests for is_typing_state.""" + + def setUp(self): + super().setUp() + self.session.puppet = self.char1 + self.char2_session = _make_mock_puppet_session(istyping=True) + # Patch char2.sessions.all — the handler can't be replaced wholesale. + self._sessions_patcher = patch.object( + self.char2.sessions, "all", return_value=[self.char2_session] + ) + self._sessions_patcher.start() + + def tearDown(self): + self._sessions_patcher.stop() + super().tearDown() + + def test_broadcasts_typing_true_to_audience(self): + """A state=True update is sent to other characters' sessions.""" + is_typing_state(self.session, state=True) + self.char2_session.msg.assert_called_once() + payload = self.char2_session.msg.call_args.kwargs["is_typing"] + self.assertEqual(payload["type"], "typing") + self.assertEqual(payload["payload"]["state"], True) + self.assertEqual(payload["payload"]["name"], self.char1.name) + + def test_broadcasts_typing_false_to_audience(self): + """A state=False update (typing stopped) is also broadcast.""" + is_typing_state(self.session, state=False) + self.char2_session.msg.assert_called_once() + payload = self.char2_session.msg.call_args.kwargs["is_typing"] + self.assertEqual(payload["payload"]["state"], False) + + def test_sender_istyping_false_sends_nothing(self): + """If the sender has ISTYPING disabled, no messages are dispatched.""" + self.session.protocol_flags["ISTYPING"] = False + is_typing_state(self.session, state=True) + self.char2_session.msg.assert_not_called() + + def test_receiver_istyping_false_skips_that_session(self): + """A receiver with ISTYPING disabled does not get the message.""" + self.char2_session.protocol_flags["ISTYPING"] = False + is_typing_state(self.session, state=True) + self.char2_session.msg.assert_not_called() + + def test_receiver_without_istyping_flag_defaults_to_enabled(self): + """A receiver session with no ISTYPING key defaults to receiving messages.""" + self.char2_session.protocol_flags = {} + is_typing_state(self.session, state=True) + self.char2_session.msg.assert_called_once() + + def test_only_sessions_with_istyping_receive_message(self): + """When char2 has two sessions, only the ISTYPING=True one gets the message.""" + silent_session = _make_mock_puppet_session(istyping=False) + self.char2.sessions.all.return_value = [self.char2_session, silent_session] + is_typing_state(self.session, state=True) + self.char2_session.msg.assert_called_once() + silent_session.msg.assert_not_called() + + def test_no_audience_means_no_messages(self): + """No messages are sent when there are no other characters in the room.""" + self.char2.location = self.room2 + is_typing_state(self.session, state=True) + self.char2_session.msg.assert_not_called() + + def test_typing_message_includes_typer_name(self): + """The broadcast message identifies the typer by name.""" + is_typing_state(self.session, state=True) + payload = self.char2_session.msg.call_args.kwargs["is_typing"]["payload"] + self.assertEqual(payload["name"], self.char1.name)