dak/process_policy.py

   1 #!/usr/bin/env python
   2 # vim:set et ts=4 sw=4:
   3
   4 """ Handles packages from policy queues
   5
   6 @contact: Debian FTP Master <ftpmaster@debian.org>
   7 @copyright: 2001, 2002, 2003, 2004, 2005, 2006  James Troup <james@nocrew.org>
   8 @copyright: 2009 Joerg Jaspert <joerg@debian.org>
   9 @copyright: 2009 Frank Lichtenheld <djpig@debian.org>
  10 @copyright: 2009 Mark Hymers <mhy@debian.org>
  11 @license: GNU General Public License version 2 or later
  12 """
  13 # This program is free software; you can redistribute it and/or modify
  14 # it under the terms of the GNU General Public License as published by
  15 # the Free Software Foundation; either version 2 of the License, or
  16 # (at your option) any later version.
  17
  18 # This program is distributed in the hope that it will be useful,
  19 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  20 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  21 # GNU General Public License for more details.
  22
  23 # You should have received a copy of the GNU General Public License
  24 # along with this program; if not, write to the Free Software
  25 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  26
  27 ################################################################################
  28
  29 # <mhy> So how do we handle that at the moment?
  30 # <stew> Probably incorrectly.
  31
  32 ################################################################################
  33
  34 import os
  35 import datetime
  36 import re
  37 import sys
  38 import traceback
  39 import apt_pkg
  40
  41 from daklib.dbconn import *
  42 from daklib import daklog
  43 from daklib import utils
  44 from daklib.dak_exceptions import CantOpenError, AlreadyLockedError, CantGetLockError
  45 from daklib.config import Config
  46 from daklib.archive import ArchiveTransaction
  47 from daklib.urgencylog import UrgencyLog
  48
  49 import daklib.announce
  50
  51 # Globals
  52 Options = None
  53 Logger = None
  54
  55 ################################################################################
  56
  57 def do_comments(dir, srcqueue, opref, npref, line, fn, transaction):
  58     session = transaction.session
  59     for comm in [ x for x in os.listdir(dir) if x.startswith(opref) ]:
  60         lines = open(os.path.join(dir, comm)).readlines()
  61         if len(lines) == 0 or lines[0] != line + "\n": continue
  62
  63         # If the ACCEPT includes a _<arch> we only accept that .changes.
  64         # Otherwise we accept all .changes that start with the given prefix
  65         changes_prefix = comm[len(opref):]
  66         if changes_prefix.count('_') < 2:
  67             changes_prefix = changes_prefix + '_'
  68         else:
  69             changes_prefix = changes_prefix + '.changes'
  70
  71         uploads = session.query(PolicyQueueUpload).filter_by(policy_queue=srcqueue) \
  72             .join(PolicyQueueUpload.changes).filter(DBChange.changesname.startswith(changes_prefix)) \
  73             .order_by(PolicyQueueUpload.source_id)
  74         for u in uploads:
  75             print "Processing changes file: %s" % u.changes.changesname
  76             fn(u, srcqueue, "".join(lines[1:]), transaction)
  77
  78         if opref != npref:
  79             newcomm = npref + comm[len(opref):]
  80             transaction.fs.move(os.path.join(dir, comm), os.path.join(dir, newcomm))
  81
  82 ################################################################################
  83
  84 def try_or_reject(function):
  85     def wrapper(upload, srcqueue, comments, transaction):
  86         try:
  87             function(upload, srcqueue, comments, transaction)
  88         except Exception as e:
  89             comments = 'An exception was raised while processing the package:\n{0}\nOriginal comments:\n{1}'.format(traceback.format_exc(), comments)
  90             try:
  91                 transaction.rollback()
  92                 real_comment_reject(upload, srcqueue, comments, transaction)
  93             except Exception as e:
  94                 comments = 'In addition an exception was raised while trying to reject the upload:\n{0}\nOriginal rejection:\n{1}'.format(traceback.format_exc(), comments)
  95                 transaction.rollback()
  96                 real_comment_reject(upload, srcqueue, comments, transaction, notify=False)
  97         if not Options['No-Action']:
  98             transaction.commit()
  99     return wrapper
 100
 101 ################################################################################
 102
 103 @try_or_reject
 104 def comment_accept(upload, srcqueue, comments, transaction):
 105     for byhand in upload.byhand:
 106         path = os.path.join(srcqueue.path, byhand.filename)
 107         if os.path.exists(path):
 108             raise Exception('E: cannot ACCEPT upload with unprocessed byhand file {0}'.format(byhand.filename))
 109
 110     cnf = Config()
 111
 112     fs = transaction.fs
 113     session = transaction.session
 114     changesname = upload.changes.changesname
 115     allow_tainted = srcqueue.suite.archive.tainted
 116
 117     # We need overrides to get the target component
 118     overridesuite = upload.target_suite
 119     if overridesuite.overridesuite is not None:
 120         overridesuite = session.query(Suite).filter_by(suite_name=overridesuite.overridesuite).one()
 121
 122     def binary_component_func(db_binary):
 123         override = session.query(Override).filter_by(suite=overridesuite, package=db_binary.package) \
 124             .join(OverrideType).filter(OverrideType.overridetype == db_binary.binarytype) \
 125             .join(Component).one()
 126         return override.component
 127
 128     def source_component_func(db_source):
 129         override = session.query(Override).filter_by(suite=overridesuite, package=db_source.source) \
 130             .join(OverrideType).filter(OverrideType.overridetype == 'dsc') \
 131             .join(Component).one()
 132         return override.component
 133
 134     all_target_suites = [upload.target_suite]
 135     all_target_suites.extend([q.suite for q in upload.target_suite.copy_queues])
 136
 137     for suite in all_target_suites:
 138         if upload.source is not None:
 139             transaction.copy_source(upload.source, suite, source_component_func(upload.source), allow_tainted=allow_tainted)
 140         for db_binary in upload.binaries:
 141             transaction.copy_binary(db_binary, suite, binary_component_func(db_binary), allow_tainted=allow_tainted, extra_archives=[upload.target_suite.archive])
 142
 143     # Copy .changes if needed
 144     if upload.target_suite.copychanges:
 145         src = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 146         dst = os.path.join(upload.target_suite.path, upload.changes.changesname)
 147         fs.copy(src, dst, mode=upload.target_suite.archive.mode)
 148
 149     if upload.source is not None and not Options['No-Action']:
 150         urgency = upload.changes.urgency
 151         if urgency not in cnf.value_list('Urgency::Valid'):
 152             urgency = cnf['Urgency::Default']
 153         UrgencyLog().log(upload.source.source, upload.source.version, urgency)
 154
 155     print "  ACCEPT"
 156     if not Options['No-Action']:
 157         Logger.log(["Policy Queue ACCEPT", srcqueue.queue_name, changesname])
 158
 159     pu = get_processed_upload(upload)
 160     daklib.announce.announce_accept(upload)
 161
 162     # TODO: code duplication. Similar code is in process-upload.
 163     # Move .changes to done
 164     src = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 165     now = datetime.datetime.now()
 166     donedir = os.path.join(cnf['Dir::Done'], now.strftime('%Y/%m/%d'))
 167     dst = os.path.join(donedir, upload.changes.changesname)
 168     dst = utils.find_next_free(dst)
 169     fs.copy(src, dst, mode=0o644)
 170
 171     remove_upload(upload, transaction)
 172
 173 ################################################################################
 174
 175 @try_or_reject
 176 def comment_reject(*args):
 177     real_comment_reject(*args, manual=True)
 178
 179 def real_comment_reject(upload, srcqueue, comments, transaction, notify=True, manual=False):
 180     cnf = Config()
 181
 182     fs = transaction.fs
 183     session = transaction.session
 184     changesname = upload.changes.changesname
 185     queuedir = upload.policy_queue.path
 186     rejectdir = cnf['Dir::Reject']
 187
 188     ### Copy files to reject/
 189
 190     poolfiles = [b.poolfile for b in upload.binaries]
 191     if upload.source is not None:
 192         poolfiles.extend([df.poolfile for df in upload.source.srcfiles])
 193     # Not beautiful...
 194     files = [ af.path for af in session.query(ArchiveFile) \
 195                   .filter_by(archive=upload.policy_queue.suite.archive) \
 196                   .join(ArchiveFile.file) \
 197                   .filter(PoolFile.file_id.in_([ f.file_id for f in poolfiles ])) ]
 198     for byhand in upload.byhand:
 199         path = os.path.join(queuedir, byhand.filename)
 200         if os.path.exists(path):
 201             files.append(path)
 202     files.append(os.path.join(queuedir, changesname))
 203
 204     for fn in files:
 205         dst = utils.find_next_free(os.path.join(rejectdir, os.path.basename(fn)))
 206         fs.copy(fn, dst, link=True)
 207
 208     ### Write reason
 209
 210     dst = utils.find_next_free(os.path.join(rejectdir, '{0}.reason'.format(changesname)))
 211     fh = fs.create(dst)
 212     fh.write(comments)
 213     fh.close()
 214
 215     ### Send mail notification
 216
 217     if notify:
 218         rejected_by = None
 219         reason = comments
 220
 221         # Try to use From: from comment file if there is one.
 222         # This is not very elegant...
 223         match = re.match(r"\AFrom: ([^\n]+)\n\n", comments)
 224         if match:
 225             rejected_by = match.group(1)
 226             reason = '\n'.join(comments.splitlines()[2:])
 227
 228         pu = get_processed_upload(upload)
 229         daklib.announce.announce_reject(pu, reason, rejected_by)
 230
 231     print "  REJECT"
 232     if not Options["No-Action"]:
 233         Logger.log(["Policy Queue REJECT", srcqueue.queue_name, upload.changes.changesname])
 234
 235     remove_upload(upload, transaction)
 236
 237 ################################################################################
 238
 239 def remove_upload(upload, transaction):
 240     fs = transaction.fs
 241     session = transaction.session
 242     changes = upload.changes
 243
 244     # Remove byhand and changes files. Binary and source packages will be
 245     # removed from {bin,src}_associations and eventually removed by clean-suites automatically.
 246     queuedir = upload.policy_queue.path
 247     for byhand in upload.byhand:
 248         path = os.path.join(queuedir, byhand.filename)
 249         if os.path.exists(path):
 250             fs.unlink(path)
 251         session.delete(byhand)
 252     fs.unlink(os.path.join(queuedir, upload.changes.changesname))
 253
 254     session.delete(upload)
 255     session.delete(changes)
 256     session.flush()
 257
 258 ################################################################################
 259
 260 def get_processed_upload(upload):
 261     pu = daklib.announce.ProcessedUpload()
 262
 263     pu.maintainer = upload.changes.maintainer
 264     pu.changed_by = upload.changes.changedby
 265     pu.fingerprint = upload.changes.fingerprint
 266
 267     pu.suites = []
 268     pu.from_policy_suites = [ upload.target_suite ]
 269
 270     changes_path = os.path.join(upload.policy_queue.path, upload.changes.changesname)
 271     pu.changes = open(changes_path, 'r').read()
 272     pu.changes_filename = upload.changes.changesname
 273     pu.sourceful = upload.source is not None
 274     pu.source = upload.changes.source
 275     pu.version = upload.changes.version
 276     pu.architecture = upload.changes.architecture
 277     pu.bugs = upload.changes.closes
 278
 279     pu.program = "process-policy"
 280
 281     return pu
 282
 283 ################################################################################
 284
 285 def remove_unreferenced_binaries(policy_queue, transaction):
 286     """Remove binaries that are no longer referenced by an upload
 287
 288     @type  policy_queue: L{daklib.dbconn.PolicyQueue}
 289
 290     @type  transaction: L{daklib.archive.ArchiveTransaction}
 291     """
 292     session = transaction.session
 293     suite = policy_queue.suite
 294
 295     query = """
 296        SELECT b.*
 297          FROM binaries b
 298          JOIN bin_associations ba ON b.id = ba.bin
 299         WHERE ba.suite = :suite_id
 300           AND NOT EXISTS (SELECT 1 FROM policy_queue_upload_binaries_map pqubm
 301                                    JOIN policy_queue_upload pqu ON pqubm.policy_queue_upload_id = pqu.id
 302                                   WHERE pqu.policy_queue_id = :policy_queue_id
 303                                     AND pqubm.binary_id = b.id)"""
 304     binaries = session.query(DBBinary).from_statement(query) \
 305         .params({'suite_id': policy_queue.suite_id, 'policy_queue_id': policy_queue.policy_queue_id})
 306
 307     for binary in binaries:
 308         Logger.log(["removed binary from policy queue", policy_queue.queue_name, binary.package, binary.version])
 309         transaction.remove_binary(binary, suite)
 310
 311 def remove_unreferenced_sources(policy_queue, transaction):
 312     """Remove sources that are no longer referenced by an upload or a binary
 313
 314     @type  policy_queue: L{daklib.dbconn.PolicyQueue}
 315
 316     @type  transaction: L{daklib.archive.ArchiveTransaction}
 317     """
 318     session = transaction.session
 319     suite = policy_queue.suite
 320
 321     query = """
 322        SELECT s.*
 323          FROM source s
 324          JOIN src_associations sa ON s.id = sa.source
 325         WHERE sa.suite = :suite_id
 326           AND NOT EXISTS (SELECT 1 FROM policy_queue_upload pqu
 327                                   WHERE pqu.policy_queue_id = :policy_queue_id
 328                                     AND pqu.source_id = s.id)
 329           AND NOT EXISTS (SELECT 1 FROM binaries b
 330                                    JOIN bin_associations ba ON b.id = ba.bin
 331                                   WHERE b.source = s.id
 332                                     AND ba.suite = :suite_id)"""
 333     sources = session.query(DBSource).from_statement(query) \
 334         .params({'suite_id': policy_queue.suite_id, 'policy_queue_id': policy_queue.policy_queue_id})
 335
 336     for source in sources:
 337         Logger.log(["removed source from policy queue", policy_queue.queue_name, source.source, source.version])
 338         transaction.remove_source(source, suite)
 339
 340 ################################################################################
 341
 342 def main():
 343     global Options, Logger
 344
 345     cnf = Config()
 346     session = DBConn().session()
 347
 348     Arguments = [('h',"help","Process-Policy::Options::Help"),
 349                  ('n',"no-action","Process-Policy::Options::No-Action")]
 350
 351     for i in ["help", "no-action"]:
 352         if not cnf.has_key("Process-Policy::Options::%s" % (i)):
 353             cnf["Process-Policy::Options::%s" % (i)] = ""
 354
 355     queue_name = apt_pkg.parse_commandline(cnf.Cnf,Arguments,sys.argv)
 356
 357     if len(queue_name) != 1:
 358         print "E: Specify exactly one policy queue"
 359         sys.exit(1)
 360
 361     queue_name = queue_name[0]
 362
 363     Options = cnf.subtree("Process-Policy::Options")
 364
 365     if Options["Help"]:
 366         usage()
 367
 368     Logger = daklog.Logger("process-policy")
 369     if not Options["No-Action"]:
 370         urgencylog = UrgencyLog()
 371
 372     with ArchiveTransaction() as transaction:
 373         session = transaction.session
 374         try:
 375             pq = session.query(PolicyQueue).filter_by(queue_name=queue_name).one()
 376         except NoResultFound:
 377             print "E: Cannot find policy queue %s" % queue_name
 378             sys.exit(1)
 379
 380         commentsdir = os.path.join(pq.path, 'COMMENTS')
 381         # The comments stuff relies on being in the right directory
 382         os.chdir(pq.path)
 383
 384         do_comments(commentsdir, pq, "ACCEPT.", "ACCEPTED.", "OK", comment_accept, transaction)
 385         do_comments(commentsdir, pq, "ACCEPTED.", "ACCEPTED.", "OK", comment_accept, transaction)
 386         do_comments(commentsdir, pq, "REJECT.", "REJECTED.", "NOTOK", comment_reject, transaction)
 387
 388         remove_unreferenced_binaries(pq, transaction)
 389         remove_unreferenced_sources(pq, transaction)
 390
 391     if not Options['No-Action']:
 392         urgencylog.close()
 393
 394 ################################################################################
 395
 396 if __name__ == '__main__':
 397     main()