Extract tar

hwinfo/pci/lspci.py

@@ -5,7 +5,7 @@
 class ParserException(Exception):
     pass
 
-LABEL_REGEX = r'[\w+\ \.\,\+\&\-\/\[\]\(\)]+'
+LABEL_REGEX = r'[\w+\ \.\,\:\+\&\-\/\[\]\(\)]+'
 CODE_REGEX = r'[0-9a-fA-F]{4}'
 BUSID_REGEX = r'[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-9a-fA-F]'
 

hwinfo/pci/tests/test_lspci.py

@@ -225,6 +225,24 @@ class EmulexNicDeviceParse(TestSingleDeviceNNMMParse):
         'pci_subdevice_id': 'e70b',
     }
 
+class EmulexHbDeviceParse(TestSingleDeviceNNMMParse):
+
+    SAMPLE_DATA = '07:00.0 "Fibre Channel [0c04]" "Emulex Corporation [10df]" "Saturn-X: LightPulse Fibre Channel Host Adapter [f100]" -r03 "Hewlett-Packard Company [103c]" "Device [3282]"'
+
+    DEVICE_REC = {
+        'pci_device_bus_id': '07:00.0',
+        'pci_device_class': '0c04',
+        'pci_device_class_name': 'Fibre Channel',
+        'pci_vendor_name': 'Emulex Corporation',
+        'pci_vendor_id': '10df',
+        'pci_device_id': 'f100',
+        'pci_device_name': 'Saturn-X: LightPulse Fibre Channel Host Adapter',
+        'pci_subvendor_name': 'Hewlett-Packard Company',
+        'pci_subvendor_id': '103c',
+        'pci_subdevice_name': 'Device',
+        'pci_subdevice_id': '3282',
+    }
+
 class LsiSASDeviceParse(TestSingleDeviceNNMMParse):
 
     SAMPLE_DATA = '06:00.0 "Serial Attached SCSI controller [0107]" "LSI Logic / Symbios Logic [1000]" "SAS2004 PCI-Express Fusion-MPT SAS-2 [Spitfire] [0070]" -r03 "IBM [1014]" "Device [03f8]"'

hwinfo/tools/inspector.py

@@ -6,6 +6,11 @@
 import subprocess
 import os
 import sys
+import tarfile
+import random
+import string
+import tempfile
+import shutil
 
 from hwinfo.pci import PCIDevice
 from hwinfo.pci.lspci import *
@@ -40,6 +45,34 @@ def local_command(cmd):
         print stdout
         raise Exception("stderr: %s" % str(stderr))
 
+def find_in_tarball(tarball, filename):
+    tar = tarfile.open(tarball)
+    members = tar.getmembers()
+    tar.close()
+    matches = []
+    for m in members:
+        if m.name.endswith(filename):
+            matches.append(m.name)
+
+    if len(matches) > 1:
+        raise Exception("Error: more than one match for that filename")
+
+    return matches.pop()
+
+def read_from_tarball(tarball, filename):
+    tar = tarfile.open(tarball)
+    data = None
+    tmpdir = tempfile.mkdtemp()
+
+    tar.extract(filename, tmpdir)
+    data = read_from_file("%s/%s" % (tmpdir, filename))
+
+    tar.close()
+    if os.path.exists(tmpdir):
+        shutil.rmtree(tmpdir)
+
+    return data
+
 class Host(object):
 
     client = None
@@ -155,6 +188,15 @@ def get_pci_devices(self):
             recs = combine_recs(all_recs, 'pci_device_bus_id')
             return [PCIDevice(rec) for rec in recs]
 
+class HostFromTarball(HostFromLogs):
+
+    def __init__(self, filename):
+        self.tarloc = filename
+
+    def _load_from_file(self, filename):
+        """Find filename in tar, and load it"""
+        filepath = find_in_tarball(self.tarloc, filename)
+        return read_from_tarball(self.tarloc, filepath)
 
 def pci_filter(devices, types):
     res = []

hwinfo/tools/tests/test_inspector.py

@@ -48,6 +48,57 @@ def test_is_remote(self, get_ssh_client):
         host = inspector.Host('test', 'user', 'pass')
         self.assertEqual(host.is_remote(), True)
 
+
+class HostFromTarballTests(unittest.TestCase):
+
+    @patch('hwinfo.tools.inspector.find_in_tarball')
+    @patch('hwinfo.tools.inspector.read_from_tarball')
+    def test__load_from_file(self, read_from_tarball, find_in_tarball):
+        tarball_name = 'test.tar.gz'
+        filename = 'foobar.txt'
+        filepath = find_in_tarball.return_value = \
+            'somedir/anotherdir/%s' % filename
+
+        host = inspector.HostFromTarball('test.tar.gz')
+        host._load_from_file(filename)
+
+        find_in_tarball.assert_called_once_with(tarball_name, filename)
+        read_from_tarball.assert_called_once_with(tarball_name, filepath)
+
+class UtilTests(unittest.TestCase):
+
+    @patch('tarfile.open')
+    def test_find_in_tarball(self, tar_open):
+        mtarfh = tar_open.return_value = mock.MagicMock()
+        filename = 'foo.bar'
+        filepath = 'testing/path/%s' % filename
+
+        paths = [filepath, 'something/else/foo.pdf', 'extra/file.bin']
+
+        mtar_infos = []
+        for p in paths:
+            m = mock.MagicMock()
+            m.name = p
+            mtar_infos.append(m)
+
+        mtarfh.getmembers.return_value = mtar_infos
+
+        path = inspector.find_in_tarball(mtarfh, filename)
+        self.assertEqual(filepath, path)
+
+
+    @patch('hwinfo.tools.inspector.read_from_file')
+    @patch('tarfile.open')
+    def test_read_from_tarball(self, tar_open, read_from_file):
+        mtarfh = tar_open.return_value = mock.MagicMock()
+        filepath = 'something/else/foo.pdf'
+
+        mread = read_from_file.return_value = 'blah blah blah'
+
+        data = inspector.read_from_tarball('testtar.tar.gz', filepath)
+        self.assertEqual(mread, data)
+
+
 class RemoteCommandTests(unittest.TestCase):
 
     def setUp(self):