Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Newer
Older
100644 268 lines (205 sloc) 9.562 kb
1a868d1 @gma Imported from SourceForge.
authored
1 # Copyright (C) 2005 Graham Ashton <ashtong@users.sourceforge.net>
2 #
3 # This module is free software, and you may redistribute it and/or modify
4 # it under the same terms as Python itself, so long as this copyright message
5 # and disclaimer are retained in their original form.
6 #
7 # IN NO EVENT SHALL THE AUTHOR BE LIABLE TO ANY PARTY FOR DIRECT, INDIRECT,
8 # SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE USE OF
9 # THIS CODE, EVEN IF THE AUTHOR HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH
10 # DAMAGE.
11 #
12 # THE AUTHOR SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT
13 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
14 # PARTICULAR PURPOSE. THE CODE PROVIDED HEREUNDER IS ON AN "AS IS" BASIS,
15 # AND THERE IS NO OBLIGATION WHATSOEVER TO PROVIDE MAINTENANCE,
16 # SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
17 #
18 # $Id: netsyslog_test.py,v 1.4 2005/07/15 09:49:29 ashtong Exp $
19
20
21 import copy
22 import os
23 import socket
24 import sys
25 import syslog
26 import time
27 import unittest
28
29 from pmock import *
30
31 import netsyslog
32
33
34 class PriPartTest(unittest.TestCase):
35
36 def test_priority_format(self):
37 """Check PRI is correctly formatted"""
38 pri = netsyslog.PriPart(syslog.LOG_LOCAL4, syslog.LOG_NOTICE)
39 self.assertEqual(str(pri), "<165>")
40
41
42 DEFAULT_TIMESTAMP = "Jun 7 09:00:00"
43 DEFAULT_HOSTNAME = "myhost"
44 DEFAULT_HEADER = "%s %s" % (DEFAULT_TIMESTAMP, DEFAULT_HOSTNAME)
45
46
47 class MockHeaderTest(unittest.TestCase):
48
49 def mock_localtime(self):
50 return (2005, 6, 7, 9, 0, 0, 1, 158, 1) # see DEFAULT_TIMESTAMP
51
52 def mock_gethostname(self):
53 return "myhost"
54
55 def setUp(self):
56 self.real_localtime = time.localtime
57 time.localtime = self.mock_localtime
58 self.real_gethostname = socket.gethostname
59 socket.gethostname = self.mock_gethostname
60
61 def tearDown(self):
62 time.localtime = self.real_localtime
63 socket.gethostname = self.real_gethostname
64
65
66 class HeaderPartTest(MockHeaderTest):
67
68 def test_automatic_timestamp(self):
69 """Check HEADER is automatically calculated if not set"""
70 header = netsyslog.HeaderPart()
71 self.assertEqual(str(header),
72 " ".join((DEFAULT_TIMESTAMP, DEFAULT_HOSTNAME)))
73
74 def test_incorrect_characters_disallowed(self):
75 """Check only valid characters are used in the HEADER"""
76 # Only allowed characters are ABNF VCHAR values and space.
77 # Basically, if ord() returns between 32 and 126 inclusive
78 # it's okay.
79 bad_char = u"\x1f" # printable, ord() returns 31
80 header = netsyslog.HeaderPart()
81 header.timestamp = header.timestamp[:-1] + bad_char
82 self.assertEqual(str(header),
83 " ".join((DEFAULT_TIMESTAMP, DEFAULT_HOSTNAME)))
84
85 def test_set_timestamp_manually(self):
86 """Check it is possible to set the timestamp in HEADER manually"""
87 timestamp = "Jan 31 18:12:34"
88 header = netsyslog.HeaderPart(timestamp=timestamp)
89 self.assertEqual(str(header), "%s %s" % (timestamp, DEFAULT_HOSTNAME))
90
91 def test_set_hostname_manually(self):
92 """Check it is possible to set the hostname in HEADER manually"""
93 hostname = "otherhost"
94 header = netsyslog.HeaderPart(hostname=hostname)
95 self.assertEqual(str(header), "%s %s" % (DEFAULT_TIMESTAMP, hostname))
96
97
98 # check format of time and hostname, set automatically if incorrect
99 # - time is "Mmm dd hh:mm:ss" where dd has leading space, hh leading 0
100 # - single space between time and hostname
101 # - no space in hostname
102 # - if using hostname, not IP, no dots allowed
103 # print message to stderr if badly formatted message encountered
104
105
106 DEFAULT_TAG = "program"
107 MOCK_PID = 1234
108
109
110 class MockMsgTest(unittest.TestCase):
111
112 def mock_getpid(self):
113 return MOCK_PID
114
115 def setUp(self):
116 self.real_argv = sys.argv
117 sys.argv = [DEFAULT_TAG]
118 self.real_getpid = os.getpid
119 os.getpid = self.mock_getpid
120
121 def tearDown(self):
122 sys.argv = self.real_argv
123 os.getpid = self.real_getpid
124
125
126 class MsgPartTest(MockMsgTest):
127
128 def test_tag_defaults_to_progname(self):
129 """Check TAG defaults to program name"""
130 msg = netsyslog.MsgPart()
131 self.assertEqual(msg.tag, DEFAULT_TAG)
132
133 def test_override_tag(self):
134 """Check TAG can be set manually"""
135 msg = netsyslog.MsgPart(tag="mytag")
136 self.assertEqual(msg.tag, "mytag")
137
138 def test_tag_trimmed_if_too_long(self):
139 """Check long TAGs are trimmed to 32 characters"""
140 tag = "abcd" * 10
141 msg = netsyslog.MsgPart(tag=tag)
142 self.assertEqual(msg.tag, tag[:32])
143
144 def test_space_prefixed_to_content(self):
145 """Check single space inserted infront of CONTENT if necessary"""
146 msg = netsyslog.MsgPart("program", content="hello")
147 self.assertEqual(str(msg), "program: hello")
148
149 def test_space_only_added_if_necessary(self):
150 """Check space only added to CONTENT if necessary"""
151 msg = netsyslog.MsgPart("program", content=" hello")
152 self.assertEqual(str(msg), "program hello")
153
154 def test_include_pid(self):
155 """Check the program's pid can be included in CONTENT"""
156 msg = netsyslog.MsgPart("program", "hello", pid=MOCK_PID)
157 self.assertEqual(str(msg), "program[%d]: hello" % (MOCK_PID))
158
159
160 DEFAULT_PRI = netsyslog.PriPart(syslog.LOG_LOCAL4, syslog.LOG_NOTICE)
161 DEFAULT_HEADER = netsyslog.HeaderPart(DEFAULT_TIMESTAMP, DEFAULT_HOSTNAME)
162 DEFAULT_MSG = netsyslog.MsgPart(DEFAULT_TAG, "hello")
163
164
165 class PacketTest(unittest.TestCase):
166
167 def test_message_format(self):
168 """Check syslog message is correctly constructed"""
169 packet = netsyslog.Packet(DEFAULT_PRI, DEFAULT_HEADER, DEFAULT_MSG)
170 header = " ".join((DEFAULT_TIMESTAMP, DEFAULT_HOSTNAME))
171 start_of_packet = "<165>%s %s" % (header, DEFAULT_TAG)
172 self.assert_(str(packet).startswith(start_of_packet))
173
174 def test_max_length(self):
175 """Check that no syslog packet is longer than 1024 bytes"""
176 message = "a" * 2048
177 packet = netsyslog.Packet(DEFAULT_PRI, DEFAULT_HEADER, message)
178 self.assertEqual(len(str(packet)), netsyslog.Packet.MAX_LEN)
179
180
181 class LoggerTest(MockHeaderTest, MockMsgTest):
182
183 def mock_socket(self, family, proto):
184 return self.mock_sock
185
186 def setUp(self):
187 MockHeaderTest.setUp(self)
188 MockMsgTest.setUp(self)
189 self.mock_sock = Mock()
190 self.real_socket = socket.socket
191 socket.socket = self.mock_socket
192
193 def tearDown(self):
194 socket.socket = self.real_socket
195 MockMsgTest.tearDown(self)
196 MockHeaderTest.tearDown(self)
197
198 def test_send_message(self):
199 """Check we can send a message via UDP"""
200 logger = netsyslog.Logger()
201
202 packet = netsyslog.Packet(DEFAULT_PRI, DEFAULT_HEADER, DEFAULT_MSG)
203 hostname = "localhost"
204 address = (hostname, netsyslog.Logger.PORT)
205 self.mock_sock.expects(once()).sendto(eq(str(packet)), eq(address))
206 logger.add_host(hostname)
207
208 hostname = "remotehost"
209 address = (hostname, netsyslog.Logger.PORT)
210 self.mock_sock.expects(once()).sendto(eq(str(packet)), eq(address))
211 logger.add_host(hostname)
212
213 logger.log(syslog.LOG_LOCAL4, syslog.LOG_NOTICE, "hello")
214 self.mock_sock.verify()
215
216 def test_remove_host(self):
217 """Check host can be removed from list of those receiving messages"""
218 hostname = "localhost"
219 logger = netsyslog.Logger()
220 logger.add_host(hostname)
221 logger.remove_host(hostname)
222 logger.log(syslog.LOG_LOCAL4, syslog.LOG_NOTICE, "hello")
223 self.mock_sock.verify()
224
225 def test_pid_not_included_by_default(self):
226 """Check the program's pid is not included by default"""
227 packet = "<165>%s myhost program: hello" % DEFAULT_TIMESTAMP
228 hostname = "localhost"
229 address = (hostname, netsyslog.Logger.PORT)
230 self.mock_sock.expects(once()).sendto(eq(packet), eq(address))
231
232 logger = netsyslog.Logger()
233 logger.add_host(hostname)
234 logger.log(syslog.LOG_LOCAL4, syslog.LOG_NOTICE, "hello")
235 self.mock_sock.verify()
236
237 def test_include_pid(self):
238 """Check the program's pid can be included in a log message"""
239 packet = "<165>%s myhost program[1234]: hello" % DEFAULT_TIMESTAMP
240 hostname = "localhost"
241 address = (hostname, netsyslog.Logger.PORT)
242 self.mock_sock.expects(once()).sendto(eq(packet), eq(address))
243
244 logger = netsyslog.Logger()
245 logger.add_host(hostname)
246 logger.log(syslog.LOG_LOCAL4, syslog.LOG_NOTICE, "hello", pid=True)
247 self.mock_sock.verify()
248
249 def test_send_packets_by_hand(self):
250 """Check we can send a hand crafted log packet"""
251 hostname = "localhost"
252 address = (hostname, netsyslog.Logger.PORT)
253 packet_text = "<165>Jan 1 10:00:00 myhost myprog: hello"
254 self.mock_sock.expects(once()).sendto(eq(packet_text), eq(address))
255
256 pri = netsyslog.PriPart(syslog.LOG_LOCAL4, syslog.LOG_NOTICE)
257 header = netsyslog.HeaderPart("Jan 1 10:00:00", "myhost")
258 msg = netsyslog.MsgPart("myprog", "hello")
259 packet = netsyslog.Packet(pri, header, msg)
260 logger = netsyslog.Logger()
261 logger.add_host(hostname)
262 logger.send_packet(packet)
263 self.mock_sock.verify()
264
265
266 if __name__ == "__main__":
267 unittest.main()
Something went wrong with that request. Please try again.