add a test for utf8 enforcement

irctest/client_mock.py

@@ -89,9 +89,14 @@ class ClientMock:
             if not filter_pred or filter_pred(message):
                 return message
     def sendLine(self, line):
-        if not line.endswith('\r\n'):
-            line += '\r\n'
-        encoded_line = line.encode()
+        if isinstance(line, str):
+            encoded_line = line.encode()
+        elif isinstance(line, bytes):
+            encoded_line = line
+        else:
+            raise ValueError(line)
+        if not encoded_line.endswith(b'\r\n'):
+            encoded_line += b'\r\n'
         try:
             ret = self.conn.sendall(encoded_line)
         except BrokenPipeError:

irctest/controllers/oragono.py

@@ -35,6 +35,7 @@ BASE_CONFIG = {
             "ban-message": "Try again later",
             "exempted": ["localhost"],
         },
+        'enforce-utf8': True,
     },
 
     'accounts': {

irctest/server_tests/test_utf8.py

@@ -0,0 +1,22 @@
+from irctest import cases
+from irctest.numerics import ERR_UNKNOWNERROR
+
+class Utf8TestCase(cases.BaseServerTestCase, cases.OptionalityHelper):
+    @cases.SpecificationSelector.requiredBySpecification('Oragono')
+    def testUtf8Validation(self):
+        self.connectClient('bar', capabilities=['batch', 'echo-message', 'labeled-response', 'message-tags'])
+        self.joinChannel(1, '#qux')
+        self.sendLine(1, 'PRIVMSG #qux hi')
+        ms = self.getMessages(1)
+        self.assertMessageEqual([m for m in ms if m.command == 'PRIVMSG'][0], params=['#qux', 'hi'])
+
+        self.sendLine(1, b'PRIVMSG #qux hi\xaa')
+        ms = self.getMessages(1)
+        self.assertEqual(len(ms), 1)
+        self.assertEqual(ms[0].command, ERR_UNKNOWNERROR)
+
+        self.sendLine(1, b'@label=xyz PRIVMSG #qux hi\xaa')
+        ms = self.getMessages(1)
+        self.assertEqual(len(ms), 1)
+        self.assertEqual(ms[0].command, ERR_UNKNOWNERROR)
+        self.assertEqual(ms[0].tags.get('label'), 'xyz')