~vila/udd/717204-stop-too-fast

#!/usr/bin/python

import cgi
import datetime
import errno
import fcntl
import operator
import os
import random
import re
import shutil
import simplejson
import sqlite3
import StringIO
import sys
import time
import urllib
import urllib2
import urlparse

try:
    from debian import changelog
except ImportError:
    from debian_bundle import changelog

from bzrlib import branch, bzrdir, delta, diff, errors, merge, tag, testament, transport, ui
from bzrlib.trace import mutter


from launchpadlib.credentials import Credentials
from launchpadlib.errors import HTTPError
from launchpadlib.launchpad import Launchpad, EDGE_SERVICE_ROOT


SERVICE_ROOT = EDGE_SERVICE_ROOT.replace("edge.", "")

base_dir = "/srv/package-import.canonical.com/new/"
if not os.path.exists(base_dir):
    base_dir = "."
logs_dir = os.path.join(base_dir, "logs")
lists_dir = os.path.join(base_dir, "lists")
# FIXME: This is needed for tests and tests may be run in various dirs,
# creating a new credentials file for any random branch is not optimal :-)
# -- vila 20110219
lp_creds_file = os.path.join(base_dir, "lp_creds.txt")
stop_file = os.path.join(base_dir, "STOP_PLEASE")
debug_log_file = os.path.join(base_dir, "debug_log")
progress_log_file = os.path.join(base_dir, "progress_log")
max_threads_file = os.path.join(base_dir, "max_threads")
locks_dir = os.path.join(base_dir, "locks")
updates_dir = os.path.join(base_dir, "updates")
download_cache_dir = os.path.join(base_dir, "download-cache")
localbranches_dir = os.path.join(base_dir, "localbranches")
sqlite_file = os.path.join(base_dir, "meta.db")
sqlite_package_file = os.path.join(base_dir, "packages.db")
sqlite_old_failures_file = os.path.join(base_dir, "failures.db")
sqlite_history_file = os.path.join(base_dir, "history.db")
explanations_file = os.path.join(base_dir, "explanations")
sqlite_timeout = 30
web_base_dir = logs_dir
debian_diffs_dir = os.path.join(web_base_dir, "diffs")
ubuntu_merge_dir = os.path.join(web_base_dir, "merges")

running_sentinel = "Apparently the supervisor died\n"
no_lock_returncode = 139


for directory in (logs_dir, lists_dir,
                  locks_dir, updates_dir,
                  debian_diffs_dir, ubuntu_merge_dir,
                 ):
    if not os.path.exists(directory):
        os.mkdir(directory)

urllib_launchpad_base_url = "https+urllib://launchpad.net/"
urllib_debian_base_url = "http+urllib://ftp.uk.debian.org/debian/"
urllib_debian_archive_base_url = "http+urllib://archive.debian.org/debian/"

distro_releases = {
    "ubuntu": ["warty", "hoary", "breezy",
               "dapper", "edgy", "feisty", "gutsy",
               "hardy", "intrepid", "jaunty", "karmic",
               "lucid", "maverick", "natty"],
    "debian": ["woody", "sarge", "etch", "lenny",
               "squeeze", "sid", "experimental"],
}

distro_pockets = {
    "ubuntu": ["release", "security", "proposed", "updates", "backports"],
    "debian": ["release",],
}

# development releases first for stacking, no etch or earlier as they aren't on
# lp
lp_distro_releases = {"ubuntu": ["natty", "warty", "hoary", "breezy",
                                 "dapper", "edgy", "feisty", "gutsy",
                                 "hardy", "intrepid", "jaunty", "karmic",
                                 "lucid", "maverick"],
                      "debian": ["squeeze", "lenny", "sid", "experimental"],
}

default_debian_diff_release = "sid"
default_ubuntu_merge_source = "squeeze"


def get_lp(cache=None):
    if os.path.exists(lp_creds_file):
        credentials = Credentials()
        credentials.load(file(lp_creds_file, "r"))
        launchpad = Launchpad(credentials, SERVICE_ROOT, cache)
        # TODO: handle the case of having credentials that have expired etc
    else:
        launchpad = Launchpad.get_token_and_login('package-import',
                                                  SERVICE_ROOT, cache)
        launchpad.credentials.save(file(lp_creds_file, "w"))
    return launchpad


def get_sqlite_connection(path):
    conn = sqlite3.connect(path, timeout=sqlite_timeout,
            detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)
    conn.row_factory = sqlite3.Row
    return conn


def debian_base_url(release):
    if release in ["hamm", "slink", "potato", "woody", "sarge", "etch"]:
        return urllib_debian_archive_base_url
    return urllib_debian_base_url


def make_suite(release, pocket):
    if pocket == "release":
        suite = release
    elif pocket in ("security", "proposed", "updates", "backports"):
        suite = "%s-%s" % (release, pocket)
    else:
        assert False, "Unkown pocket: %s" % pocket
    return suite


def _lock_path(path):
    f = open(path, "wb")
    try:
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        return f
    except IOError, e:
        if e.errno in (errno.EACCES, errno.EAGAIN):
            return None
        raise

def lock_package(package):
    path = os.path.join(locks_dir, package)
    return _lock_path(path)


def lock_main():
    path = os.path.join(base_dir, "main_lock")
    return _lock_path(path)


def lock_list_packages():
    path = os.path.join(base_dir, "list_packages")
    return _lock_path(path)


def lock_add_import_jobs():
    path = os.path.join(base_dir, "add_import_jobs")
    return _lock_path(path)


def lock_categorise_failures():
    path = os.path.join(base_dir, "categorise_failures")
    return _lock_path(path)


def lock_email_failures():
    path = os.path.join(base_dir, "email_failures")
    return _lock_path(path)


def lock_update_lists():
    path = os.path.join(base_dir, "update_lists")
    return _lock_path(path)


def lp_call(callable, *args, **kwargs):
    retries = 4
    retried = 0
    while retried < retries:
        try:
            return callable(*args, **kwargs)
        except HTTPError, e:
            retried += 1
            if retried >= retries:
                raise
            if e.response.status // 100 == 5:
                # randomized exponential backoff
                delay = random.randint(10,15)**(retried//2)
                mutter("Sleeping for %d due to %s from LP." % (delay, e))
                time.sleep(delay)
            else:
                if e.response.status == 302:
                    print "Redirected to %s" % e.response.get("location")
                    print "\n".join(["%s: %s"
                                     % pair for pair in e.response.items()])
                    print "\n-----\n%s\n----\n" % e.content
                raise
    assert False, "Shouldn't get here"


class PackageToImport(object):

    def __init__(self, name, version, distro, release, pocket, component="",
            on_lp=True, url=None):
        self.name = name
        self.version = version
        self.distro = distro
        self.release = release
        self.component = component
        self.pocket = pocket
        self.on_lp = on_lp
        self.url = url

    def __repr__(self):
        return ('%s(%s %s %s/%s/%s, c=%s, url=%s)'
                % (self.__class__.__name__, self.name, self.version,
                   self.distro, self.release, self.pocket, self.component,
                   self.url))

    def same_as(self, other):
        return (self.name == other.name
                and self.version == other.version
                and self.distro == other.distro
                and self.release == other.release
                and self.pocket == other.pocket)

    def get_url(self):
        if self.url is not None:
            return self.url
        if self.version.debian_version is not None:
            version_str = "%s-%s" % (self.version.upstream_version,
                    self.version.debian_version)
        else:
            version_str = self.version.upstream_version
        if self.distro == "ubuntu":
            files_url = (urllib_launchpad_base_url
                         + "ubuntu/%s/+source/%s/%s/+files/"
                         % (self.release, self.name, str(self.version)))
        else:
            if self.on_lp:
                files_url = (urllib_launchpad_base_url
                             + "debian/%s/+source/%s/%s/+files/"
                             % (self.release, self.name, str(self.version)))
            else:
                start = self.name[0]
                if self.name.startswith("lib"):
                    start = self.name[:4]
                files_url = (debian_base_url(self.release)
                             + "pool/%s/%s/%s/"
                             % (self.component, start, self.name))
        self.url = files_url + "%s_%s.dsc" % (self.name, version_str)
        return self.url

    def __str__(self):
        return "PackageToImport(%s, %s, %s, %s, %s, %s)" % (self.name,
                self.version, self.distro, self.release, self.pocket,
                self.component)


class ImportList(object):

    def __init__(self):
        self.plist = []

    def __repr__(self):
        return '%s(%s)' % (self.__class__.__name__, self.plist)

    def sort(self):
        def compare_versions(a, b):
            if a.distro == "ubuntu" and a.distro == b.distro:
                if a.pocket == "backports" and b.pocket != "backports":
                    return -1
                if a.pocket != "backports" and b.pocket == "backports":
                    return 1
            if a.distro != b.distro:
                if a.distro == "debian":
                    return 1
                return -1
            if a.release != b.release:
                a_release = distro_releases[a.distro].index(a.release)
                b_release = distro_releases[b.distro].index(b.release)
                if a_release < b_release:
                    return 1
                return -1
            if a.pocket != b.pocket:
                a_pocket = distro_pockets[a.distro].index(a.pocket)
                b_pocket = distro_pockets[b.distro].index(b.pocket)
                if a_pocket < b_pocket:
                    return 1
                return -1
            if a.version < b.version:
                return 1
            if a.version > b.version:
                return -1
            return 0
        self.plist.sort(cmp=compare_versions)

    def reverse(self):
        self.plist.reverse()

    def add_if_needed(self, newp):
        for oldp in self.plist:
            if oldp.same_as(newp):
                return
        self.plist.append(newp)

    def __str__(self):
        return "[" + ", ".join(map(str, self.plist)) + "]"


def import_from_publications(status_db, publications, last_known_published):
    checked = set()
    newest_published = last_known_published
    for publication in publications:
        published_time = datetime.datetime.strptime(
            str(publication.date_created)[:19], "%Y-%m-%d %H:%M:%S")
        name = publication.source_package_name
        checked.add(name)
        if (newest_published is None or published_time > newest_published):
            newest_published = published_time
    status_db.add_import_jobs(checked, newest_published)


def create_import_jobs(lp, status_db):
    last_known_published = status_db.last_import_time()
    last_published_call = None
    if last_known_published is not None:
        last_known_published = \
                last_known_published - datetime.timedelta(0, 600)
        last_published_call = last_known_published.isoformat()
    # We want all the new Published records since the last run
    publications = set()
    for p in iterate_collection(lp_call(call_with_limited_size,
            lp.distributions['ubuntu'].main_archive.getPublishedSources,
            status="Published",
            created_since_date=last_published_call)):
        publications.add(p)
    for p in iterate_collection(lp_call(call_with_limited_size,
            lp.distributions['debian'].main_archive.getPublishedSources,
            status="Pending",
            created_since_date=last_published_call)):
        publications.add(p)
    import_from_publications(status_db, publications, last_known_published)


class PackageStatus(object):

    def __init__(self, name):
        self.name = name
        self.failure_count = 0
        self.running = False
        self.queued = False
        self.auto_retry = False
        self.auto_retry_masked = False
        self.timestamp = None
        self.auto_retry_time = None
        self.signature = None
        self.raw_reason = None


class StatusDatabase(object):

    FAILURES_TABLE = "failures"
    FAILURES_TABLE_CREATE = '''create table if not exists %s
                     (package text constraint nonull NOT NULL,
                      reason blob,
                      when_failed timestamp,
                      emailed integer default 0,
                      constraint isprimary PRIMARY KEY
                            (package))''' % FAILURES_TABLE
    FAILURES_TABLE_FIND = '''select * from %s where package=?''' % FAILURES_TABLE
    FAILURES_TABLE_DELETE = '''delete from %s where package=?''' % FAILURES_TABLE

    OLD_FAILURES_TABLE = "old_failures"
    OLD_FAILURES_TABLE_CREATE = '''create table if not exists %s
                     (package text constraint nonull NOT NULL,
                      reason blob,
                      when_failed timestamp,
                      last_failed timestamp,
                      failure_count integer,
                      constraint isprimary PRIMARY KEY
                            (package))''' % OLD_FAILURES_TABLE
    OLD_FAILURES_TABLE_FIND = '''select * from %s where package=?''' % OLD_FAILURES_TABLE
    OLD_FAILURES_TABLE_DELETE = '''delete from %s where package=?''' % OLD_FAILURES_TABLE

    JOBS_TABLE = "jobs"
    JOBS_TABLE_CREATE = '''create table if not exists %s
                    (id integer primary key asc,
                     package text constraint nonull NOT NULL,
                     active integer constraint nonull NOT NULL,
                     type integer constraint nonull NOT NULL,
                     date_requested timestamp constraint nonull NOT NULL,
                     date_started timestamp,
                     date_completed timestamp)''' % JOBS_TABLE
    JOBS_TABLE_FIND = '''select * from %s where package=?
                    and active=? order by type desc''' % JOBS_TABLE
    JOBS_TABLE_NEXT = '''select * from %s where active=1
                    order by type desc''' % JOBS_TABLE
    JOBS_TABLE_SELECT_JOB = '''select * from %s where id=?''' % JOBS_TABLE
    JOBS_TABLE_INSERT = '''insert into %s values (?, ?, ?, ?, ?, ?, ?)''' % JOBS_TABLE
    JOBS_TABLE_UPDATE = '''update %s set package=?, active=?, type=?,
                    date_requested=?, date_started=?, date_completed=?
                    where id=?''' % JOBS_TABLE

    IMPORT_TABLE = "import"
    IMPORT_TABLE_CREATE = '''create table if not exists %s
                    (import timestamp constraint nonull NOT NULL)
                    ''' % IMPORT_TABLE

    RETRY_TABLE = "should_retry"
    RETRY_TABLE_CREATE = '''create table if not exists %s
                    (signature blob constraint nonull NOT NULL,
                     constraint isprimary PRIMARY KEY
                            (signature))''' % RETRY_TABLE
    RETRY_TABLE_SELECT = '''select * from %s where signature=?''' % RETRY_TABLE
    RETRY_TABLE_INSERT = '''insert into %s values (?)''' % RETRY_TABLE

    JOB_TYPE_PRIORITY = 5
    JOB_TYPE_NEW = 3
    JOB_TYPE_RETRY = 1

    AUTO_RETRY_SECONDS = 60*60*3
    MAX_AUTO_RETRY_COUNT = 5

    def __init__(self, sqlite_path):
        self.conn = get_sqlite_connection(sqlite_path)
        c = self.conn.cursor()
        c.execute(self.FAILURES_TABLE_CREATE)
        c.execute(self.OLD_FAILURES_TABLE_CREATE)
        c.execute(self.JOBS_TABLE_CREATE)
        c.execute(self.IMPORT_TABLE_CREATE)
        c.execute(self.RETRY_TABLE_CREATE)
        self.conn.commit()
        c.close()

    def _has_failed(self, c, package):
        rows = c.execute(self.FAILURES_TABLE_FIND, (package,)).fetchall()
        return (len(rows) > 0)

    def _set_failure(self, c, package, reason, now):
        if self._has_failed(c, package):
            c.execute('update %s set package=?, reason=?, when_failed=? '
                    'where package=?' % self.FAILURES_TABLE,
                    (package, reason, now, package))
        else:
            c.execute('insert into %s values (?, ?, ?)'
                    % self.FAILURES_TABLE, (package, reason, now))

    def add_jobs_for_interrupted(self):
        c = self.conn.cursor()
        try:
            rows = c.execute('select * from %s where reason=?'
                    % self.FAILURES_TABLE, (running_sentinel,)).fetchall()
            for row in rows:
                c.execute(self.FAILURES_TABLE_DELETE, (row[0],))
                self._add_job(c, row[0], self.JOB_TYPE_PRIORITY)
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def _start_package(self, c, package):
        ret = None
        rows = c.execute(self.JOBS_TABLE_FIND, (package, 1)).fetchall()
        now = datetime.datetime.utcnow()
        first = now
        job_id = 0
        for row in rows:
            c.execute(self.JOBS_TABLE_UPDATE,
                    (row[1], 0, row[3], row[4], first, row[6], row[0]))
            first = None
            job_id = row[0]
        if self._has_failed(c, package):
            job_id = None
        else:
            c.execute('insert into %s values (?, ?, ?, 0)'
                      % self.FAILURES_TABLE,
                      (package, running_sentinel, now))
            ret = job_id
        return ret

    def start_package(self, package):
        c = self.conn.cursor()
        try:
            ret = self._start_package(c, package)
            self.conn.commit()
            return ret
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def finish_job(self, package, job_id, success, output):
        c = self.conn.cursor()
        try:
            now = datetime.datetime.utcnow()
            if job_id > 0:
                row = c.execute(self.JOBS_TABLE_SELECT_JOB,
                                (job_id,)).fetchone()
                if row is not None:
                    c.execute(self.JOBS_TABLE_UPDATE,
                            (row[1], 0, row[3], row[4], row[5], now, row[0]))
            if success:
                row = c.execute(self.FAILURES_TABLE_FIND, (package,)).fetchone()
                if row is not None:
                    c.execute('delete from %s where package=?'
                            % self.JOBS_TABLE, (package,))
                c.execute(self.FAILURES_TABLE_DELETE, (package,))
                c.execute(self.OLD_FAILURES_TABLE_DELETE, (package,))
            else:
                self._set_failure(c, package, output, now)
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def _add_job(self, c, package, job_type):
        c.execute(self.JOBS_TABLE_INSERT, (None, package, 1, job_type,
                    datetime.datetime.utcnow(), None, None))

    def add_jobs(self, packages, job_type):
        c = self.conn.cursor()
        try:
            for package in packages:
                self._add_job(c, package, job_type)
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def last_import_time(self):
        c = self.conn.cursor()
        try:
            row = c.execute('select * from %s order by import desc'
                    % self.IMPORT_TABLE).fetchone()
            if row is not None:
                return row[0]
            return row
        finally:
            self.conn.rollback()
            c.close()

    def add_import_jobs(self, packages, newest_published):
        c = self.conn.cursor()
        try:
            for package in packages:
                self._add_job(c, package, self.JOB_TYPE_NEW)
            c.execute('insert into %s values (?)' % self.IMPORT_TABLE,
                    (newest_published,))
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def next_job(self):
        c = self.conn.cursor()
        try:
            job_id = 0
            package = None
            found = False
            while not found:
                row = c.execute(self.JOBS_TABLE_NEXT).fetchone()
                if row is not None:
                    package = row[1]
                    job_id = self._start_package(c, package)
                    if job_id is None:
                        c.execute(self.JOBS_TABLE_UPDATE,
                                (row[1], 0, row[3], row[4], row[5],
                                 row[6], row[0]))
                    else:
                        found = True
                else:
                    found = True
                    job_id = 0
                    package = None
            self.conn.commit()
            return (job_id, package)
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def failure_reason(self, package):
        c = self.conn.cursor()
        try:
            row = c.execute(self.FAILURES_TABLE_FIND, (package,)).fetchone()
            if row is None:
                return row
            return row[1]
        finally:
            self.conn.rollback()
            c.close()

    def failure_signature(self, raw_reason):
        trace = raw_reason.splitlines()
        sig = ''
        if len(trace) == 1:
            if trace[0] == running_sentinel:
                return None
            # sometimes, Python exceptions do not have file references
            m = re.match('(\w+): ', trace[0])
            if m:
                return m.group(1)
            else:
                return trace[0].strip().replace("\n", " ")
        elif len(trace) < 3:
            return " ".join(trace).strip().replace("\n", " ")

        for l in trace:
            if l.startswith('  File'):
                sig += ':' + l.split()[-1]

        return trace[-1].split(':')[0].replace("\n", " ") + sig

    def retry(self, package, force=False, priority=False, auto=False,
            all=False):
        c = self.conn.cursor()
        try:
            row = c.execute(self.FAILURES_TABLE_FIND, (package,)).fetchone()
            if row is None:
                if not force:
                    return False
                job_type = self.JOB_TYPE_RETRY
                if priority:
                    job_type = self.JOB_TYPE_PRIORITY
                self._add_job(c, package, job_type)
            else:
                raw_reason = row[1]
                raw_reason = raw_reason.encode("ascii", "replace")
                sig = self.failure_signature(raw_reason)
                self._retry(c, package, sig, row[2], priority=priority)
                if auto and sig != None:
                    row = c.execute(self.RETRY_TABLE_SELECT, (sig,)).fetchone()
                    if row is None:
                        c.execute(self.RETRY_TABLE_INSERT, (sig,))
                if all and sig != None:
                    rows = c.execute('select * from %s'
                            % self.FAILURES_TABLE).fetchall()
                    for row in rows:
                        this_raw_reason = row[1].encode("ascii", "replace")
                        this_sig = self.failure_signature(raw_reason)
                        if this_sig == sig:
                            self._retry(c, row[0], sig, row[2],
                                    priority=priority)
            self.conn.commit()
            return True
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def _retry(self, c, package, signature, timestamp, priority=False):
        row = c.execute(self.OLD_FAILURES_TABLE_FIND, (package,)).fetchone()
        if row is not None:
            if signature == row[1]:
                if row[4] > self.MAX_AUTO_RETRY_COUNT:
                    print ("Warning: %s has failed %d times in the same way"
                            % (package, row[4]))
                c.execute('update %s set package=?, reason=?, '
                        'when_failed=?, last_failed=?, failure_count=? '
                        'where package=?'
                        % self.OLD_FAILURES_TABLE,
                        (package, signature, timestamp, timestamp, row[4]+1,
                         package))
            else:
                c.execute('update %s set package=?, reason=?, when_failed=?, '
                        'last_failed=?, failure_count=? where package=?'
                        % self.OLD_FAILURES_TABLE,
                        (package, signature, timestamp, timestamp, 1, package))
        else:
            c.execute('insert into %s values (?, ?, ?, ?, ?)'
                    % self.OLD_FAILURES_TABLE,
                    (package, signature, timestamp, timestamp, 1))
        c.execute(self.FAILURES_TABLE_DELETE, (package,))
        job_type = self.JOB_TYPE_RETRY
        if priority:
            job_type = self.JOB_TYPE_PRIORITY
        self._add_job(c, package, job_type)

    def _attempt_retry(self, c, info):
        row = c.execute(self.RETRY_TABLE_SELECT, (info.signature,)).fetchone()
        if row is None:
            return False
        info.auto_retry = True
        info.auto_retry_time = info.timestamp + datetime.timedelta(0,
                self.AUTO_RETRY_SECONDS)
        if info.auto_retry_time > datetime.datetime.utcnow():
            return False
        row = c.execute(self.OLD_FAILURES_TABLE_FIND, (info.name,)).fetchone()
        if row is not None and row[1] == info.signature:
            info.failure_count += row[4]
            if row[4] > self.MAX_AUTO_RETRY_COUNT:
                info.auto_retry_masked = True
                return False
        self._retry(c, info.name, info.signature, info.timestamp)

    def summarise_failures(self):
        package_info = []
        c = self.conn.cursor()
        try:
            rows = c.execute('select * from %s'
                    % self.FAILURES_TABLE).fetchall()
            db_failures = [(row[0], row[1], row[2]) for row in rows]
            for row in c.execute(self.JOBS_TABLE_NEXT).fetchall():
                included = False
                for info in package_info:
                    if info.name == row[1]:
                        included = True
                if included:
                    continue
                package_info.append(PackageStatus(row[1]))
                package_info[-1].queued = True
            reasons = {}
            for package, raw_reason, ts in db_failures:
                package_info.append(PackageStatus(package))
                if raw_reason == running_sentinel:
                    package_info[-1].running = True
                    package_info[-1].timestamp = ts
                    continue
                raw_reason = raw_reason.encode("ascii", "replace")
                sig = self.failure_signature(raw_reason)
                if sig is None:
                    continue
                else:
                    info = package_info[-1]
                    info.failure_count = 1
                    info.signature = sig
                    info.raw_reason = raw_reason
                    info.timestamp = ts
                    if self._attempt_retry(c, info):
                        continue
                    reasons.setdefault(sig, [])
                    reasons[sig].append(info)
            self.conn.commit()
            return (reasons, package_info)
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def unemailed_failures(self):
        c = self.conn.cursor()
        try:
            rows = c.execute('select * from %s where emailed=0'
                    % self.FAILURES_TABLE).fetchall()
            db_failures = [(row[0], row[1], row[2]) for row in rows]
            return db_failures
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def set_failures_emailed(self, failures):
        c = self.conn.cursor()
        try:
            for failure in failures:
                c.execute('update %s set emailed=1 where package=? '
                        '' % (self.FAILURES_TABLE),
                        (failure[0], ))
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()


class CommitDatabase(object):

    COMMIT_TABLE = "commits"
    COMMIT_TABLE_CREATE = '''create table if not exists %s
                     (package text constraint nonull NOT NULL,
                      constraint isprimary PRIMARY KEY
                            (package))''' % COMMIT_TABLE
    COMMIT_TABLE_FIND = '''select * from %s where package=?''' % COMMIT_TABLE
    COMMIT_TABLE_INSERT = '''insert into %s values (?)''' % COMMIT_TABLE
    COMMIT_TABLE_DELETE = "delete from %s where package=?" % COMMIT_TABLE

    def __init__(self, sqlite_path, package):
        self.package = package
        self.conn = get_sqlite_connection(sqlite_path)
        c = self.conn.cursor()
        c.execute(self.COMMIT_TABLE_CREATE)
        self.conn.commit()
        c.close()

    def has_commit_started(self):
        c = self.conn.cursor()
        try:
            rows = c.execute(self.COMMIT_TABLE_FIND, (self.package,)).fetchall()
            return len(rows) > 0
        finally:
            self.conn.rollback()
            c.close()

    def start_commit(self):
        c = self.conn.cursor()
        try:
            rows = c.execute(self.COMMIT_TABLE_FIND, (self.package,)).fetchall()
            assert len(rows) == 0, "Already started commit"
            c.execute(self.COMMIT_TABLE_INSERT, (self.package,))
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def finish_commit(self):
        c = self.conn.cursor()
        try:
            rows = c.execute(self.COMMIT_TABLE_FIND, (self.package,)).fetchall()
            assert len(rows) == 1, "Commit not started"
            c.execute(self.COMMIT_TABLE_DELETE, (self.package,))
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()


class PackageDatabase(object):

    PACKAGE_TABLE = "packages"
    PACKAGE_TABLE_CREATE = '''create table if not exists %s
                     (package text constraint nonull NOT NULL,
                      main integer constraint nonull NOT NULL,
                      constraint isprimary PRIMARY KEY
                            (package))''' % PACKAGE_TABLE
    PACKAGE_TABLE_FIND = '''select * from %s where package=?''' % PACKAGE_TABLE
    PACKAGE_TABLE_INSERT = '''insert into %s values (?, ?)''' % PACKAGE_TABLE
    PACKAGE_TABLE_UPDATE = '''update %s set package=?, main=? where
                    package=?''' % PACKAGE_TABLE
    PACKAGE_TABLE_FIND_MAIN = '''select package from %s where main=1''' % PACKAGE_TABLE
    PACKAGE_TABLE_FIND_UNIVERSE = '''select package from %s where main=0''' % PACKAGE_TABLE

    UPDATE_TABLE = "packages_update"
    UPDATE_TABLE_CREATE = '''create table if not exists %s
                    (anupdate timestamp constraint nonull NOT NULL)
                    ''' % UPDATE_TABLE

    def __init__(self, sqlite_path):
        self.conn = get_sqlite_connection(sqlite_path)
        c = self.conn.cursor()
        c.execute(self.PACKAGE_TABLE_CREATE)
        c.execute(self.UPDATE_TABLE_CREATE)
        self.conn.commit()
        c.close()

    def last_update(self):
        c = self.conn.cursor()
        try:
            row = c.execute('select * from %s order by anupdate desc'
                    % self.UPDATE_TABLE).fetchone()
            if row is not None:
                return row[0]
            return row
        finally:
            self.conn.rollback()
            c.close()

    def update_packages(self, package_dict):
        c = self.conn.cursor()
        try:
            for (package, main) in package_dict.items():
                if len(c.execute(self.PACKAGE_TABLE_FIND, (package,)
                            ).fetchall()) > 0:
                    c.execute(self.PACKAGE_TABLE_UPDATE,
                            (package, main, package))
                else:
                    c.execute(self.PACKAGE_TABLE_INSERT, (package, main))
            c.execute('insert into %s values (?)' % self.UPDATE_TABLE,
                    (datetime.datetime.utcnow(),))
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def list_packages_in_main(self):
        c = self.conn.cursor()
        try:
            rows = c.execute(self.PACKAGE_TABLE_FIND_MAIN).fetchall()
            return set([row[0] for row in rows])
        finally:
            self.conn.rollback()
            c.close()

    def get_one(self, these_ones):
        c = self.conn.cursor()
        try:
            rows = c.execute(self.PACKAGE_TABLE_FIND_MAIN).fetchall()
            count = len(rows)
            for row in rows:
                if row[0] not in these_ones:
                    return row[0]
            rows = c.execute(self.PACKAGE_TABLE_FIND_UNIVERSE).fetchall()
            count += len(rows)
            if count == 0:
                raise StopIteration
            for row in rows:
                if row[0] not in these_ones:
                    return row[0]
        finally:
            self.conn.rollback()
            c.close()


class RevidDatabase(object):

    REVID_TABLE = "revids"
    REVID_TABLE_CREATE = '''create table if not exists %s
                     (package text constraint nonull NOT NULL,
                      version text constraint nonull NOT NULL,
                      suite text constraint nonull NOT NULL,
                      revid text constraint nonull NOT NULL,
                      testament text constraint nonull NOT NULL,
                      constraint isprimary PRIMARY KEY
                            (package, version, suite))'''
    REVID_TABLE_FIND = '''select * from %s where version=?
                          and package=? and suite=?'''
    REVID_TABLE_FIND_SUITE = '''select version from %s where package=?
                                and suite=?'''
    REVID_WORKING_TABLE = "revids_working"
    REVID_TABLE_FIND_PACKAGE = '''select * from %s where package=?'''
    REVID_TABLE_INSERT = "insert into %s values (?, ?, ?, ?, ?)"
    DELETE_WORKING = "delete from %s where package=?" % REVID_WORKING_TABLE

    SUFFIX_TABLE = "suffixes"
    SUFFIX_TABLE_CREATE = '''create table if not exists %s
                    (package text constraint nonull NOT NULL,
                     suffix text constraint nonull NOT NULL,
                     constraint isprimary PRIMARY KEY (package))''' % SUFFIX_TABLE


    def __init__(self, sqlite_path, package):
        self.package = package
        self.conn = get_sqlite_connection(sqlite_path)
        c = self.conn.cursor()
        c.execute(self.REVID_TABLE_CREATE % self.REVID_TABLE)
        c.execute(self.REVID_TABLE_CREATE % self.REVID_WORKING_TABLE)
        c.execute(self.SUFFIX_TABLE_CREATE)
        self.conn.commit()
        c.close()
        self.outstanding_marks = {}

    def _memory_rows(self):
        return self.outstanding_marks.keys()

    def _working_rows(self):
        c = self.conn.cursor()
        try:
            return c.execute(self.REVID_TABLE_FIND_PACKAGE
                    % self.REVID_WORKING_TABLE, (self.package,)).fetchall()
        finally:
            self.conn.rollback()
            c.close()

    def _saved_rows(self):
        c = self.conn.cursor()
        try:
            return c.execute(self.REVID_TABLE_FIND_PACKAGE
                    % self.REVID_TABLE, (self.package,)).fetchall()
        finally:
            self.conn.rollback()
            c.close()

    def is_marked(self, version, suite):
        c = self.conn.cursor()
        try:
            rows = c.execute(self.REVID_TABLE_FIND % self.REVID_TABLE,
                    (str(version), self.package, suite)).fetchall()
            rows += c.execute(self.REVID_TABLE_FIND % self.REVID_WORKING_TABLE,
                    (str(version), self.package, suite)).fetchall()
            rows += [foo for foo in self.outstanding_marks
                        if foo == (str(version), suite)]
            assert len(rows) < 2, "Multiple versions for package/suite?"
            if len(rows) > 0:
                return True
            return False
        finally:
            self.conn.rollback()
            c.close()

    def last_marked_version(self, suite):
        c = self.conn.cursor()
        try:
            versions = [a[0] for a in c.execute(self.REVID_TABLE_FIND_SUITE
                    % self.REVID_WORKING_TABLE,
                    (self.package, suite)).fetchall()]
            versions += [a[0] for a in c.execute(self.REVID_TABLE_FIND_SUITE
                    % self.REVID_TABLE,
                    (self.package, suite)).fetchall()]
            versions += [str(foo[0]) for foo in self.outstanding_marks
                            if foo[1] == suite]
            if len(versions) > 0:
                versions.sort(key=lambda x: changelog.Version(x))
                return versions[-1]
            return None
        finally:
            self.conn.rollback()
            c.close()

    def check(self, version, revid, suite, branch):
        branch.repository.lock_read()
        try:
            tment = testament.StrictTestament3.from_revision(
                    branch.repository, revid)
            sha = tment.as_sha1()
        finally:
            branch.repository.unlock()
        c = self.conn.cursor()
        try:
            rows = c.execute(self.REVID_TABLE_FIND % self.REVID_TABLE,
                    (str(version), self.package, suite)).fetchall()
            rows += c.execute(self.REVID_TABLE_FIND % self.REVID_WORKING_TABLE,
                    (str(version), self.package, suite)).fetchall()
            if (version, suite) in self.outstanding_marks:
                rows += dict(revid=self.outstanding_marks[(str(version), suite)][0],
                        testament=self.outstanding_marks[(str(version), suite)][1])
            if len(rows) < 1:
                return False
            assert len(rows) < 2, "Multiple versions for a package/suite?"
            row = rows[0]
            if (revid, sha) != (row['revid'], row['testament']):
                assert False, ("%s != %s for %s %s, something has changed"
                     % (str((revid, sha)), str((row['revid'],
                                 row['testament'])), self.package,
                         str(version)))
            return True
        finally:
            self.conn.rollback()
            c.close()

    def _commit(self, c):
        rows = c.execute(self.REVID_TABLE_FIND_PACKAGE
                % self.REVID_WORKING_TABLE, (self.package,)).fetchall()
        if len(rows) > 0:
            def get_values():
                for row in rows:
                    yield tuple([row[i] for i in range(5)])
            c.executemany(self.REVID_TABLE_INSERT % self.REVID_TABLE,
                    get_values())
            c.execute(self.DELETE_WORKING, (self.package,))

    def commit(self):
        c = self.conn.cursor()
        try:
            self._commit(c)
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def cleanup_last_run(self, cleanup_cb, push_cb):
        c = self.conn.cursor()
        try:
            rows = c.execute(self.REVID_TABLE_FIND_PACKAGE
                    % self.REVID_WORKING_TABLE, (self.package,)).fetchall()
            if len(rows) < 1:
                mutter("Nothing to cleanup")
                cleanup_cb()
            else:
                self.conn.rollback()
                push_cb()
                self._commit(c)
                self.conn.commit()
                cleanup_cb()
        finally:
            self.conn.rollback()
            c.close()

    def discard_last_run(self):
        c = self.conn.cursor()
        try:
            c.execute(self.DELETE_WORKING, (self.package,))
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def mark(self, version, revid, suite, branch):
        branch.lock_read()
        try:
            tment = testament.StrictTestament3.from_revision(branch.repository, revid)
            sha = tment.as_sha1()
        finally:
            branch.unlock()
        self.outstanding_marks[(str(version), suite)] = (revid, sha)

    def commit_outstanding(self):
        c = self.conn.cursor()
        try:
            for version, suite in self.outstanding_marks:
                revid, sha = self.outstanding_marks[(version, suite)]
                rows = c.execute(self.REVID_TABLE_FIND % self.REVID_TABLE,
                        (str(version), self.package, suite)).fetchall()
                rows += c.execute(self.REVID_TABLE_FIND % self.REVID_WORKING_TABLE,
                        (str(version), self.package, suite)).fetchall()
                assert len(rows) < 1, "Trying to mark version %s in %s again" \
                    % (str(version), suite)
                c.execute(self.REVID_TABLE_INSERT % self.REVID_WORKING_TABLE,
                        (self.package, str(version), suite, revid, sha))
            self.outstanding_marks = {}
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def get_suffix(self):
        c = self.conn.cursor()
        try:
            rows = c.execute("select * from %s where package=?"
                    % self.SUFFIX_TABLE, (self.package,)).fetchall()
            if len(rows) < 1:
                return ""
            return rows[0][1]
        finally:
            self.conn.rollback()
            c.close()


class HistoryDatabase(object):

    FAILED_TABLE = "failed"
    FAILED_TABLE_CREATE = '''create table if not exists %s
                     (attime timestamp constraint nonull NOT NULL,
                      queue integer constraint nonull NOT NULL,
                      failed integer constraint nonull NOT NULL,
                      constraint isprimary PRIMARY KEY (attime))'''
    MAIN_FAILED_TABLE = "main_failed"

    def __init__(self, sqlite_path):
        self.conn = get_sqlite_connection(sqlite_path)
        c = self.conn.cursor()
        c.execute(self.FAILED_TABLE_CREATE % self.FAILED_TABLE)
        c.execute(self.FAILED_TABLE_CREATE % self.MAIN_FAILED_TABLE)
        self.conn.commit()
        c.close()

    def _set_counts(self, table, queued, failed):
        c = self.conn.cursor()
        try:
            rows = c.execute("insert into %s values (?, ?, ?)"
                    % table,
                    (datetime.datetime.utcnow(), queued, failed))
            self.conn.commit()
        except:
            self.conn.rollback()
            raise
        finally:
            c.close()

    def set_counts(self, queued, failed):
        self._set_counts(self.FAILED_TABLE, queued, failed)

    def set_main_counts(self, queued, failed):
        self._set_counts(self.MAIN_FAILED_TABLE, queued, failed)

    def _get_counts(self, table):
        c = self.conn.cursor()
        try:
            rows = c.execute("select * from %s"
                    % table).fetchall()
            return [(r[0], r[1], r[2]) for r in rows]
        finally:
            self.conn.rollback()
            c.close()

    def get_main_counts(self):
        return self._get_counts(self.MAIN_FAILED_TABLE)

    def get_counts(self):
        return self._get_counts(self.FAILED_TABLE)


class MemoryRevidDatabase(RevidDatabase):

    def __init__(self, package):
        self.package = package
        self._maps = {}

    def is_marked(self, version, suite):
        revid_map = self._get_map(suite)
        return str(version) in revid_map

    def last_marked_version(self, suite):
        revid_map = self._get_map(suite)
        versions = revid_map.keys()
        if len(versions) > 0:
            versions.sort(key=lambda x: changelog.Version(x))
            return versions[-1]
        return None

    def check(self, version, revid, suite, branch):
        revid_map = self._get_map(suite)
        branch.repository.lock_read()
        try:
            tment = testament.StrictTestament3.from_revision(branch.repository,
                                                             revid)
            sha = tment.as_sha1()
            if not str(version) in revid_map:
                return False
            if (revid, sha) != revid_map[str(version)]:
                mutter("%s != %s for %s %s, something has changed"
                       % (str((revid, sha)), str(revid_map[str(version)]),
                          self.package, str(version)))
                return False
            return True
        finally:
            branch.repository.unlock()

    def commit(self):
        raise NotImplementedError(self.finish_commit)

    def cleanup_last_run(self, cleanup_cb, push_cb):
        pass

    def discard_last_run(self):
        pass

    def _get_map(self, suite):
        self._maps.setdefault(suite, {})
        return self._maps[suite]

    def _save_map(self, suite, map):
        self._maps[suite] = map

    def mark(self, version, revid, suite, branch):
        revid_map = self._get_map(suite)
        branch.lock_read()
        try:
            tment = testament.StrictTestament3.from_revision(branch.repository,
                                                             revid)
            sha = tment.as_sha1()
        finally:
            branch.unlock()
        assert str(version) not in revid_map, \
              "Asked to mark %s %s %s, but it is already marked" \
                    % (self.package, version, suite)
        revid_map[str(version)] = (revid, sha)
        self._save_map(suite, revid_map)

    def commit_outstanding(self):
        pass


class BranchStore(object):

    def __init__(self, package, lp, suffix):
        self.package = package
        self.lp = lp
        self.suffix = suffix
        self._series_cache = {}
        self._sp_cache = {}

    def _branch_location(self, distro, release, pocket, readonly=False):
        lp_branch = None
        lp_series = self._get_lp_series(distro, release)
        scheme = "bzr+ssh://"
        if readonly:
            scheme = "http://"
        if lp_series is not None:
            lp_sp = self._get_lp_source_package(distro, release)
            lp_branch = lp_call(lp_sp.getBranch, pocket=self._lp_pocket(pocket))
        if lp_branch is None:
            distro, suite = self._translate_suite(distro,
                                                  make_suite(release, pocket))
            return ("%sbazaar.launchpad.net/~ubuntu-branches/%s/"
                    "%s/%s/%s%s" % (scheme, distro, release,
                                    self.package, suite, self.suffix))
        else:
            path = urlparse.urlparse(lp_branch.self_link).path
            lp_branch_path = path[path[1:].index('/')+1:]
            return "%sbazaar.launchpad.net%s" % (scheme, lp_branch_path)

    def side_branch_location(self, distro, release, suite):
        distro, suite = self._translate_suite(distro, suite)
        now = datetime.datetime.utcnow()
        now_str = now.strftime("%Y%m%d%H%M")
        return ("bzr+ssh://bazaar.launchpad.net/~ubuntu-branches/%s/"
                "%s/%s/%s" % (distro, release, self.package, suite + "-%s"
                              % now_str))


    def get_branch(self, importp, possible_transports=None, readonly=False):
        return self.get_branch_parts(
            importp.distro, importp.release, importp.pocket,
            possible_transports=possible_transports, readonly=readonly)

    def get_branch_parts(self, distro, release, pocket,
                         possible_transports=None, readonly=False):
        try:
            return branch.Branch.open(
                self._branch_location(distro, release, pocket,
                                      readonly=readonly),
                possible_transports=possible_transports)
        except errors.NotBranchError:
            return None

    def push_location(self, distro, release, pocket):
        return self._branch_location(distro, release, pocket)

    def _set_status(self, distro, release, branch, _retry_count=0):
        # series statuses are:
        # Experimental
        # Active Development
        # Pre-release Freeze
        # Current Stable Release
        # Supported
        # Obsolete
        # Future
        #
        # branch statuses are:
        # Experimental
        # Development
        # Mature
        # Merged
        # Abandoned
        series = self._get_lp_series(distro, release)
        if series.status in ('Experimental', 'Active Development',
            'Pre-release Freeze', 'Future'):
            branch.lifecycle_status = 'Development'
        elif series.status in ('Current Stable Release', 'Supported'):
            branch.lifecycle_status = 'Mature'
        elif series.status in ('Obsolete',):
            branch.lifecycle_status = 'Abandoned'
        try:
            lp_call(branch.lp_save)
        except HTTPError, e:
            if e.response.status == 412 and _retry_count < 3:
                time.sleep(1)
                self._set_status(distro, release,
                                 lp_call(self.lp.load, branch.self_link),
                                 _retry_count=_retry_count+1)
            else:
                raise

    def _set_reviewer(self, branch, _retry_count=0):
        ubuntu_dev = lp_call(operator.getitem, self.lp.people, 'ubuntu-dev')
        branch.reviewer = ubuntu_dev
        try:
            lp_call(branch.lp_save)
        except HTTPError, e:
            if e.response.status == 412 and _retry_count < 3:
                time.sleep(1)
                self._set_reviewer(lp_call(self.lp.load, branch.self_link),
                        _retry_count=_retry_count+1)
            else:
                raise

    def _get_lp_source_package(self, distro, release):
        if (distro, release) in self._sp_cache:
            return self._sp_cache[(distro, release)]
        lp_series = self._get_lp_series(distro, release)
        lp_sp = lp_call(lp_series.getSourcePackage, name=self.package)
        self._sp_cache[(distro, release)] = lp_sp
        return lp_sp

    def _get_lp_series(self, distro, release):
        if (distro, release) in self._series_cache:
            return self._series_cache[(distro, release)]
        if distro == "debian" and release in ["woody", "sarge", "etch"]:
            lp_series = None
        else:
            lp_distro = lp_call(operator.getitem, self.lp.distributions, distro)
            lp_series = lp_call(lp_distro.getSeries, name_or_version=release)
        self._series_cache[(distro, release)] = lp_series
        return lp_series

    def _lp_pocket(self, pocket):
        return pocket[0].upper() + pocket[1:]

    def _translate_suite(self, distro, suite):
        if distro == "debian":
            if suite == "oldstable":
                suite = "etch"
            elif suite == "stable":
                suite = "lenny"
            elif suite == "testing":
                suite = "squeeze"
            elif suite == "unstable":
                suite = "sid"
        return distro, suite

    def _branch_url(self, distro, release, pocket):
        lp_sp = self._get_lp_source_package(distro, release)
        lp_branch = lp_call(lp_sp.getBranch, pocket=self._lp_pocket(pocket))
        if lp_branch is None:
            distro, suite = self._translate_suite(
                distro, make_suite(release, pocket))
            return (str(self.lp._root_uri) + '~ubuntu-branches/%s/%s/%s/%s'
                    % (distro, release, self.package, suite))
        else:
            return lp_branch.self_link

    def set_official(self, distro, release, pocket, set_status=False,
            set_reviewer=False):
        lp_sp = self._get_lp_source_package(distro, release)
        branch_url = self._branch_url(distro, release, pocket)
        lp_branch = lp_call(self.lp.load, branch_url)
        lp_call(lp_sp.setBranch, branch=lp_branch,
                pocket=self._lp_pocket(pocket))
        if set_status:
            self._set_status(distro, release, lp_branch)
        if set_reviewer:
            self._set_reviewer(lp_branch)

    def _set_official_direct(self, distro, release, pocket, branch_url):
        lp_sp = self._get_lp_source_package(distro, release)
        if branch_url is not None:
            lp_branch = lp_call(self.lp.load, branch_url)
        else:
            lp_branch = None
        lp_call(lp_sp.setBranch, branch=lp_branch,
                pocket=self._lp_pocket(pocket))


class LocalShadowBranchStore(BranchStore):

    def __init__(self, package, lp, suffix):
        BranchStore.__init__(self, package, lp, suffix)
        self.branches_dir = os.path.join(localbranches_dir, package)
        if not os.path.exists(self.branches_dir):
            os.makedirs(self.branches_dir)
            format = bzrdir.format_registry.make_bzrdir('2a')
            to_transport = transport.get_transport(self.branches_dir)
            newdir = format.initialize_on_transport(to_transport)
            repo = newdir.create_repository(shared=True)
            repo.set_make_working_trees(False)

    def side_branch_location(self, distro, release, suite):
        distro, suite = self._translate_suite(distro, suite)
        now = datetime.datetime.utcnow()
        now_str = now.strftime("%Y%m%d%H%M")
        return os.path.join(self.branches_dir, "%s-%s-%s"
                            % (distro, suite, now_str))

    def get_branch_parts(self, distro, release, pocket,
                         possible_transports=None, readonly=False):
        local_location = self.push_location(distro, release, pocket)
        try:
            return branch.Branch.open(local_location,
                                      possible_transports=possible_transports)
        except errors.NotBranchError:
            try:
                br_from = branch.Branch.open(
                    self._branch_location(distro, release, pocket,
                                          readonly=readonly),
                    possible_transports=possible_transports)
                br_from.lock_read()
                try:
                    dir = br_from.bzrdir.sprout(local_location)
                    new_br = dir.open_branch()
                    tag._merge_tags_if_possible(br_from, new_br)
                finally:
                    br_from.unlock()
                return new_br
            except errors.NotBranchError:
                return None

    def push_location(self, distro, release, pocket):
        distro, suite = self._translate_suite(
            distro, make_suite(release, pocket))
        return os.path.join(self.branches_dir, "%s-%s%s"
                            % (distro, suite, self.suffix))

    def set_official(self, distro, release, pocket, set_status=False,
            set_reviewer=False):
        pass # No-op in this implementation


class MemoryBranchStore(BranchStore):

    def __init__(self, package, lp, suffix):
        self.package = package

    def side_branch_location(self, distro, release, suite):
        distro, suite = self._translate_suite(distro, suite)
        now = datetime.datetime.utcnow()
        now_str = now.strftime("%Y%m%d%H%M")
        return os.path.join(updates_dir, self.package, suite + "-%s" % now_str)


    def get_branch_parts(self, distro, release, pocket,
            possible_transports=None, readonly=False):
        return None

    def push_location(self, distro, release, pocket):
        raise NotImplementedError(self.push_location)

    def set_official(self, distro, release, pocket, set_status=False,
            set_reviewer=False):
        raise NotImplementedError(self.set_official)


def get_collection_slice(collection, start, stop, max_size):
    existing_representation = collection._wadl_resource.representation
    if (existing_representation is not None
        and start < len(existing_representation['entries'])):
        # An optimization: the first page of entries has already
        # been loaded. This can happen if this collection is the
        # return value of a named operation, or if the client did
        # something like check the length of the collection.
        #
        # Either way, we've already made an HTTP request and
        # gotten some entries back. The client has requested a
        # slice that includes some of the entries we already have.
        # In the best case, we can fulfil the slice immediately,
        # without making another HTTP request.
        #
        # Even if we can't fulfil the entire slice, we can get one
        # or more objects from the first page and then have fewer
        # objects to retrieve from the server later. This saves us
        # time and bandwidth, and it might let us save a whole
        # HTTP request.
        entry_page = existing_representation['entries']

        first_page_size = len(entry_page)
        entry_dicts = entry_page[start:stop]
        page_url = existing_representation.get('next_collection_link')
    else:
        # No part of this collection has been loaded yet, or the
        # slice starts beyond the part that has been loaded. We'll
        # use our secret knowledge of lazr.restful to set a value for
        # the ws.start variable. That way we start reading entries
        # from the first one we want.
        first_page_size = max_size + 1
        entry_dicts = []
        page_url = collection._with_url_query_variable_set(
            collection._wadl_resource.url, 'ws.start', start)

    desired_size = stop-start
    more_needed = desired_size - len(entry_dicts)

    # Iterate over pages until we have the correct number of entries.
    while more_needed > 0 and page_url is not None:
        if (first_page_size is not None
                and more_needed > 0 and more_needed < first_page_size):
            # An optimization: it's likely that we need less than
            # a full page of entries, because the number we need
            # is less than the size of the first page we got.
            # Instead of requesting a full-sized page, we'll
            # request only the number of entries we think we'll
            # need. If we're wrong, there's no problem; we'll just
            # keep looping.
            page_url = collection._with_url_query_variable_set(
                page_url, 'ws.size', more_needed)
        representation = simplejson.loads(
            unicode(collection._root._browser.get(page_url)))
        current_page_entries = representation['entries']
        entry_dicts += current_page_entries[:more_needed]
        more_needed = desired_size - len(entry_dicts)

        page_url = representation.get('next_collection_link')
        if page_url is None:
            # We've gotten the entire collection; there are no
            # more entries.
            break
        if first_page_size is None:
            first_page_size = len(current_page_entries)


    # Convert entry_dicts into a list of Entry objects.
    return [resource for resource
            in collection._convert_dicts_to_entries(entry_dicts)]


def iterate_collection(collection, size=20):
    for i in xrange(0, sys.maxint, size):
        slice = get_collection_slice(collection, i, i+size, size)
        if len(slice) == 0:
            break
        for item in slice:
            yield item


def call_with_limited_size(operation, size=20, *args, **kwargs):
    """Invoke the method and process the result."""
    if len(args) > 0:
        raise TypeError('Method must be called with keyword args.')
    http_method = operation.wadl_method.name
    args = operation._transform_resources_to_links(kwargs)
    request = operation.wadl_method.request

    def _with_url_query_variable_set(url, variable, new_value):
        """A helper method to set a query variable in a URL."""
        parts = urlparse.urlparse(url)
        if parts.query is None:
            params = {}
        else:
            params = cgi.parse_qs(parts.query)
        params[variable] = str(new_value)
        query = urllib.urlencode(params, True)
        return urlparse.urlunparse((parts.scheme, parts.netloc, parts.path,
                    parts.params, query, parts.fragment))

    for key, value in args.items():
        # Certain parameter values should not be JSON-encoded:
        # binary parameters (because they can't be JSON-encoded)
        # and option values (because JSON-encoding them will screw
        # up wadllib's parameter validation). The option value thing
        # is a little hacky, but it's the best solution for now.
        if not isinstance(value, basestring):
            args[key] = simplejson.dumps(value)
    if http_method in ('get', 'head', 'delete'):
        url = operation.wadl_method.build_request_url(**args)
        in_representation = ''
        extra_headers = {}
        url = _with_url_query_variable_set(url, 'ws.size', size)
    else:
        url = operation.wadl_method.build_request_url()
        (media_type,
         in_representation) = operation.wadl_method.build_representation(
            **args)
        extra_headers = { 'Content-type' : media_type }
    response, content = operation.root._browser._request(
        url, in_representation, http_method, extra_headers=extra_headers)

    if response.status == 201:
        return operation._handle_201_response(url, response, content)
    else:
        if http_method == 'post':
            # The method call probably modified this resource in
            # an unknown way. Refresh its representation.
            operation.resource.lp_refresh()
        return operation._handle_200_response(url, response, content)


def get_debian_ubuntu_branches(lp, temp_dir, package, bstore,
        ensure_ubuntu_local=False, possible_transports=None):
    if not os.path.exists(temp_dir):
        return (None, None, None)
    try:
        debian_b = branch.Branch.open(os.path.join(temp_dir,
                    default_debian_diff_release))
    except errors.NotBranchError:
        debian_b = bstore.get_branch_parts("debian",
                default_debian_diff_release, "release",
                possible_transports=possible_transports)
    if debian_b is None:
        # Not in unstable
        return (None, None, None)
    ubuntu_tree = None
    ubuntu_current_series = lp.distributions['ubuntu'].current_series.name
    try:
        ubuntu_b = branch.Branch.open(os.path.join(temp_dir,
                    ubuntu_current_series))
        ubuntu_tree = ubuntu_b.bzrdir.open_workingtree()
    except errors.NotBranchError:
        ubuntu_b = bstore.get_branch_parts("ubuntu", ubuntu_current_series,
                "release", possible_transports=possible_transports)
        if ubuntu_b is not None and ensure_ubuntu_local:
            to_transport = transport.get_transport(os.path.join(temp_dir,
                        ubuntu_current_series))
            old_ubuntu_b = ubuntu_b
            old_ubuntu_b.lock_read()
            try:
                revid = old_ubuntu_b.last_revision()
                to_transport.mkdir('.')
                abzrdir = old_ubuntu_b.bzrdir.sprout(to_transport.base,
                        revid, possible_transports=possible_transports)
                ubuntu_b = abzrdir.open_branch()
                tag._merge_tags_if_possible(old_ubuntu_b, ubuntu_b)
                ubuntu_tree = abzrdir.open_workingtree()
            finally:
                old_ubuntu_b.unlock()
    if ubuntu_b is None:
        # Not in Ubuntu
        return (None, None, None)
    return (debian_b, ubuntu_b, ubuntu_tree)


def generate_debian_diffs(lp, temp_dir, package, bstore,
        possible_transports=None):
    target_path = os.path.join(debian_diffs_dir, package)
    debian_b, ubuntu_b, _ = get_debian_ubuntu_branches(lp, temp_dir, package,
            bstore, possible_transports=possible_transports)
    if debian_b is None:
        if os.path.exists(target_path):
            os.unlink(target_path)
        return
    debian_tree = debian_b.basis_tree()
    ubuntu_tree = ubuntu_b.basis_tree()
    out = StringIO.StringIO()
    diff.show_diff_trees(debian_tree, ubuntu_tree, out, old_label='debian/',
            new_label='ubuntu/')
    if len(out.getvalue().strip()) > 0:
        f = open(target_path, 'wb')
        try:
            f.write(out.getvalue())
        finally:
            f.close()
    else:
        if os.path.exists(target_path):
            os.unlink(target_path)


def possibly_generate_debian_diffs(lp, temp_dir, package, bstore,
        possible_transports=None):
    if os.path.exists(os.path.join(debian_diffs_dir, package)):
        return
    generate_debian_diffs(lp, temp_dir, package, bstore,
            possible_transports=possible_transports)


def generate_ubuntu_merges(lp, temp_dir, package, bstore,
        possible_transports=None):
    debian_b, ubuntu_b, ubuntu_tree = get_debian_ubuntu_branches(lp,
            temp_dir, package, bstore, ensure_ubuntu_local=True,
            possible_transports=possible_transports)
    target_path = os.path.join(ubuntu_merge_dir, package)
    if debian_b is None:
        if os.path.exists(target_path):
            os.unlink(target_path)
        return
    debian_b.lock_read()
    try:
        ubuntu_tree.lock_write()
        try:
            out = StringIO.StringIO()
            debian_revid = debian_b.last_revision()
            ubuntu_revid = ubuntu_b.last_revision()
            pb = ui.ui_factory.nested_progress_bar()
            try:
                merger = merge.Merger.from_revision_ids(pb, ubuntu_tree,
                        debian_revid, other_branch=debian_b)
            except errors.UnrelatedBranches:
                out.write("Unrelated branches!")
                return
            merger.merge_type = merge.Merge3Merger
            if (merger.base_rev_id == merger.other_rev_id
                    and merger.other_rev_id is not None):
                if os.path.exists(target_path):
                    os.unlink(target_path)
                return
            if (merger.base_rev_id == ubuntu_revid):
                if os.path.exists(target_path):
                    os.unlink(target_path)
                return
            tree_merger = merger.make_merger()
            tt = tree_merger.make_preview_transform()
            try:
                result_tree = tt.get_preview_tree()
                tree_delta = result_tree.changes_from(merger.this_tree)
                delta.report_delta(out, tree_delta)
                if len(tree_merger.cooked_conflicts) > 0:
                    out.write("conflicts:\n")
                    for conflict in tree_merger.cooked_conflicts:
                        out.write('  %s\n' % conflict)
                out.write("\n\n")
                # TODO: better labels
                old_label = "ubuntu-current/"
                new_label = "ubuntu-merged/"
                out2 = StringIO.StringIO()
                diff.show_diff_trees(merger.this_tree, result_tree, out2,
                        old_label=old_label, new_label=new_label)
                f = open(target_path, 'wb')
                try:
                    f.write(out.getvalue())
                    f.write(out2.getvalue())
                finally:
                    f.close()
            finally:
                tt.finalize()
        finally:
            ubuntu_tree.unlock()
    finally:
        debian_b.unlock()


def possibly_generate_ubuntu_merges(lp, temp_dir, package, bstore,
        possible_transports=None):
    if os.path.exists(os.path.join(ubuntu_merge_dir, package)):
        return
    generate_ubuntu_merges(lp, temp_dir, package, bstore,
            possible_transports=possible_transports)


def find_earliest_merge(b, revid):
    for (stop_revid, depth, revno, eom) in b.iter_merge_sorted_revisions(
        start_revision_id=b.last_revision(), stop_revision_id=revid,
        stop_rule='include', direction='forward'):
        if depth == 0:
            return stop_revid
    return None


def load_explanations():
    explanations = {}
    if not os.path.exists(explanations_file):
        return explanations
    f = open(explanations_file)
    reasons = []
    explanation = ""
    try:
        for line in f:
            if line.startswith(" "):
                explanation += line
            else:
                if explanation != "":
                    for reason in reasons:
                        explanations[reason] = explanation
                    explanation = ""
                    reasons = []
                reasons.append(line[:-1])
        if explanation != "":
            for reason in reasons:
                explanations[reason] = explanation
        return explanations
    finally:
        f.close()


def update_list(release, component, source_url):
    target_dir = os.path.join(lists_dir, "dists", release, component, "source")
    if not os.path.exists(target_dir):
        os.makedirs(target_dir)
    target_path = os.path.join(target_dir, "Sources.gz")
    target_lcfile = target_path + ".lc"
    last_check = None
    if os.path.exists(target_lcfile):
        f = open(target_lcfile)
        try:
            line = f.read().strip()
            try:
                last_check = int(line)
            except ValueError:
                last_check = None
        finally:
            f.close()
    if last_check is not None:
        now = time.time()
        if now < last_check + (60*60):
            return
    target_tsfile = target_path + ".ts"
    ts = None
    if os.path.exists(target_tsfile):
        f = open(target_tsfile)
        try:
            ts = f.read().strip()
        finally:
            f.close()
    req = urllib2.Request(source_url)
    if ts is not None:
        req.add_header('If-Modified-Since', ts)
    opener = urllib2.build_opener()
    try:
        resp = opener.open(req)
    except urllib2.HTTPError, e:
        if e.code != 304:
            raise
    except urllib2.URLError, e:
        print "Error contacting %s" % source_url
        raise
    else:
        f = open(target_path, "wb")
        try:
            shutil.copyfileobj(resp, f)
        finally:
            f.close()
        ts = resp.headers.get('Last-Modified')
        f = open(target_tsfile, "wb")
        try:
            f.write("%s\n" % ts)
        finally:
            f.close()
        f = open(target_lcfile, "wb")
        try:
            f.write("%d\n" % time.time())
        finally:
            f.close()