Mario Fetka
133 lines
4.9 KiB
Python
133 lines
4.9 KiB
Python
# /****************************************************************************
|
|
# * <Novell-copyright>
|
|
# * Copyright (c) 2001 Novell, Inc. All Rights Reserved.
|
|
# *
|
|
# * This program is free software; you can redistribute it and/or
|
|
# * modify it under the terms of version 2 of the GNU General Public License
|
|
# * as published by the Free Software Foundation.
|
|
# *
|
|
# * This program is distributed in the hope that it will be useful,
|
|
# * but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# * GNU General Public License for more details.
|
|
# *
|
|
# * You should have received a copy of the GNU General Public License
|
|
# * along with this program; if not, contact Novell, Inc.
|
|
# *
|
|
# * To contact Novell about this file by physical or electronic mail, you
|
|
# * may find current contact information at www.novell.com.
|
|
# * </Novell-copyright>
|
|
# ****************************************************************************/
|
|
|
|
import unittest
|
|
|
|
from bongo_web.compose import build_message, send_message
|
|
from bongo.store.CommandStream import CommandStream
|
|
|
|
|
|
class FakeQueue:
|
|
def __init__(self, user):
|
|
self.user = user
|
|
self.calls = []
|
|
|
|
def __getattr__(self, name):
|
|
return lambda *values: self.calls.append((name, *values))
|
|
|
|
|
|
class FakeWire:
|
|
def __init__(self):
|
|
self.writes = []
|
|
|
|
def write(self, value):
|
|
if not isinstance(value, bytes):
|
|
raise TypeError("wire protocol must be bytes")
|
|
self.writes.append(value)
|
|
|
|
def readline(self):
|
|
return b"1000 ready\r\n"
|
|
|
|
def read(self, _length=-1):
|
|
return b""
|
|
|
|
def flush(self):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class FakeSocket:
|
|
def __init__(self):
|
|
self.wire = FakeWire()
|
|
|
|
def makefile(self, mode):
|
|
if mode != "rwb":
|
|
raise AssertionError("binary socket stream required")
|
|
return self.wire
|
|
|
|
|
|
class ComposeTests(unittest.TestCase):
|
|
def test_builds_unicode_message_without_bcc_header(self):
|
|
message, recipients = build_message(
|
|
"sender@example.test",
|
|
{"to": "Alice <alice@example.test>",
|
|
"bcc": "hidden@example.test", "subject": "Grüße", "text": "Hallo"})
|
|
self.assertEqual(recipients, ["alice@example.test", "hidden@example.test"])
|
|
self.assertNotIn("Bcc", message)
|
|
self.assertIn("Grüße", str(message["Subject"]))
|
|
|
|
def test_rejects_injection_and_missing_recipient(self):
|
|
with self.assertRaises(ValueError):
|
|
build_message("sender@example.test", {"to": "a@example.test\r\nBcc: x@y.test"})
|
|
with self.assertRaises(ValueError):
|
|
build_message("sender@example.test", {"subject": "nobody"})
|
|
|
|
def test_uses_native_queue_sequence(self):
|
|
queues = []
|
|
def factory(user):
|
|
queue = FakeQueue(user)
|
|
queues.append(queue)
|
|
return queue
|
|
send_message("sender@example.test", {"to": "to@example.test", "text": "Hi"}, factory)
|
|
names = [call[0] for call in queues[0].calls]
|
|
self.assertEqual(names, ["Create", "StoreFrom", "StoreTo",
|
|
"StoreMessage", "Run", "Quit"])
|
|
|
|
def test_external_sender_keeps_authenticated_owner_and_adds_signature(self):
|
|
queues = []
|
|
def factory(user):
|
|
queue = FakeQueue(user)
|
|
queues.append(queue)
|
|
return queue
|
|
send_message("owner@example.test",
|
|
{"to": "to@example.test", "text": "Hello"}, factory,
|
|
sender="external@example.test", display_name="Alice",
|
|
signature={"text": "Regards", "html": ""})
|
|
self.assertEqual(queues[0].user, "owner@example.test")
|
|
self.assertEqual(queues[0].calls[1],
|
|
("StoreFrom", "owner@example.test",
|
|
"external@example.test"))
|
|
wire_message = queues[0].calls[3][1]
|
|
self.assertIn("From: Alice <external@example.test>", wire_message)
|
|
self.assertIn("-- \r\nRegards", wire_message)
|
|
|
|
def test_html_signature_is_a_multipart_alternative(self):
|
|
message, _recipients = build_message(
|
|
"sender@example.test",
|
|
{"to": "to@example.test", "text": "Hello", "html": "<p>Hello</p>"},
|
|
signature={"text": "Text signature", "html": "<b>HTML signature</b>"})
|
|
self.assertTrue(message.is_multipart())
|
|
self.assertIn("Text signature", message.get_body(preferencelist=("plain",)).get_content())
|
|
self.assertIn("HTML signature", message.get_body(preferencelist=("html",)).get_content())
|
|
|
|
def test_python3_command_stream_uses_binary_protocol(self):
|
|
sock = FakeSocket()
|
|
stream = CommandStream(sock)
|
|
stream.Write("NOOP")
|
|
self.assertEqual(sock.wire.writes, [b"NOOP", b"\r\n"])
|
|
self.assertEqual(stream.GetResponse().message, "ready")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|