#!/usr/bin/env python
-# Checks Debian packages from Incoming
+""" Checks Debian packages from Incoming """
# Copyright (C) 2000, 2001, 2002, 2003, 2004, 2005, 2006 James Troup <james@nocrew.org>
# This program is free software; you can redistribute it and/or modify
from daklib import queue
from daklib import utils
from daklib.dak_exceptions import *
+from daklib.regexes import re_valid_version, re_valid_pkg_name, re_changelog_versions, \
+ re_strip_revision, re_strip_srcver, re_spacestrip, \
+ re_isanum, re_no_epoch, re_no_revision, re_taint_free, \
+ re_isadeb, re_extract_src_version, re_issource, re_default_answer
from types import *
################################################################################
-re_valid_version = re.compile(r"^([0-9]+:)?[0-9A-Za-z\.\-\+:~]+$")
-re_valid_pkg_name = re.compile(r"^[\dA-Za-z][\dA-Za-z\+\-\.]+$")
-re_changelog_versions = re.compile(r"^\w[-+0-9a-z.]+ \([^\(\) \t]+\)")
-re_strip_revision = re.compile(r"-([^-]+)$")
-re_strip_srcver = re.compile(r"\s+\(\S+\)$")
-re_spacestrip = re.compile('(\s)')
################################################################################
('h',"help","Dinstall::Options::Help"),
('n',"no-action","Dinstall::Options::No-Action"),
('p',"no-lock", "Dinstall::Options::No-Lock"),
- ('s',"no-mail", "Dinstall::Options::No-Mail")]
+ ('s',"no-mail", "Dinstall::Options::No-Mail"),
+ ('d',"directory", "Dinstall::Options::Directory", "HasArg")]
for i in ["automatic", "help", "no-action", "no-lock", "no-mail",
- "override-distribution", "version"]:
+ "override-distribution", "version", "directory"]:
Cnf["Dinstall::Options::%s" % (i)] = ""
changes_files = apt_pkg.ParseCommandLine(Cnf,Arguments,sys.argv)
if Options["Help"]:
usage()
+ # If we have a directory flag, use it to find our files
+ if Cnf["Dinstall::Options::Directory"] != "":
+ # Note that we clobber the list of files we were given in this case
+ # so warn if the user has done both
+ if len(changes_files) > 0:
+ utils.warn("Directory provided so ignoring files given on command line")
+
+ changes_files = utils.get_changes_files(Cnf["Dinstall::Options::Directory"])
+
Upload = queue.Upload(Cnf)
changes = Upload.pkg.changes
except ParseChangesError, line:
reject("%s: parse error, can't grok: %s." % (filename, line))
return 0
+ except ChangesUnicodeError:
+ reject("%s: changes file not proper utf-8" % (filename))
+ return 0
# Parse the Files field from the .changes into another dictionary
try:
# Ensure all the values in Closes: are numbers
if changes.has_key("closes"):
for i in changes["closes"].keys():
- if queue.re_isanum.match (i) == None:
+ if re_isanum.match (i) == None:
reject("%s: `%s' from Closes field isn't a number." % (filename, i))
# chopversion = no epoch; chopversion2 = no epoch and no revision (e.g. for .orig.tar.gz comparison)
- changes["chopversion"] = utils.re_no_epoch.sub('', changes["version"])
- changes["chopversion2"] = utils.re_no_revision.sub('', changes["chopversion"])
+ changes["chopversion"] = re_no_epoch.sub('', changes["version"])
+ changes["chopversion2"] = re_no_revision.sub('', changes["chopversion"])
# Check there isn't already a changes file of the same name in one
# of the queue directories.
(source, dest) = args[1:3]
if changes["distribution"].has_key(source):
for arch in changes["architecture"].keys():
- if arch not in Cnf.ValueList("Suite::%s::Architectures" % (source)):
+ if arch not in database.get_suite_architectures(source):
reject("Mapping %s to %s for unreleased architecture %s." % (source, dest, arch),"")
del changes["distribution"][source]
changes["distribution"][dest] = 1
################################################################################
def check_deb_ar(filename):
- """Sanity check the ar of a .deb, i.e. that there is:
+ """
+ Sanity check the ar of a .deb, i.e. that there is:
- o debian-binary
- o control.tar.gz
- o data.tar.gz or data.tar.bz2
+ 1. debian-binary
+ 2. control.tar.gz
+ 3. data.tar.gz or data.tar.bz2
-in that order, and nothing else."""
+ in that order, and nothing else.
+ """
cmd = "ar t %s" % (filename)
(result, output) = commands.getstatusoutput(cmd)
if result != 0:
if not Cnf.has_key("Dir::Queue::%s" % (d)): continue
if os.path.exists(Cnf["Dir::Queue::%s" % (d) ] + '/' + f):
reject("%s file already exists in the %s directory." % (f, d))
- if not utils.re_taint_free.match(f):
+ if not re_taint_free.match(f):
reject("!!WARNING!! tainted filename: '%s'." % (f))
# Check the file is readable
if os.access(f, os.R_OK) == 0:
files[f]["byhand"] = 1
files[f]["type"] = "byhand"
# Checks for a binary package...
- elif utils.re_isadeb.match(f):
+ elif re_isadeb.match(f):
has_binaries = 1
files[f]["type"] = "deb"
default_suite = Cnf.get("Dinstall::DefaultSuite", "Unstable")
architecture = control.Find("Architecture")
upload_suite = changes["distribution"].keys()[0]
- if architecture not in Cnf.ValueList("Suite::%s::Architectures" % (default_suite)) and architecture not in Cnf.ValueList("Suite::%s::Architectures" % (upload_suite)):
+ if architecture not in database.get_suite_architectures(default_suite) and architecture not in database.get_suite_architectures(upload_suite):
reject("Unknown architecture '%s'." % (architecture))
# Ensure the architecture of the .deb is one of the ones
source = files[f]["source"]
source_version = ""
if source.find("(") != -1:
- m = utils.re_extract_src_version.match(source)
+ m = re_extract_src_version.match(source)
source = m.group(1)
source_version = m.group(2)
if not source_version:
files[f]["source version"] = source_version
# Ensure the filename matches the contents of the .deb
- m = utils.re_isadeb.match(f)
+ m = re_isadeb.match(f)
# package name
file_package = m.group(1)
if files[f]["package"] != file_package:
reject("%s: package part of filename (%s) does not match package name in the %s (%s)." % (f, file_package, files[f]["dbtype"], files[f]["package"]))
- epochless_version = utils.re_no_epoch.sub('', control.Find("Version"))
+ epochless_version = re_no_epoch.sub('', control.Find("Version"))
# version
file_version = m.group(2)
if epochless_version != file_version:
# Check in the SQL database
if not Upload.source_exists(source_package, source_version, changes["distribution"].keys()):
# Check in one of the other directories
- source_epochless_version = utils.re_no_epoch.sub('', source_version)
+ source_epochless_version = re_no_epoch.sub('', source_version)
dsc_filename = "%s_%s.dsc" % (source_package, source_epochless_version)
if os.path.exists(Cnf["Dir::Queue::Byhand"] + '/' + dsc_filename):
files[f]["byhand"] = 1
# Checks for a source package...
else:
- m = utils.re_issource.match(f)
+ m = re_issource.match(f)
if m:
has_source = 1
files[f]["package"] = m.group(1)
reject("%s: parse error, can't grok: %s." % (dsc_filename, line))
except InvalidDscError, line:
reject("%s: syntax error on line %s." % (dsc_filename, line))
+ except ChangesUnicodeError:
+ reject("%s: dsc file not proper utf-8." % (dsc_filename))
+
# Build up the file list of files mentioned by the .dsc
try:
dsc_files.update(utils.build_file_list(dsc, is_a_dsc=1))
pass
# Ensure the version number in the .dsc matches the version number in the .changes
- epochless_dsc_version = utils.re_no_epoch.sub('', dsc["version"])
+ epochless_dsc_version = re_no_epoch.sub('', dsc["version"])
changes_version = files[dsc_filename]["version"]
if epochless_dsc_version != files[dsc_filename]["version"]:
reject("version ('%s') in .dsc does not match version ('%s') in .changes." % (epochless_dsc_version, changes_version))
# Ensure there is a .tar.gz in the .dsc file
has_tar = 0
for f in dsc_files.keys():
- m = utils.re_issource.match(f)
+ m = re_issource.match(f)
if not m:
reject("%s: %s in Files field not recognised as source." % (dsc_filename, f))
continue
# Create a symlink mirror of the source files in our temporary directory
for f in files.keys():
- m = utils.re_issource.match(f)
+ m = re_issource.match(f)
if m:
src = os.path.join(source_dir, f)
# If a file is missing for whatever reason, give up.
return
# Get the upstream version
- upstr_version = utils.re_no_epoch.sub('', dsc["version"])
+ upstr_version = re_no_epoch.sub('', dsc["version"])
if re_strip_revision.search(upstr_version):
upstr_version = re_strip_revision.sub('', upstr_version)
################################################################################
def lookup_uid_from_fingerprint(fpr):
- q = Upload.projectB.query("SELECT u.uid, u.name, u.debian_maintainer FROM fingerprint f, uid u WHERE f.uid = u.id AND f.fingerprint = '%s'" % (fpr))
+ q = Upload.projectB.query("SELECT u.uid, u.name, k.debian_maintainer FROM fingerprint f JOIN keyrings k ON (f.keyring=k.id), uid u WHERE f.uid = u.id AND f.fingerprint = '%s'" % (fpr))
qs = q.getresult()
if len(qs) == 0:
- return (None, None)
+ return (None, None, None)
else:
return qs[0]
highest_sid, highest_version = None, None
- is_allowed = 1
+ should_reject = True
for si in q.getresult():
if highest_version == None or apt_pkg.VersionCompare(si[1], highest_version) == 1:
highest_sid = si[0]
highest_version = si[1]
- print highest_sid
if highest_sid == None:
reject("Source package %s does not have 'DM-Upload-Allowed: yes' in its most recent version" % changes["source"])
else:
for m in q.getresult():
(rfc822, rfc2047, name, email) = utils.fix_maintainer(m[0])
if email == uid_email or name == uid_name:
- is_allowed=0
+ should_reject=False
break
- if is_allowed:
+ if should_reject == True:
reject("%s is not in Maintainer or Uploaders of source package %s" % (uid, changes["source"]))
for b in changes["binary"].keys():
while prompt.find(answer) == -1:
answer = utils.our_raw_input(prompt)
- m = queue.re_default_answer.match(prompt)
+ m = re_default_answer.match(prompt)
if answer == "":
answer = m.group(1)
answer = answer[:1].upper()
def do_stableupdate (summary, short_summary):
print "Moving to PROPOSED-UPDATES holding area."
- Logger.log(["Moving to proposed-updates", pkg.changes_file]);
+ Logger.log(["Moving to proposed-updates", pkg.changes_file])
- Upload.dump_vars(Cnf["Dir::Queue::ProposedUpdates"]);
+ Upload.dump_vars(Cnf["Dir::Queue::ProposedUpdates"])
move_to_dir(Cnf["Dir::Queue::ProposedUpdates"], perms=0664)
# Check for override disparities
- Upload.Subst["__SUMMARY__"] = summary;
- Upload.check_override();
+ Upload.Subst["__SUMMARY__"] = summary
+ Upload.check_override()
################################################################################
def do_oldstableupdate (summary, short_summary):
print "Moving to OLDSTABLE-PROPOSED-UPDATES holding area."
- Logger.log(["Moving to oldstable-proposed-updates", pkg.changes_file]);
+ Logger.log(["Moving to oldstable-proposed-updates", pkg.changes_file])
- Upload.dump_vars(Cnf["Dir::Queue::OldProposedUpdates"]);
+ Upload.dump_vars(Cnf["Dir::Queue::OldProposedUpdates"])
move_to_dir(Cnf["Dir::Queue::OldProposedUpdates"], perms=0664)
# Check for override disparities
- Upload.Subst["__SUMMARY__"] = summary;
- Upload.check_override();
+ Upload.Subst["__SUMMARY__"] = summary
+ Upload.check_override()
################################################################################
projects / dak.git / blobdiff