Dependency cleanup: Move test.utils.var_log_messages_before_reboot to sys_utils.
The var_log_messages_before_reboot uses only standard Python module and utils
API and can be in sys_utils. This helps to prevent pulling extra dependency.
Also changed the name from lower_underline to CamelCase:
GetVarLogMessagesBeforeReboot
cros.factory.test.utils now becomes a pure package with empty __init__.py.
BUG=chromium:403712
TEST=make test
Change-Id: Icb1a522d8fb149b28e622ff8af70f4526067f1ac
Reviewed-on: https://chromium-review.googlesource.com/321050
Commit-Ready: Hung-Te Lin <hungte@chromium.org>
Tested-by: Hung-Te Lin <hungte@chromium.org>
Reviewed-by: Wei-Han Chen <stimim@chromium.org>
diff --git a/py/goofy/goofy.py b/py/goofy/goofy.py
index 2739301..5a08bbb 100755
--- a/py/goofy/goofy.py
+++ b/py/goofy/goofy.py
@@ -45,7 +45,6 @@
from cros.factory.test import phase
from cros.factory.test import shopfloor
from cros.factory.test import state
-from cros.factory.test import utils
from cros.factory.test.test_lists import test_lists
from cros.factory.test.e2e_test.common import (
AutomationMode, AutomationModePrompt, ParseAutomationMode)
@@ -399,7 +398,7 @@
return
try:
- var_log_messages = utils.var_log_messages_before_reboot()
+ var_log_messages = sys_utils.GetVarLogMessagesBeforeReboot()
logging.info(
'Tail of /var/log/messages before last reboot:\n'
'%s', ('\n'.join(
@@ -556,7 +555,7 @@
if var_log_messages is None:
try:
var_log_messages = (
- utils.var_log_messages_before_reboot())
+ sys_utils.GetVarLogMessagesBeforeReboot())
# Write it to the log, to make it easier to
# correlate with /var/log/messages.
logging.info(
diff --git a/py/goofy/goofy_rpc.py b/py/goofy/goofy_rpc.py
index 25659df..9f09140 100755
--- a/py/goofy/goofy_rpc.py
+++ b/py/goofy/goofy_rpc.py
@@ -30,7 +30,6 @@
from cros.factory.test import dut
from cros.factory.test import factory
from cros.factory.test import shopfloor
-from cros.factory.test import utils
from cros.factory.test.diagnosis.diagnosis_tool import DiagnosisToolRPC
from cros.factory.test.event import Event, EventClient
from cros.factory.test.test_lists.test_lists import SetActiveTestList
@@ -232,13 +231,13 @@
def GetVarLogMessagesBeforeReboot(
self, lines=100, max_length=5 * 1024 * 1024):
- """Returns the last few lines in /var/log/messages before the current boot.
+ """Returns the last few lines in /var/log/messages before current boot.
Args:
- See utils.var_log_messages_before_reboot.
+ See sys_utils.GetVarLogMessagesBeforeReboot.
"""
- lines = utils.var_log_messages_before_reboot(lines=lines,
- max_length=max_length)
+ lines = sys_utils.GetVarLogMessagesBeforeReboot(lines=lines,
+ max_length=max_length)
if lines:
return unicode('\n'.join(lines) + '\n',
encoding='utf-8', errors='replace')
diff --git a/py/goofy/goofy_rpc_unittest.py b/py/goofy/goofy_rpc_unittest.py
index 9ad96c8..16417fe 100755
--- a/py/goofy/goofy_rpc_unittest.py
+++ b/py/goofy/goofy_rpc_unittest.py
@@ -13,8 +13,8 @@
import factory_common # pylint: disable=W0611
from cros.factory.goofy import goofy_rpc
-from cros.factory.test import utils
from cros.factory.utils import process_utils
+from cros.factory.utils import sys_utils
@contextmanager
@@ -33,7 +33,7 @@
self.mox = mox.Mox()
self.goofy_rpc = goofy_rpc.GoofyRPC(None)
- self.mox.StubOutWithMock(utils, 'var_log_messages_before_reboot')
+ self.mox.StubOutWithMock(sys_utils, 'GetVarLogMessagesBeforeReboot')
def tearDown(self):
self.mox.UnsetStubs()
@@ -54,14 +54,15 @@
self.goofy_rpc.GetVarLogMessages(max_length=(len(data) + 5)))
def testGetVarLogMessagesBeforeReboot(self):
- utils.var_log_messages_before_reboot(lines=2, max_length=100).AndReturn(
+ sys_utils.GetVarLogMessagesBeforeReboot(lines=2, max_length=100).AndReturn(
['foo\xFF', 'bar'])
self.mox.ReplayAll()
self.assertEquals(u'foo\ufffd\nbar\n',
self.goofy_rpc.GetVarLogMessagesBeforeReboot(2, 100))
def testGetVarLogMessagesBeforeRebootEmpty(self):
- utils.var_log_messages_before_reboot(lines=2, max_length=100).AndReturn([])
+ sys_utils.GetVarLogMessagesBeforeReboot(lines=2,
+ max_length=100).AndReturn([])
self.mox.ReplayAll()
self.assertEquals(None,
self.goofy_rpc.GetVarLogMessagesBeforeReboot(2, 100))
diff --git a/py/test/utils/__init__.py b/py/test/utils/__init__.py
index a7f3301..e69de29 100644
--- a/py/test/utils/__init__.py
+++ b/py/test/utils/__init__.py
@@ -1,56 +0,0 @@
-#!/usr/bin/python -u
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2012 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.
-
-"""Utility functions that are useful to factory tests."""
-
-import os
-import re
-
-import factory_common # pylint: disable=W0611
-
-
-def var_log_messages_before_reboot(lines=100,
- max_length=5 * 1024 * 1024,
- path='/var/log/messages'):
- """Returns the last few lines in /var/log/messages before the current boot.
-
- Args:
- lines: number of lines to return.
- max_length: maximum amount of data at end of file to read.
- path: path to /var/log/messages.
-
- Returns:
- An array of lines. Empty if the marker indicating kernel boot
- could not be found.
- """
- offset = max(0, os.path.getsize(path) - max_length)
- with open(path) as f:
- f.seek(offset)
- data = f.read()
-
- # Find the last element matching the RE signaling kernel start.
- matches = list(re.finditer(
- r'^(\S+)\s.*kernel:\s+\[\s+0\.\d+\] Linux version', data, re.MULTILINE))
- if not matches:
- return []
-
- match = matches[-1]
- tail_lines = data[:match.start()].split('\n')
- tail_lines.pop() # Remove incomplete line at end
-
- # Skip some common lines that may have been written before the Linux
- # version.
- while tail_lines and any(
- re.search(x, tail_lines[-1])
- for x in [r'0\.000000\]',
- r'rsyslogd.+\(re\)start',
- r'/proc/kmsg started']):
- tail_lines.pop()
-
- # Done! Return the last few lines.
- return tail_lines[-lines:] + [
- '<after reboot, kernel came up at %s>' % match.group(1)]
diff --git a/py/test/utils/utils_unittest.py b/py/test/utils/utils_unittest.py
deleted file mode 100755
index 31159b1..0000000
--- a/py/test/utils/utils_unittest.py
+++ /dev/null
@@ -1,66 +0,0 @@
-#!/usr/bin/python
-#
-# Copyright (c) 2012 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 tempfile
-import unittest
-
-import factory_common # pylint: disable=W0611
-from cros.factory.test import utils
-from cros.factory.utils import type_utils
-
-
-EARLIER_VAR_LOG_MESSAGES = '''19:26:17 kernel: That's all, folks.
-19:26:56 kernel: [ 0.000000] Initializing cgroup subsys cpuset
-19:26:56 kernel: [ 0.000000] Initializing cgroup subsys cpu
-19:26:56 kernel: [ 0.000000] Linux version blahblahblah
-'''
-
-VAR_LOG_MESSAGES = '''19:00:00 kernel: 7 p.m. and all's well.
-19:27:17 kernel: That's all, folks.
-19:27:17 kernel: Kernel logging (proc) stopped.
-19:27:56 kernel: imklog 4.6.2, log source = /proc/kmsg started.
-19:27:56 rsyslogd: [origin software="rsyslogd" blahblahblah] (re)start
-19:27:56 kernel: [ 0.000000] Initializing cgroup subsys cpuset
-19:27:56 kernel: [ 0.000000] Initializing cgroup subsys cpu
-19:27:56 kernel: [ 0.000000] Linux version blahblahblah
-19:27:56 kernel: [ 0.000000] Command line: blahblahblah
-'''
-
-
-class VarLogMessagesTest(unittest.TestCase):
-
- def _GetMessages(self, data, lines):
- with tempfile.NamedTemporaryFile() as f:
- path = f.name
- f.write(data)
- f.flush()
-
- return utils.var_log_messages_before_reboot(path=path, lines=lines)
-
- def runTest(self):
- self.assertEquals([
- "19:27:17 kernel: That's all, folks.",
- "19:27:17 kernel: Kernel logging (proc) stopped.",
- "<after reboot, kernel came up at 19:27:56>",
- ], self._GetMessages(VAR_LOG_MESSAGES, 2))
- self.assertEquals([
- "19:27:17 kernel: Kernel logging (proc) stopped.",
- "<after reboot, kernel came up at 19:27:56>",
- ], self._GetMessages(VAR_LOG_MESSAGES, 1))
- self.assertEquals([
- "19:00:00 kernel: 7 p.m. and all's well.",
- "19:27:17 kernel: That's all, folks.",
- "19:27:17 kernel: Kernel logging (proc) stopped.",
- "<after reboot, kernel came up at 19:27:56>",
- ], self._GetMessages(VAR_LOG_MESSAGES, 100))
- self.assertEquals([
- "19:26:17 kernel: That's all, folks.",
- "<after reboot, kernel came up at 19:26:56>",
- ], self._GetMessages(EARLIER_VAR_LOG_MESSAGES, 1))
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/py/utils/sys_utils.py b/py/utils/sys_utils.py
index 4fc314c..c4a77b5 100644
--- a/py/utils/sys_utils.py
+++ b/py/utils/sys_utils.py
@@ -315,3 +315,46 @@
# this in the future to find a more deterministic way to probe freon board.
return (dut.Call('[ -e /sbin/frecon ]') == 0 if dut else
os.path.exists('/sbin/frecon'))
+
+
+def GetVarLogMessagesBeforeReboot(lines=100,
+ max_length=5 * 1024 * 1024,
+ path='/var/log/messages'):
+ """Returns the last few lines in /var/log/messages before the current boot.
+
+ Args:
+ lines: number of lines to return.
+ max_length: maximum amount of data at end of file to read.
+ path: path to /var/log/messages.
+
+ Returns:
+ An array of lines. Empty if the marker indicating kernel boot
+ could not be found.
+ """
+ offset = max(0, os.path.getsize(path) - max_length)
+ with open(path) as f:
+ f.seek(offset)
+ data = f.read()
+
+ # Find the last element matching the RE signaling kernel start.
+ matches = list(re.finditer(
+ r'^(\S+)\s.*kernel:\s+\[\s+0\.\d+\] Linux version', data, re.MULTILINE))
+ if not matches:
+ return []
+
+ match = matches[-1]
+ tail_lines = data[:match.start()].split('\n')
+ tail_lines.pop() # Remove incomplete line at end
+
+ # Skip some common lines that may have been written before the Linux
+ # version.
+ while tail_lines and any(
+ re.search(x, tail_lines[-1])
+ for x in [r'0\.000000\]',
+ r'rsyslogd.+\(re\)start',
+ r'/proc/kmsg started']):
+ tail_lines.pop()
+
+ # Done! Return the last few lines.
+ return tail_lines[-lines:] + [
+ '<after reboot, kernel came up at %s>' % match.group(1)]
diff --git a/py/utils/sys_utils_unittest.py b/py/utils/sys_utils_unittest.py
index 100c211..8f940df 100755
--- a/py/utils/sys_utils_unittest.py
+++ b/py/utils/sys_utils_unittest.py
@@ -9,6 +9,7 @@
import mox
import os
import tempfile
+import textwrap
import unittest
import factory_common # pylint: disable=W0611
@@ -264,5 +265,56 @@
sys_utils.MountDeviceAndReadFile('no_device', self.file_name)
+class VarLogMessagesTest(unittest.TestCase):
+
+ EARLIER_VAR_LOG_MESSAGES = textwrap.dedent('''
+ 19:26:17 kernel: That's all, folks.
+ 19:26:56 kernel: [ 0.000000] Initializing cgroup subsys cpuset
+ 19:26:56 kernel: [ 0.000000] Initializing cgroup subsys cpu
+ 19:26:56 kernel: [ 0.000000] Linux version blahblahblah
+ ''')
+
+ VAR_LOG_MESSAGES = textwrap.dedent('''\
+ 19:00:00 kernel: 7 p.m. and all's well.
+ 19:27:17 kernel: That's all, folks.
+ 19:27:17 kernel: Kernel logging (proc) stopped.
+ 19:27:56 kernel: imklog 4.6.2, log source = /proc/kmsg started.
+ 19:27:56 rsyslogd: [origin software="rsyslogd" blahblahblah] (re)start
+ 19:27:56 kernel: [ 0.000000] Initializing cgroup subsys cpuset
+ 19:27:56 kernel: [ 0.000000] Initializing cgroup subsys cpu
+ 19:27:56 kernel: [ 0.000000] Linux version blahblahblah
+ 19:27:56 kernel: [ 0.000000] Command line: blahblahblah
+ ''')
+
+ def _GetMessages(self, data, lines):
+ with tempfile.NamedTemporaryFile() as f:
+ path = f.name
+ f.write(data)
+ f.flush()
+
+ return sys_utils.GetVarLogMessagesBeforeReboot(path=path, lines=lines)
+
+ def runTest(self):
+ self.assertEquals([
+ "19:27:17 kernel: That's all, folks.",
+ "19:27:17 kernel: Kernel logging (proc) stopped.",
+ "<after reboot, kernel came up at 19:27:56>",
+ ], self._GetMessages(self.VAR_LOG_MESSAGES, 2))
+ self.assertEquals([
+ "19:27:17 kernel: Kernel logging (proc) stopped.",
+ "<after reboot, kernel came up at 19:27:56>",
+ ], self._GetMessages(self.VAR_LOG_MESSAGES, 1))
+ self.assertEquals([
+ "19:00:00 kernel: 7 p.m. and all's well.",
+ "19:27:17 kernel: That's all, folks.",
+ "19:27:17 kernel: Kernel logging (proc) stopped.",
+ "<after reboot, kernel came up at 19:27:56>",
+ ], self._GetMessages(self.VAR_LOG_MESSAGES, 100))
+ self.assertEquals([
+ "19:26:17 kernel: That's all, folks.",
+ "<after reboot, kernel came up at 19:26:56>",
+ ], self._GetMessages(self.EARLIER_VAR_LOG_MESSAGES, 1))
+
+
if __name__ == '__main__':
unittest.main()