lorax/tests/pylorax/test_monitor.py
Brian C. Lane 6160d340c4 Fix monitor problem with split UTF8 characters
Sometimes, depending on the buffering, or length of data being received,
the end of the data could be the 1st byte or so of a UTF8 character.
This would cause a crash when trying to decode the raw data buffer.

This switches it to only decode once a full line has been found.
It also adds tests for the LogMonitor class.
2021-11-17 15:37:59 -08:00

70 lines
3.0 KiB
Python

import socket
import time
import unittest
from pylorax.monitor import LogMonitor
class LogMonitorTest(unittest.TestCase):
def test_monitor(self):
monitor = LogMonitor(timeout=1)
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((monitor.host, monitor.port))
s.sendall("Just a test string\nwith two and a half\nlines in it".encode("utf8"))
time.sleep(1)
self.assertFalse(monitor.server.log_check())
s.sendall("\nAnother line\nTraceback (Not a real traceback)\n".encode("utf8"))
time.sleep(1)
self.assertTrue(monitor.server.log_check())
self.assertEqual(monitor.server.error_line, "Traceback (Not a real traceback)")
finally:
monitor.shutdown()
def test_monitor_IGNORED(self):
monitor = LogMonitor(timeout=1)
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((monitor.host, monitor.port))
s.sendall("Just a test string\nwith two and a half\nlines in it".encode("utf8"))
time.sleep(1)
self.assertFalse(monitor.server.log_check())
s.sendall("\nAnother line\nIGNORED: Traceback (Not a real traceback)\n".encode("utf8"))
time.sleep(1)
self.assertFalse(monitor.server.log_check())
self.assertEqual(monitor.server.error_line, "")
finally:
monitor.shutdown()
def test_monitor_timeout(self):
# Timeout is in minutes so to shorten the test we pass 0.1
monitor = LogMonitor(timeout=0.1)
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((monitor.host, monitor.port))
s.sendall("Just a test string\nwith two and a half\nlines in it".encode("utf8"))
time.sleep(1)
self.assertFalse(monitor.server.log_check())
time.sleep(7)
self.assertTrue(monitor.server.log_check())
self.assertEqual(monitor.server.error_line, "")
finally:
monitor.shutdown()
def test_monitor_utf8(self):
## If a utf8 character spans the end of the 4096 byte buffer it will fail to
## decode. Test to make sure it is reassembled correctly.
monitor = LogMonitor(timeout=1)
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((monitor.host, monitor.port))
# Simulate a UTF8 character that gets broken into parts by buffering, etc.
data = "Just a test string\nTraceback (Not a real traceback)\nWith A"
s.sendall(data.encode("utf8") + b"\xc3")
time.sleep(1)
self.assertTrue(monitor.server.log_check())
self.assertEqual(monitor.server.error_line, "Traceback (Not a real traceback)")
finally:
monitor.shutdown()