debian_support.inspect_deb(): attempt add a .deb parser

Also adds debian_support.BinaryPackage class.


git-svn-id: svn+ssh://svn.debian.org/svn/secure-testing@16575 e39458fd-73e7-0310-bf30-c45bca0a0e42

1 files changed, 114 insertions, 0 deletions

diff --git a/lib/python/debian_support.py b/lib/python/debian_support.py
index 3477291d25..59d4c99a89 100644
--- a/lib/python/debian_support.py
+++ b/lib/python/debian_support.py
@@ -20,6 +20,9 @@
 import types
 import os
 import re
+import subprocess
+import tempfile
+from cStringIO import StringIO
 
 try:
     from hashlib import sha1
@@ -415,6 +418,117 @@ def mergeAsSets(*args):
     l.sort()
     return l
 
+class BinaryPackage(object):
+    __slots__ = ("name", "version", "arch", "source", "source_version")
+    re_source = re.compile\
+        (r'^([a-zA-Z0-9.+-]+)(?:\s+\(([a-zA-Z0-9.+:~-]+)\))?$')
+
+    def __init__(self, data=None):
+        if data is not None:
+            self.loadtuple(data)
+
+    def loadentry(self, lines):
+        """Loads an entry from the Packages file.
+
+        LINES is a sequence of string pairs (KEY, VALUE).
+        """
+        pkg_name = None
+        pkg_version = None
+        pkg_arch = None
+        pkg_source = None
+        pkg_source_version = None
+        for (name, contents) in lines:
+            name = name.lower()
+            if name == "package":
+                pkg_name = contents
+            elif name == "version":
+                pkg_version = contents
+            elif name == "source":
+                match = self.re_source.match(contents)
+                if match is None:
+                    raise SyntaxError(('package %s references '
+                                       + 'invalid source package %s') %
+                                      (pkg_name, `contents`))
+                (pkg_source, pkg_source_version) = match.groups()
+            elif name == "architecture":
+                pkg_arch = contents
+        if pkg_name is None:
+            raise SyntaxError\
+                  ("package record does not contain package name")
+        if pkg_version is None:
+            raise SyntaxError\
+                  ("package record for %s does not contain version"
+                   % pkg_name)
+        if pkg_arch is None:
+            raise SyntaxError\
+                  ("package record for %s lacks Architecture: field"
+                   % pkg_name)
+        if pkg_source is None:
+            pkg_source = pkg_name
+        if pkg_source_version is None:
+            pkg_source_version = pkg_version
+        self.loadtuple((pkg_name, pkg_version, pkg_arch,
+                        pkg_source, pkg_source_version))
+
+    def loadtuple(self, data):
+        self.name, self.version, self.arch, self.source, self.source_version =\
+            data
+
+    def astuple(self):
+        return (self.name, self.version, self.arch,
+                self.source, self.source_version)
+
+    def __repr__(self):
+        return "BinaryPackage(" + repr(self.astuple()) + ")"
+
+def inspect_deb(path):
+    "Extracts meta-data from a Debian package file (.deb)."
+    devnull = file("/dev/null", "r")
+    tempout = tempfile.TemporaryFile()
+    temperr = tempfile.TemporaryFile()
+    dpkg = subprocess.Popen(("dpkg", "-I", "--", path),
+                            stdin=devnull,
+                            stdout=tempout,
+                            stderr=temperr)
+    dpkg.wait()
+    temperr.seek(0)
+    temperr = temperr.read()
+    if temperr:
+        raise IOError("unexpected dpkg output for " + repr(path) + ": "
+                      + repr(temperr))
+    if dpkg.returncode <> 0:
+        raise IOError("unexpected dpkg status for " + repr(path) + ": "
+                      + repr(dpkg.returncode))
+    tempout.seek(0)
+    lines = list(tempout.readlines())
+    if not lines:
+        raise IOError("empty dpkg output for " + repr(path))
+    if lines[0] <> " new debian package, version 2.0.\n":
+        raise IOError("unexpected dpkg format for " +repr(path) + ":"
+                      + repr(lines[0]))
+    while lines and not lines[0].startswith(" Package: "):
+        del lines[0]
+    if not lines:
+        raise IOError("no Package: line in dpkg output for " +repr(path))
+    stripped = StringIO()
+    for line in lines:
+        if line.startswith(" "):
+            stripped.write(line[1:])
+        else:
+            break
+    stripped.seek(0)
+    pkgfile = PackageFile(path, stripped)
+    result = None
+    for pkg in pkgfile:
+        if result is None:
+            result = BinaryPackage()
+            result.loadentry(pkg)
+        else:
+            raise IOError("multiple package entries for " + repr(path))
+    if result is None:
+        raise IOError("no package entries for " + repr(path))
+    return result
+
 def test():
     # Version
     assert Version('0') < Version('a')