[master 9/21] Make a logger for test data that we can automatically use later
vathpela
installerbot-noreply at redhat.com
Wed Oct 28 15:53:28 UTC 2015
From: Peter Jones <pjones at redhat.com>
This adds another logger that we can use to write data to about what we
find on the filesystem, in such a way that we can use that data to
recreate test cases later, automatically, from the log.
This also adds that data to all the EDD code.
Signed-off-by: Peter Jones <pjones at redhat.com>
---
blivet/__init__.py | 1 +
blivet/devicelibs/edd.py | 79 +++++++++++++++++++++++++++------------
blivet/util.py | 31 ++++++++++++---
tests/devicelibs_test/edd_test.py | 28 ++++++++++----
4 files changed, 102 insertions(+), 37 deletions(-)
diff --git a/blivet/__init__.py b/blivet/__init__.py
index 37094c2..e6fd42c 100644
--- a/blivet/__init__.py
+++ b/blivet/__init__.py
@@ -59,6 +59,7 @@ def cb(self, exn):
import logging
log = logging.getLogger("blivet")
program_log = logging.getLogger("program")
+testdata_log = logging.getLogger("testdata")
# Tell the warnings module not to ignore DeprecationWarning, which it does by
# default since python-2.7.
diff --git a/blivet/devicelibs/edd.py b/blivet/devicelibs/edd.py
index 0a83aef..1e29bd5 100644
--- a/blivet/devicelibs/edd.py
+++ b/blivet/devicelibs/edd.py
@@ -33,6 +33,8 @@
from .. import util
log = logging.getLogger("blivet")
+testdata_log = logging.getLogger("testdata")
+testdata_log.setLevel(logging.DEBUG)
re_host_bus_pci = re.compile(r'^(PCIX|PCI|XPRS|HTPT)\s*(\S*)\s*channel: (\S*)\s*$')
re_interface_atapi = re.compile(r'^ATAPI\s*device: (\S*)\s*lun: (\S*)\s*$')
@@ -322,8 +324,12 @@ def __init__(self, edd_entry):
def devname_from_ata_pci_dev(self):
pattern = '%s/sys/block/*' % (fsroot,)
+ testdata_log.debug("sysfs glob: %s", pattern[len(fsroot):])
for path in glob.iglob(pattern):
+ testdata_log.debug("sysfs glob match: %s", path[len(fsroot):])
link = os.readlink(path)
+ testdata_log.debug("sysfs link: \"%s\" -> \"%s\"",
+ path[len(fsroot):], link)
# just add /sys/block/ at the beginning so it's always valid
# paths in the filesystem...
components = ['%s/sys/block' % (fsroot,)] + link.split('/')
@@ -382,28 +388,32 @@ def devname_from_ata_pci_dev(self):
pmp_glob = os.path.join(*pmp_glob)
dev_glob = fn + ['dev%d.*' % (ata_port_idx,)]
dev_glob = os.path.join(*dev_glob)
- atapaths = tuple(glob.iglob(pmp_glob)) + tuple(glob.iglob(dev_glob))
- for atapath in atapaths:
- match = expmatcher.match(atapath)
- if match is None:
- continue
-
- # so at this point it's devW.X.Y or devW.Z as such:
- # dev_set_name(dev, "dev%d.%d", ap->print_id,ata_dev->devno);
- # dev_set_name(dev, "dev%d.%d.0", ap->print_id, link->pmp);
- # we care about print_id and pmp for matching and the ATA
- # channel if applicable. We already checked print_id above.
- if match.group(3) is None:
- channel = int(match.group(1))
- if (self.edd.channel == 255 and channel == 0) or \
- (self.edd.channel == channel):
- self.edd.sysfslink = link
- return path[len(fsroot):].split('/')[-1]
- else:
- pmp = int(match.group(1))
- if self.edd.ata_pmp == pmp:
- self.edd.sysfslink = link
- return path[len(fsroot):].split('/')[-1]
+ for ataglob in [pmp_glob, dev_glob]:
+ testdata_log.debug("sysfs glob: %s", ataglob)
+ for atapath in glob.glob(ataglob):
+ testdata_log.debug("sysfs glob match: %s",
+ atapath[len(fsroot):])
+ match = expmatcher.match(atapath)
+ if match is None:
+ continue
+
+ # so at this point it's devW.X.Y or devW.Z as such:
+ # dev_set_name(dev, "dev%d.%d",
+ # ap->print_id,ata_dev->devno); dev_set_name(dev,
+ # "dev%d.%d.0", ap->print_id, link->pmp); we care about
+ # print_id and pmp for matching and the ATA channel if
+ # applicable. We already checked print_id above.
+ if match.group(3) is None:
+ channel = int(match.group(1))
+ if (self.edd.channel == 255 and channel == 0) or \
+ (self.edd.channel == channel):
+ self.edd.sysfslink = link
+ return path[len(fsroot):].split('/')[-1]
+ else:
+ pmp = int(match.group(1))
+ if self.edd.ata_pmp == pmp:
+ self.edd.sysfslink = link
+ return path[len(fsroot):].split('/')[-1]
return None
def devname_from_scsi_pci_dev(self):
@@ -423,7 +433,10 @@ def devname_from_scsi_pci_dev(self):
}
path = tmpl0 % args
pattern = tmpl1 % args
+ testdata_log.debug("sysfs glob: %s", pattern[len(fsroot):])
matching_paths = glob.glob(pattern)
+ for mp in matching_paths:
+ testdata_log.debug("sysfs glob match: %s", mp[len(fsroot):])
if os.path.isdir(path):
block_entries = os.listdir(path)
if len(block_entries) == 1:
@@ -456,7 +469,11 @@ def devname_from_virt_pci_dev(self):
'pci_dev' : self.edd.pci_dev
}
pattern = "%(fsroot)s/sys/devices/pci0000:00/0000:%(pci_dev)s/virtio*"
- matching_paths = glob.glob(pattern % args)
+ matching_paths = tuple(glob.glob(pattern % args))
+ testdata_log.debug("sysfs glob: %s",
+ (pattern % args)[len(fsroot):])
+ for mp in matching_paths:
+ testdata_log.debug("sysfs glob match: %s", mp[len(fsroot):])
if len(matching_paths) == 1 and os.path.exists(matching_paths[0]):
# Normal VirtIO devices just have the block link right there...
@@ -491,6 +508,13 @@ def devname_from_pci_dev(self):
name = self.devname_from_ata_pci_dev()
elif self.edd.type == "SCSI":
name = self.devname_from_scsi_pci_dev()
+ if self.edd.sysfslink:
+ path = "/".join([fsroot, "sys/block", self.edd.sysfslink,
+ "device"])
+ link = os.readlink(path)
+ testdata_log.debug("sysfs link: \"%s\" -> \"%s\"",
+ path[len(fsroot):], link)
+
return name
def match_via_mbrsigs(self, mbr_dict):
@@ -508,7 +532,9 @@ def collect_edd_data():
edd_data_dict = {}
exp = re.compile(r'.*/int13_dev(\d+)$')
globstr = os.path.join(fsroot, "sys/firmware/edd/int13_dev*")
+ testdata_log.debug("sysfs glob: %s", globstr[len(fsroot):])
for path in glob.glob(globstr):
+ testdata_log.debug("sysfs glob match: %s", path[len(fsroot):])
match = exp.match(path)
biosdev = int("0x%s" % (match.group(1),), base=16)
log.debug("edd: found device 0x%x at %s", biosdev, path[len(fsroot):])
@@ -530,9 +556,14 @@ def collect_mbrs(devices):
fd = util.eintr_retry_call(os.open, path, os.O_RDONLY)
# The signature is the unsigned integer at byte 440:
os.lseek(fd, 440, 0)
- mbrsig = struct.unpack('I', util.eintr_retry_call(os.read, fd, 4))
+ data = util.eintr_retry_call(os.read, fd, 4)
+ mbrsig = struct.unpack('I', data)
+ sdata = struct.unpack("BBBB", data)
+ sdata = "".join(["%02x" % (x,) for x in sdata])
util.eintr_ignore(os.close, fd)
+ testdata_log.debug("device %s data[440:443] = %s", path, sdata)
except OSError as e:
+ testdata_log.debug("device %s data[440:443] raised %s", path, e)
log.error("edd: could not read mbrsig from disk %s: %s",
dev.name, str(e))
continue
diff --git a/blivet/util.py b/blivet/util.py
index 302861e..6a5652f 100644
--- a/blivet/util.py
+++ b/blivet/util.py
@@ -26,6 +26,7 @@
import logging
log = logging.getLogger("blivet")
program_log = logging.getLogger("program")
+testdata_log = logging.getLogger("testdata")
from threading import Lock
# this will get set to anaconda's program_log_lock in enable_installer_mode
@@ -211,7 +212,12 @@ def get_sysfs_attr(path, attr):
log.warning("%s is not a valid attribute", attr)
return None
- return open(attribute, "r").read().strip()
+ f = open(attribute, "r")
+ data = f.read()
+ f.close()
+ sdata = "".join(["%02x" % (ord(x),) for x in data])
+ testdata_log.debug("sysfs attr %s: %s", attribute, sdata)
+ return data.strip()
def get_sysfs_path_by_name(dev_node, class_name="block"):
""" Return sysfs path for a given device.
@@ -510,17 +516,25 @@ def compare(first, second):
##
## Convenience functions for examples and tests
##
-def set_up_logging(log_file='/tmp/blivet.log'):
+def set_up_logging(log_dir="/tmp", log_prefix="blivet"):
""" Configure the blivet logger to write out a log file.
:keyword str log_file: path to the log file (default: /tmp/blivet.log)
"""
+
log.setLevel(logging.DEBUG)
program_log.setLevel(logging.DEBUG)
- handler = logging.FileHandler(log_file)
- handler.setLevel(logging.DEBUG)
- formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
- handler.setFormatter(formatter)
+
+ def make_handler(path, prefix, level):
+ log_file = "%s/%s.log" % (path, log)
+ log_file = os.path.realpath(log_file)
+ handler = logging.FileHandler(log_file)
+ handler.setLevel(level)
+ formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
+ handler.setFormatter(formatter)
+ return handler
+
+ handler = make_handler(log_dir, log_prefix, logging.DEBUG)
log.addHandler(handler)
program_log.addHandler(handler)
@@ -530,6 +544,11 @@ def set_up_logging(log_file='/tmp/blivet.log'):
log.info("sys.argv = %s", sys.argv)
+ prefix = "%s-testdata" % (log_prefix,)
+ handler = make_handler(log_dir, prefix, logging.DEBUG)
+ testdata_log.setLevel(logging.DEBUG)
+ testdata_log.addHandler(handler)
+
def create_sparse_tempfile(name, size):
""" Create a temporary sparse file.
diff --git a/tests/devicelibs_test/edd_test.py b/tests/devicelibs_test/edd_test.py
index 3be3795..adf0f76 100644
--- a/tests/devicelibs_test/edd_test.py
+++ b/tests/devicelibs_test/edd_test.py
@@ -40,17 +40,26 @@ class EddTestCase(unittest.TestCase):
def setUp(self):
super(EddTestCase, self).setUp()
if 'WORKSPACE' in os.environ.keys():
- ws = os.path.join(os.environ['WORKSPACE'], "blivet-edd.log")
+ ws = os.environ['WORKSPACE']
else:
- ws = "/tmp/blivet-edd.log"
- self.handler = logging.FileHandler(ws)
- self.handler.setLevel(logging.DEBUG)
+ ws = "/tmp"
+ self.log_handler = logging.FileHandler("%s/%s" % (ws, "blivet-edd.log"))
+ self.log_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
- self.handler.setFormatter(formatter)
- edd.log.addHandler(self.handler)
+ self.log_handler.setFormatter(formatter)
+ edd.log.addHandler(self.log_handler)
+
+ self.td_log_handler = logging.FileHandler("%s/%s" %
+ (ws, "blivet-edd-testdata.log"))
+ self.td_log_handler.setLevel(logging.DEBUG)
+ formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
+ self.td_log_handler.setFormatter(formatter)
+ edd.testdata_log.addHandler(self.td_log_handler)
self._edd_logger = edd.log
self._edd_logger_level = edd.log.level
+ self._edd_testdata_logger = edd.testdata_log
+ self._edd_testdata_logger_level = edd.testdata_log.level
edd.log.setLevel(logging.DEBUG)
newlog = mock.MagicMock(name='log')
newlog.debug = mock.Mock(name='debug',
@@ -66,7 +75,8 @@ def setUp(self):
def tearDown(self):
edd.log = self._edd_logger
edd.log.setLevel(self._edd_logger_level)
- edd.log.removeHandler(self.handler)
+ edd.log.removeHandler(self.log_handler)
+ edd.testdata_log.removeHandler(self.td_log_handler)
super(EddTestCase, self).tearDown()
def checkLogs(self, debugs=None, infos=None, warnings=None, errors=None):
@@ -103,6 +113,8 @@ def debug(self, *args):
def test_collect_edd_data_sata_usb(self):
# test with sata sda, usb sdb
self._set_fs_root(edd, "sata_usb")
+ self._edd_logger.debug("starting test %s", self._testMethodName)
+ edd.testdata_log.debug("starting test %s", self._testMethodName)
fakeedd = {
0x80: FakeEddEntry(version="0x30", mbr_sig="0x00000000",
sectors=312581808, host_bus="PCI", type="SATA",
@@ -133,6 +145,8 @@ def test_collect_edd_data_sata_usb(self):
def test_get_edd_dict_sata_usb(self):
# test with sata sda, usb sdb
self._set_fs_root(edd, "sata_usb")
+ self._edd_logger.debug("starting test %s", self._testMethodName)
+ edd.testdata_log.debug("starting test %s", self._testMethodName)
devices=(FakeDevice("sda"),
FakeDevice("sdb"),
)
--
To view this commit on github, visit https://github.com/rhinstaller/blivet/commit/cf7f99fb3cf4ee8241f92eb0ef0a096c778ed637
More information about the anaconda-patches
mailing list