blob: 1ebbf0c9b7df9f23e00b3337df216284ae082f07 [file] [log] [blame] [edit]
# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import logging.handlers
import mox
import multiprocessing
import tempfile
import time
import os
import unittest
import log_socket_server
class TestLogSocketServer(mox.MoxTestBase):
"""Test LogSocketServer can start and save logs to a local file.
"""
def log_call(self, value, port):
"""Method to be called in a new process to log to a socket server.
@param value: Data to be logged.
@param port: Port used by log socket server.
"""
logging.getLogger().handlers = []
socketHandler = logging.handlers.SocketHandler('localhost', port)
logging.getLogger().addHandler(socketHandler)
logging.getLogger().level = logging.INFO
logging.info(value)
def testMultiProcessLoggingSuccess(self):
"""Test log can be saved from multiple processes."""
# Start log TCP server.
super(TestLogSocketServer, self).setUp()
log_filename = tempfile.mktemp(suffix='_log_server')
log_socket_server.LogSocketServer.start(filename=log_filename,
level=logging.INFO)
processes = []
process_number = 10
port = log_socket_server.LogSocketServer.port
for i in range(process_number):
process = multiprocessing.Process(target=self.log_call,
args=(i, port))
process.start()
processes.append(process)
for process in processes:
process.join()
# Wait for TCP server to finish processing data. If process_number is
# increased, the wait time should be increased to avoid test flaky.
time.sleep(1)
log_socket_server.LogSocketServer.stop()
# Read log to confirm all logs are written to file.
num_lines = sum(1 for line in open(log_filename))
self.assertEqual(process_number, num_lines, 'Not all log messages were '
'written to file %s. Expected number of logs: %s, '
'Logs found in file: %s' %
(log_filename, process_number, num_lines))
os.remove(log_filename)
if __name__ == "__main__":
unittest.main()