> This branch adds some sanity checks for HWDB submission containing udev data > (as sent by the HWDB client in Karmic), and it defines a new class UdevDevice, > representing a device described by udev data. (The class is quite rudimentary > at present.) Wow - I learned lots of great stuff through this branch - sorry it took me so long to review! After we found a better way of creating test versions of some of your classes (thanks Bjorn) I've only got comments and some test suggestions left. Great work! > === modified file 'lib/canonical/launchpad/scripts/hwdbsubmissions.py' > --- lib/canonical/launchpad/scripts/hwdbsubmissions.py 2009-10-06 08:37:11 +0000 > +++ lib/canonical/launchpad/scripts/hwdbsubmissions.py 2009-10-08 09:38:16 +0000 > @@ -1141,6 +1141,75 @@ > return False > return True > > + PCI_PROPERTIES = set( > + ('PCI_CLASS', 'PCI_ID', 'PCI_SUBSYS_ID', 'PCI_SLOT_NAME')) > + pci_class_re = re.compile('^[0-9a-f]{1,6}$', re.I) > + pci_id_re = re.compile('^[0-9a-f]{4}:[0-9a-f]{4}$', re.I) > + > + def checkUdevPciProperties(self, udev_data): > + """Validation of udev PCI devices. > + > + Each PCI device must have the properties PCI_CLASS, PCI_ID, > + PCI_SUBSYS_ID, PCI_SLOT_NAME. Non-PCI devices must not have > + them. > + > + The value of PCI class must be a 24 bit integer in > + hexadecimal representation. > + > + The values of PCI_ID and PCI_SUBSYS_ID must be two 16 bit > + integers, separated by a ':'. Barry would say: could you please use the :param: etc. like you have below for UDevDevice. > + """ > + for device in udev_data: > + properties = device['E'] > + property_names = set(properties.keys()) I think this method could do with a few comments in general... maybe a lot of it is completely obvious to you, but I have to spend time interpreting the code, for example,... > + existing_pci_properties = property_names.intersection( > + self.PCI_PROPERTIES) > + subsystem = device['E'].get('SUBSYSTEM') > + if subsystem is None: > + self._logError( > + 'udev device without SUBSYSTEM property found.', > + self.submission_key) > + return False > + if subsystem == 'pci': # Check whether any of the standard pci properties were missing. > + if existing_pci_properties != self.PCI_PROPERTIES: > + missing_properties = self.PCI_PROPERTIES.difference( > + existing_pci_properties) > + > + self._logError( > + 'PCI udev device without required PCI properties: ' > + '%r %r' > + % (missing_properties, device['P']), > + self.submission_key) > + return False # Ensure that the pci class and ids for this device are valid. > + if self.pci_class_re.search(properties['PCI_CLASS']) is None: > + self._logError( > + 'Invalid udev PCI class: %r %r' > + % (properties['PCI_CLASS'], device['P']), > + self.submission_key) > + return False > + for pci_id in (properties['PCI_ID'], > + properties['PCI_SUBSYS_ID']): > + if self.pci_id_re.search(pci_id) is None: > + self._logError( > + 'Invalid udev PCI device ID: %r %r' > + % (pci_id, device['P']), > + self.submission_key) > + return False > + else: > + if len(existing_pci_properties) > 0: It might be a lack of background on my part, but is this check necessary? I mean, if checkUdevPciProperties() finds a non-pci device, shouldn't we just say so rather? Ah, perhaps (looking at the above logic) it's fine if a non PCI device is found in the udev_data, it just shouldn't have any pci properties. Right. Perhaps a comment above the previous 'if' would have saved me wondering, something along the lines of: # It's fine if we find a non-pci device while checking the udev_data, # but we should not find a non-pci device with pci properties. > + self._logError( > + 'Non-PCI udev device with PCI properties: %r %r' > + % (existing_pci_properties, device['P']), > + self.submission_key) > + return False > + return True > + > + def checkConsistentUdevDeviceData(self, udev_data): > + """Consistency checks for udev data.""" > + if not self.checkUdevDictsHavePathKey(udev_data): > + return False > + return self.checkUdevPciProperties(udev_data) > + > def checkConsistency(self, parsed_data): > """Run consistency checks on the submitted data. > > @@ -1148,10 +1217,10 @@ > :param: parsed_data: parsed submission data, as returned by > parseSubmission > """ > - if 'udev' in parsed_data['hardware']: > - if not self.checkUdevDictsHavePathKey( > - parsed_data['hardware']['udev']): > - return False > + if ('udev' in parsed_data['hardware'] > + and not self.checkConsistentUdevDeviceData( > + parsed_data['hardware']['udev'])): > + return False > duplicate_ids = self.findDuplicateIDs(parsed_data) > if duplicate_ids: > self._logError('Duplicate IDs found: %s' % duplicate_ids, > @@ -1174,7 +1243,8 @@ > self._logError(value, self.submission_key) > return False > > - circular = self.checkHALDevicesParentChildConsistency(udi_children) > + circular = self.checkHALDevicesParentChildConsistency( > + udi_children) > if circular: > self._logError('Found HAL devices with circular parent/child ' > 'relationship: %s' % circular, > @@ -2146,6 +2216,54 @@ > return self.getVendorOrProductID('product') > > > +class UdevDevice(BaseDevice): > + """The representation of a udev device node.""" > + > + def __init__(self, udev_data, sysfs_data, parser): > + """HALDevice constructor. > + > + :param udevdata: The udev data for this device > + :param sysfs_data: sysfs data for this device. > + :param parser: The parser processing a submission. > + :type parser: SubmissionParser > + """ > + super(UdevDevice, self).__init__(parser) > + self.udev = udev_data > + self.sysfs = sysfs_data > + > + @property > + def device_id(self): > + """See `BaseDevice`.""" > + return self.udev['P'] > + > + @property > + def pci_class_info(self): > + """Parse the udev property PCI_SUBSYS_ID. > + > + :return: (PCI class, PCI sub-class, version) for a PCI device > + or (None, None, None) for other devices. > + """ > + if self.udev['E'].get('SUBSYSTEM') == 'pci': Would it be worth adding a self.is_pci property? > + # SubmissionParser.checkConsistentUdevDeviceData() ensures > + # that PCI_CLASS is a 24 bit integer in hexadecimal > + # representation. Great - but what's going on below, I mean I can read the right shifts and masks, and I assume this is the format required for pci_class_info? > + class_info = int(self.udev['E']['PCI_CLASS'], 16) > + return (class_info >> 16, (class_info >> 8) & 0xFF, > + class_info & 0xFF) > + else: > + return (None, None, None) > + > + @property > + def pci_class(self): > + """See `BaseDevice`.""" > + return self.pci_class_info[0] > + > + @property > + def pci_subclass(self): > + """See `BaseDevice`.""" > + return self.pci_class_info[1] > + > + > class ProcessingLoop(object): > """An `ITunableLoop` for processing HWDB submissions.""" > > > === modified file 'lib/canonical/launchpad/scripts/tests/test_hwdb_submission_parser.py' > --- lib/canonical/launchpad/scripts/tests/test_hwdb_submission_parser.py 2009-10-06 09:31:32 +0000 > +++ lib/canonical/launchpad/scripts/tests/test_hwdb_submission_parser.py 2009-10-08 09:38:16 +0000 > @@ -3,6 +3,7 @@ > > """Tests of the HWDB submissions parser.""" > > +from copy import deepcopy > from cStringIO import StringIO > try: > import xml.etree.cElementTree as etree > @@ -1773,6 +1774,145 @@ > self.assertErrorMessage( > parser.submission_key, 'udev node found without a "P" key') > > + udev_root_device = { > + 'P': '/devices/LNXSYSTM:00', > + 'E': {'SUBSYSTEM': 'acpi'}, > + } > + udev_pci_device = { > + 'P': '/devices/pci0000:00/0000:00:1f.2', > + 'E': { > + 'SUBSYSTEM': 'pci', > + 'PCI_CLASS': '10601', > + 'PCI_ID': '8086:27C5', > + 'PCI_SUBSYS_ID': '10CF:1387', > + 'PCI_SLOT_NAME': '0000:00:1f.2', > + } > + } > + > + def testCheckUdevPciProperties(self): I'm assuming camelCase is for consistency with the rest of the file. > + """Test of SubmmissionParser.checkUdevPciProperties().""" > + # udev PCI devices must have the properties PCI_CLASS, PCI_ID, > + # PCI_SUBSYS_ID, PCI_SLOT_NAME; other devices must not have > + # these properties. > + parser = SubmissionParser() > + self.assertTrue(parser.checkUdevPciProperties( > + [self.udev_root_device, self.udev_pci_device])) > + Sheesh - that's one long single test there! Could it be split up into a bunch of unit-tests? Like, testCheckUdevPciProperties_valid_device, testCheckUdevPciProperties_bad_device, etc etc. as outlined at: https://dev.launchpad.net/TestsStyleGuide#Python%20Test%20Cases > + bad_root_device = deepcopy(self.udev_root_device) If you defined the above class properties (udev_root_device and udev_pci_device instead as instance properties within setUp, you could then re-use and modify them in each test without having to deep_copy etc. > + bad_root_device['E']['PCI_SLOT_NAME'] = '0000:00:1f.2' > + parser = SubmissionParser(self.log) > + parser.submission_key = 'invalid non-PCI device' > + self.assertFalse(parser.checkUdevPciProperties( > + [bad_root_device, self.udev_pci_device])) > + self.assertErrorMessage( > + parser.submission_key, > + "Non-PCI udev device with PCI properties: set(['PCI_SLOT_NAME']) " > + "'/devices/LNXSYSTM:00'") next test. > + > + bad_pci_device = deepcopy(self.udev_pci_device) > + del bad_pci_device['E']['PCI_CLASS'] > + parser = SubmissionParser(self.log) > + parser.submission_key = 'invalid PCI device' > + self.assertFalse(parser.checkUdevPciProperties( > + [self.udev_root_device, bad_pci_device])) > + self.assertErrorMessage( > + parser.submission_key, > + "PCI udev device without required PCI properties: " > + "set(['PCI_CLASS']) '/devices/pci0000:00/0000:00:1f.2'") next test. > + > + bad_pci_device = deepcopy(self.udev_pci_device) > + bad_pci_device['E']['PCI_CLASS'] = 'not-an-integer' > + parser = SubmissionParser(self.log) > + parser.submission_key = 'invalid PCI class value' > + self.assertFalse(parser.checkUdevPciProperties( > + [self.udev_root_device, bad_pci_device])) > + self.assertErrorMessage( > + parser.submission_key, > + "Invalid udev PCI class: 'not-an-integer' " > + "'/devices/pci0000:00/0000:00:1f.2'") next test. > + > + bad_pci_device = deepcopy(self.udev_pci_device) > + bad_pci_device['E']['PCI_CLASS'] = '1234567' > + parser = SubmissionParser(self.log) > + parser.submission_key = 'too large PCI class value' > + self.assertFalse(parser.checkUdevPciProperties( > + [self.udev_root_device, bad_pci_device])) > + self.assertErrorMessage( > + parser.submission_key, > + "Invalid udev PCI class: '1234567' " > + "'/devices/pci0000:00/0000:00:1f.2'") next test. > + > + bad_pci_device = deepcopy(self.udev_pci_device) > + bad_pci_device['E']['PCI_ID'] = 'not-an-id' > + parser = SubmissionParser(self.log) > + parser.submission_key = 'invalid PCI ID' > + self.assertFalse(parser.checkUdevPciProperties( > + [self.udev_root_device, bad_pci_device])) > + self.assertErrorMessage( > + parser.submission_key, > + "Invalid udev PCI device ID: 'not-an-id' " > + "'/devices/pci0000:00/0000:00:1f.2'") > + next test. > + bad_pci_device = deepcopy(self.udev_pci_device) > + bad_pci_device['E']['PCI_SUBSYS_ID'] = 'not-a-subsystem-id' > + parser = SubmissionParser(self.log) > + parser.submission_key = 'invalid PCI subsystem ID' > + self.assertFalse(parser.checkUdevPciProperties( > + [self.udev_root_device, bad_pci_device])) > + self.assertErrorMessage( > + parser.submission_key, > + "Invalid udev PCI device ID: 'not-a-subsystem-id' " > + "'/devices/pci0000:00/0000:00:1f.2'") > + > + def _setupUdevConsistencyCheckTests(self): > + """Prepare and return a SubmissionParser instance. OK, a few things here. First, this looks like the kind of thing that a setUp() method should be doing, so perhaps the following tests could be in a separate class with a setUp() that sets self.parser? Second, as we chatted about on irc, I'm all for using mocks/doubles etc., but not by manually patching things like this. So I'd be keen to see this changed to either (1) simply setup the required data - but you said this will get tedious as this class grows, so (2) using mocker to do what you've done below - you could potentially just patch the class in your setup method - but I'm not sure - I haven't used mocker itself. From: http://labix.org/mocker "Mocking via temporary patching of existent classes and instances." Ah, then BjornT chipped in with: noodles775, adeuring: i don't see a reason for using mocker to do that. can't you simply subclass SubmissionParser and override the methods in the new class? * barry has quit (Read error: 104 (Connection reset by peer)) BjornT: nice idea, thanks! BjornT: yep - that sounds like a good option. > + > + Replace udev related methods with dummies that just return True. > + """ > + def checkUdevDictsHavePathKey(self, udev_data): > + return True > + > + def checkUdevPciProperties(self, udev_data): > + return True > + > + parser = SubmissionParser() > + parser.checkUdevDictsHavePathKey = ( > + lambda data: checkUdevDictsHavePathKey(parser, data)) > + parser.checkUdevPciProperties = ( > + lambda data: checkUdevPciProperties(parser, data)) > + return parser > + > + def testCheckConsistentUdevDeviceData(self): > + """Test of SubmissionParser.checkConsistentUdevDeviceData(),""" > + parser = self._setupUdevConsistencyCheckTests() > + self.assertTrue(parser.checkConsistentUdevDeviceData(None)) > + > + def testCheckConsistentUdevDeviceDataInvalidPathData(self): This should be either: testCheckConsistentUdevDeviceData_invalid_path_data() (consistent), or test_checkConsistentUdevDeviceData_invalid_path_data() as per https://dev.launchpad.net/TestsStyleGuide#Python%20Test%20Cases > + """Test of SubmissionParser.checkConsistentUdevDeviceData(), > + > + Detection of invalid path data lets the check fail. > + """ > + def checkUdevDictsHavePathKey(self, udev_data): > + return False > + > + parser = self._setupUdevConsistencyCheckTests() > + parser.checkUdevDictsHavePathKey = ( > + lambda data: checkUdevDictsHavePathKey(parser, data)) Same thing as above - using BjornT's method, you could have a subclass with a property that you can just set in your test, and the method will return that property? That way you could use it for above and below. > + self.assertFalse(parser.checkConsistentUdevDeviceData(None)) > + > + def testCheckConsistentUdevDeviceDataInvalidPciData(self): Again - style-guide for tests ^^^ > + """Test of SubmissionParser.checkConsistentUdevDeviceData(), > + > + Detection of invalid PCI data lets the check fail. > + """ > + def checkUdevPciProperties(self, udev_data): > + return False > + > + parser = self._setupUdevConsistencyCheckTests() > + parser.checkUdevPciProperties = ( > + lambda data: checkUdevPciProperties(parser, data)) > + self.assertFalse(parser.checkConsistentUdevDeviceData(None)) > + > def _setupConsistencyCheckParser(self): > """Prepare and return a SubmissionParser instance. > > @@ -1799,6 +1939,9 @@ > test.assertTrue(isinstance(self, SubmissionParser)) > return [] > > + def checkConsistentUdevDeviceData(self, udev_data): > + return True > + > parser = SubmissionParser(self.log) > parser.findDuplicateIDs = ( > lambda parsed_data: findDuplicateIDs(parser, parsed_data)) > @@ -1811,6 +1954,9 @@ > parser.checkHALDevicesParentChildConsistency = ( > lambda udi_children: checkHALDevicesParentChildConsistency( > parser, udi_children)) > + parser.checkConsistentUdevDeviceData = ( > + lambda udev_data: checkConsistentUdevDeviceData( > + parser, udev_data)) > return parser Another one? OK, you've already got some test classes in place for this, so I'll stop mentioning it :) > > def assertErrorMessage(self, submission_key, log_message): > @@ -1884,8 +2030,13 @@ > > def testConsistencyCheckInvalidUdevData(self): Previously mentioned test styles ^^^ > """Test of SubmissionParser.checkConsistency.""" > + def checkConsistentUdevDeviceData(self, udev_data): > + return False > + > parser = self._setupConsistencyCheckParser() > - parser.submission_key = 'Consistency check with invalid udev data' > + parser.checkConsistentUdevDeviceData = ( > + lambda udev_data: checkConsistentUdevDeviceData( > + parser, udev_data)) > self.assertFalse(parser.checkConsistency( > { > 'hardware': { > @@ -1894,8 +2045,6 @@ > } > } > )) > - self.assertErrorMessage( > - parser.submission_key, 'udev node found without a "P" key') > > def testConsistencyCheckWithDuplicateIDs(self): > """SubmissionParser.checkConsistency detects duplicate IDs.""" > > === modified file 'lib/canonical/launchpad/scripts/tests/test_hwdb_submission_processing.py' > --- lib/canonical/launchpad/scripts/tests/test_hwdb_submission_processing.py 2009-10-01 14:48:33 +0000 > +++ lib/canonical/launchpad/scripts/tests/test_hwdb_submission_processing.py 2009-10-08 09:38:16 +0000 > @@ -27,8 +27,8 @@ > from canonical.launchpad.scripts.hwdbsubmissions import ( > HALDevice, PCI_CLASS_BRIDGE, PCI_CLASS_SERIALBUS_CONTROLLER, > PCI_CLASS_STORAGE, PCI_SUBCLASS_BRIDGE_CARDBUS, PCI_SUBCLASS_BRIDGE_PCI, > - PCI_SUBCLASS_SERIALBUS_USB, PCI_SUBCLASS_STORAGE_SATA, SubmissionParser, > - process_pending_submissions) > + PCI_SUBCLASS_SERIALBUS_USB, PCI_SUBCLASS_STORAGE_SATA, > + process_pending_submissions, SubmissionParser, UdevDevice) > from canonical.launchpad.webapp.errorlog import ErrorReportingUtility > from canonical.testing import BaseLayer, LaunchpadZopelessLayer > > @@ -2591,6 +2591,76 @@ > 'property not treated as a real device.') > > > +class TestUdevDevice(TestCase): > + """Tests of class UdevDevice.""" > + > + layer = BaseLayer > + > + root_device = { > + 'P': '/devices/LNXSYSTM:00', > + 'E': { > + 'UDEV_LOG': '3', > + 'DEVPATH': '/devices/LNXSYSTM:00', > + 'MODALIAS': 'acpi:LNXSYSTM:', > + 'SUBSYSTEM': 'acpi', > + } > + } > + > + pci_device_data = { > + 'P': '/devices/pci0000:00/0000:00:1f.2', > + 'E': { > + 'PCI_CLASS': '10602', > + 'PCI_ID': '8086:27C5', > + 'PCI_SUBSYS_ID': '10CF:1387', > + 'PCI_SLOT_NAME': '0000:00:1f.2', > + 'SUBSYSTEM': 'pci', > + } > + } > + > + def test_device_id(self): > + """Test of UdevDevice.device_id.""" > + device = UdevDevice(self.pci_device_data, None, None) > + self.assertEqual( > + '/devices/pci0000:00/0000:00:1f.2', device.device_id, > + 'Unexpected value of UdevDevice.device_id.') > + > + def test_pci_class_info(self): > + """Test of UdevDevice.pci_class_info""" > + device = UdevDevice(self.pci_device_data, None, None) > + self.assertEqual( > + (1, 6, 2), device.pci_class_info, > + 'Invalid value of UdevDevice.pci_class_info for PCI device.') > + > + device = UdevDevice(self.root_device, None, None) > + self.assertEqual( > + (None, None, None), device.pci_class_info, > + 'Invalid value of UdevDevice.pci_class_info for Non-PCI device.') > + > + def test_pci_class(self): > + """Test of UdevDevice.pci_class""" > + device = UdevDevice(self.pci_device_data, None, None) > + self.assertEqual( > + 1, device.pci_class, > + 'Invalid value of UdevDevice.pci_class for PCI device.') > + > + device = UdevDevice(self.root_device, None, None) > + self.assertEqual( > + None, device.pci_class, > + 'Invalid value of UdevDevice.pci_class for Non-PCI device.') > + > + def test_pci_subclass(self): > + """Test of UdevDevice.pci_subclass""" > + device = UdevDevice(self.pci_device_data, None, None) > + self.assertEqual( > + 6, device.pci_subclass, > + 'Invalid value of UdevDevice.pci_class for PCI device.') > + > + device = UdevDevice(self.root_device, None, None) > + self.assertEqual( > + None, device.pci_class, > + 'Invalid value of UdevDevice.pci_class for Non-PCI device.') > + > + > class TestHWDBSubmissionTablePopulation(TestCaseHWDB): > """Tests of the HWDB popoluation with submitted data.""" > *phew* Great! Thanks Abel! -- Michael