b8bb2ff4c3
Use the six library to get monasca-agent to work with python2.7 and python3. Story: 2004148 Task: 27621 Change-Id: I0de315967dd5a745741fda0c53ce8cc85cda8cc5 Signed-off-by: Chuck Short <chucks@redhat.com>
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import logging
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
class TestTail(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.log_file = tempfile.NamedTemporaryFile()
|
|
self.logrotate_config = tempfile.NamedTemporaryFile()
|
|
self.logrotate_config.write("""%s {
|
|
copytruncate
|
|
notifempty
|
|
missingok
|
|
rotate 1
|
|
weekly
|
|
}""" % self.log_file.name)
|
|
self.logrotate_config.flush()
|
|
self.logrotate_state_file = tempfile.NamedTemporaryFile()
|
|
self.last_line = None
|
|
|
|
def _trigger_logrotate(self):
|
|
subprocess.check_call([
|
|
'logrotate',
|
|
'-v', # Verbose logging
|
|
'-f', # Force the rotation even though the file isn't old
|
|
# Create a state file that you have file permissions for
|
|
'-s', self.logrotate_state_file.name,
|
|
self.logrotate_config.name
|
|
])
|
|
|
|
def test_logrotate_copytruncate(self):
|
|
from collector.checks.utils import TailFile
|
|
|
|
def line_parser(l):
|
|
self.last_line = l
|
|
|
|
tail = TailFile(logging.getLogger(), self.log_file.name, line_parser)
|
|
self.assertEqual(tail._size, 0)
|
|
|
|
# Write some data to the log file
|
|
init_string = "hey there, I am a log\n"
|
|
self.log_file.write(init_string)
|
|
self.log_file.flush()
|
|
|
|
# Consume from the tail
|
|
gen = tail.tail(line_by_line=False, move_end=True)
|
|
next(gen)
|
|
|
|
# Verify that the tail consumed the data I wrote
|
|
self.assertEqual(tail._size, len(init_string))
|
|
|
|
try:
|
|
# Trigger a copytruncate logrotation on the log file
|
|
self._trigger_logrotate()
|
|
|
|
# Write a new line to the log file
|
|
new_string = "I am shorter\n"
|
|
self.log_file.write(new_string)
|
|
self.log_file.flush()
|
|
|
|
# Verify that the tail recognized the logrotation
|
|
next(gen)
|
|
self.assertEqual(self.last_line, new_string[:-1], self.last_line)
|
|
except OSError:
|
|
"logrotate is not present"
|
|
|
|
if __name__ == '__main__':
|
|
logging.getLogger().setLevel(logging.DEBUG)
|
|
logging.getLogger().addHandler(logging.StreamHandler())
|
|
unittest.main()
|