[Secure-testing-commits] r16577 - bin lib/python

[Florian Weimer]

Author: fw
Date: 2011-04-24 14:25:18 +0000 (Sun, 24 Apr 2011)
New Revision: 16577

Modified:
   bin/list-queue
   lib/python/debian_support.py
Log:
debian_support.inspect_deb(): remove

bin/list-queue uses the parser in python-debian instead.


Modified: bin/list-queue
===================================================================
--- bin/list-queue	2011-04-24 14:07:43 UTC (rev 16576)
+++ bin/list-queue	2011-04-24 14:25:18 UTC (rev 16577)
@@ -44,13 +44,14 @@
 import sqlite3
 import json
 
+from debian.debfile import DebFile
 import debian_support
 
 def createdb():
     cache = os.path.expanduser("~/.cache")
     if not os.path.isdir(cache):
         os.mkdir(cache)
-    dbfile = os.path.join(cache, "secure-testing_list-debs.sqlite")
+    dbfile = os.path.join(cache, "secure-testing_list-queue.sqlite")
     db = sqlite3.connect(dbfile, isolation_level="IMMEDIATE")
     db.execute("PRAGMA page_size = 4096")
     db.execute("PRAGMA journal_mode = WAL")
@@ -105,7 +106,9 @@
                    ((path,) for path, _ in need_update))
     def do_update():
         for (path, stat) in need_update:
-            pkg = debian_support.inspect_deb(path)
+            deb = DebFile(path)
+            pkg = debian_support.BinaryPackage()
+            pkg.load822(deb.debcontrol())
             indb[path] = stat + (pkg,)
             yield (path,) + stat + pkg.astuple()
     db.executemany("INSERT INTO package VALUES (?, ?, ?, ?, ?, ?, ?, ?)",

Modified: lib/python/debian_support.py
===================================================================
--- lib/python/debian_support.py	2011-04-24 14:07:43 UTC (rev 16576)
+++ lib/python/debian_support.py	2011-04-24 14:25:18 UTC (rev 16577)
@@ -471,9 +471,29 @@
                         pkg_source, pkg_source_version))
 
     def loadtuple(self, data):
+        if None in data:
+            raise ValueError("None not permitted: " + repr(data))
         self.name, self.version, self.arch, self.source, self.source_version =\
             data
+    
+    def load822(self, data):
+        "Loads this object from a Deb822-like object."
 
+        pkg = data["Package"]
+        version = data["Version"]
+        if "Source" in data:
+            source = data.get("Source", None)
+            match = self.re_source.match(source)
+            if match is None:
+                raise ValueError("invalid Source field: " + repr(source))
+            src, src_version = match.groups()
+            if src_version is None:
+                src_version = version
+        else:
+            src = pkg
+            src_version = version
+        self.loadtuple((pkg, version, data["Architecture"], src, src_version))
+
     def astuple(self):
         return (self.name, self.version, self.arch,
                 self.source, self.source_version)
@@ -481,54 +501,6 @@
     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')