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'')