'''
import utils
fullpath = self.poolfile.fullpath
- deb_file = open(fullpath, 'r')
- stanza = utils.deb_extract_control(deb_file)
- deb_file.close()
-
- return stanza
+ with open(fullpath, 'r') as deb_file:
+ return utils.deb_extract_control(deb_file)
def read_control_fields(self):
'''
stanza = self.read_control()
return apt_pkg.TagSection(stanza)
+ @property
+ def proxy(self):
+ session = object_session(self)
+ query = session.query(BinaryMetadata).filter_by(binary=self)
+ return MetadataProxy(session, query)
+
__all__.append('DBBinary')
@session_wrapper
################################################################################
class Keyring(object):
- gpg_invocation = "gpg --no-default-keyring --keyring %s" +\
- " --with-colons --fingerprint --fingerprint"
-
keys = {}
fpr_lookup = {}
if not self.keyring_id:
raise Exception('Must be initialized with database information')
- k = os.popen(self.gpg_invocation % keyring, "r")
+ cmd = ["gpg", "--no-default-keyring", "--keyring", keyring,
+ "--with-colons", "--fingerprint", "--fingerprint"]
+ p = daklib.daksubprocess.Popen(cmd, stdout=subprocess.PIPE)
+
key = None
need_fingerprint = False
- for line in k:
+ for line in p.stdout:
field = line.split(":")
if field[0] == "pub":
key = field[4]
self.fpr_lookup[field[9]] = key
need_fingerprint = False
+ r = p.wait()
+ if r != 0:
+ raise subprocess.CalledProcessError(r, cmd)
+
def import_users_from_ldap(self, session):
import ldap
cnf = Config()
l = ldap.open(LDAPServer)
if ca_cert_file:
- # TODO: This should request a new context and use
- # connection-specific options (i.e. "l.set_option(...)")
-
- # Request a new TLS context. If there was already one, libldap
- # would not change the TLS options (like which CAs to trust).
- #l.set_option(ldap.OPT_X_TLS_NEWCTX, True)
- ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_HARD)
- #ldap.set_option(ldap.OPT_X_TLS_CACERTDIR, None)
- ldap.set_option(ldap.OPT_X_TLS_CACERTFILE, ca_cert_file)
+ l.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_HARD)
+ l.set_option(ldap.OPT_X_TLS_CACERTFILE, ca_cert_file)
+ l.set_option(ldap.OPT_X_TLS_NEWCTX, True)
l.start_tls_s()
l.simple_bind_s("","")
self.contents_sha1 = signed_file.contents_sha1()
return self
+ def query(self, session):
+ return session.query(SignatureHistory).filter_by(fingerprint=self.fingerprint, signature_timestamp=self.signature_timestamp, contents_sha1=self.contents_sha1).first()
+
__all__.append('SignatureHistory')
################################################################################
fileset.add(name)
return fileset
+ @property
+ def proxy(self):
+ session = object_session(self)
+ query = session.query(SourceMetadata).filter_by(source=self)
+ return MetadataProxy(session, query)
+
__all__.append('DBSource')
@session_wrapper
# FIXME: This function fails badly if it finds more than 1 source package and
# its implementation is trivial enough to be inlined.
@session_wrapper
-def get_source_in_suite(source, suite, session=None):
+def get_source_in_suite(source, suite_name, session=None):
"""
- Returns a DBSource object for a combination of C{source} and C{suite}.
+ Returns a DBSource object for a combination of C{source} and C{suite_name}.
- B{source} - source package name, eg. I{mailfilter}, I{bbdb}, I{glibc}
- - B{suite} - a suite name, eg. I{unstable}
+ - B{suite_name} - a suite name, eg. I{unstable}
@type source: string
@param source: source package name
- @type suite: string
+ @type suite_name: string
@param suite: the suite name
@rtype: string
@return: the version for I{source} in I{suite}
"""
-
- q = get_suite(suite, session).get_sources(source)
+ suite = get_suite(suite_name, session)
+ if suite is None:
+ return None
try:
- return q.one()
+ return suite.get_sources(source).one()
except NoResultFound:
return None
def path(self):
return os.path.join(self.archive.path, 'dists', self.suite_name)
+ @property
+ def release_suite_output(self):
+ if self.release_suite is not None:
+ return self.release_suite
+ return self.suite_name
+
__all__.append('Suite')
@session_wrapper
@return: Suite object for the requested suite name (None if not present)
"""
+ # Start by looking for the dak internal name
q = session.query(Suite).filter_by(suite_name=suite)
+ try:
+ return q.one()
+ except NoResultFound:
+ pass
+ # Now try codename
+ q = session.query(Suite).filter_by(codename=suite)
+ try:
+ return q.one()
+ except NoResultFound:
+ pass
+
+ # Finally give release_suite a try
+ q = session.query(Suite).filter_by(release_suite=suite)
try:
return q.one()
except NoResultFound:
################################################################################
+class MetadataProxy(object):
+ def __init__(self, session, query):
+ self.session = session
+ self.query = query
+
+ def _get(self, key):
+ metadata_key = self.session.query(MetadataKey).filter_by(key=key).first()
+ if metadata_key is None:
+ return None
+ metadata = self.query.filter_by(key=metadata_key).first()
+ return metadata
+
+ def __contains__(self, key):
+ if self._get(key) is not None:
+ return True
+ return False
+
+ def __getitem__(self, key):
+ metadata = self._get(key)
+ if metadata is None:
+ raise KeyError
+ return metadata.value
+
+ def get(self, key, default=None):
+ try:
+ return self[key]
+ except KeyError:
+ return default
+
+################################################################################
+
class VersionCheck(ORMObject):
def __init__(self, *args, **kwargs):
pass
'obsolete_any_associations',
'obsolete_any_by_all_associations',
'obsolete_src_associations',
+ 'package_list',
'source_suite',
'src_associations_bin',
'src_associations_src',