3
from landscape.broker.remote import RemoteBroker
4
from landscape.textmessage import (
5
AcceptedTypeError, EmptyMessageError, got_accepted_types, get_message,
7
from landscape.tests.helpers import (
8
LandscapeTest, FakeRemoteBrokerHelper, StandardIOHelper)
11
class SendMessageTest(LandscapeTest):
13
helpers = [StandardIOHelper, FakeRemoteBrokerHelper]
15
def test_send_message(self):
17
L{send_message} should send a message of type
18
C{text-message} to the landscape dbus messaging service.
20
service = self.broker_service
21
service.message_store.set_accepted_types(["text-message"])
23
result = send_message(u"Hi there!", self.remote)
24
def got_result(result):
25
messages = service.message_store.get_pending_messages()
26
self.assertEquals(len(messages), 1)
27
self.assertMessage(messages[0], {"type": "text-message",
28
"message": u"Hi there!"})
29
self.assertTrue(service.exchanger.is_urgent())
30
return result.addCallback(got_result)
32
def test_got_accepted_types_without_text_message_type(self):
34
If 'text-message' isn't in the list of accepted types an
35
L{AcceptedTypeError} is raised.
37
self.assertRaises(AcceptedTypeError, got_accepted_types, (),
40
def test_got_accepted_types(self):
42
If 'text-message' is an accepted type a message should be
43
retrieved from the user and sent to the broker.
45
service = self.broker_service
46
service.message_store.set_accepted_types(["text-message"])
48
input = u"Foobl\N{HIRAGANA LETTER A}"
49
self.stdin.write(input.encode("UTF-8"))
52
def got_result(result):
53
messages = service.message_store.get_pending_messages()
54
self.assertEquals(len(messages), 1)
55
self.assertMessage(messages[0],
56
{"type": "text-message",
57
"message": u"Foobl\N{HIRAGANA LETTER A}"})
59
d = got_accepted_types(["text-message"], self.remote, ())
60
d.addCallback(got_result)
64
class ScriptTest(LandscapeTest):
66
helpers = [StandardIOHelper]
68
def test_get_message(self):
70
A message should be properly decoded from the command line arguments.
72
message = get_message(
74
u"\N{HIRAGANA LETTER A}".encode(sys.stdin.encoding), "a!"])
75
self.assertEquals(message, u"\N{HIRAGANA LETTER A} a!")
77
def test_get_message_stdin(self):
79
If no arguments are specified then the message should be read
82
input = u"Foobl\N{HIRAGANA LETTER A}"
83
self.stdin.write(input.encode("UTF-8"))
85
message = get_message(["landscape-message"])
86
self.assertEquals(self.stdout.getvalue(),
87
"Please enter your message, and send EOF "
88
"(Control + D after newline) when done.\n")
89
self.assertEquals(message, input)
91
def test_get_empty_message_stdin(self):
93
If no arguments are specified then the message should be read
96
self.assertRaises(EmptyMessageError, get_message, ["landscape-message"])