from sqlalchemy.orm import sessionmaker, mapper, relation, object_session, \
backref, MapperExtension, EXT_CONTINUE, object_mapper, clear_mappers
from sqlalchemy import types as sqltypes
+from sqlalchemy.orm.collections import attribute_mapped_collection
+from sqlalchemy.ext.associationproxy import association_proxy
# Don't remove this, we re-export the exceptions to scripts which import us
from sqlalchemy.exc import *
return ['package', 'version', 'maintainer', 'source', 'poolfile', \
'binarytype']
+ metadata = association_proxy('key', 'value')
+
def get_component_name(self):
return self.poolfile.location.component.component_name
return ['source', 'version', 'install_date', 'maintainer', \
'changedby', 'poolfile', 'install_date']
+ metadata = association_proxy('key', 'value')
+
__all__.append('DBSource')
@session_wrapper
################################################################################
class BinaryMetadata(ORMObject):
- def __init__(self, binary = None, key = None, value = None):
- self.binary = binary
+ def __init__(self, key = None, value = None, binary = None):
self.key = key
self.value = value
+ self.binary = binary
def properties(self):
return ['binary', 'key', 'value']
################################################################################
class SourceMetadata(ORMObject):
- def __init__(self, source = None, key = None, value = None):
- self.source = source
+ def __init__(self, key = None, value = None, source = None):
self.key = key
self.value = value
+ self.source = source
def properties(self):
return ['source', 'key', 'value']
suites = relation(Suite, secondary=self.tbl_bin_associations,
backref=backref('binaries', lazy='dynamic')),
extra_sources = relation(DBSource, secondary=self.tbl_extra_src_references,
- backref=backref('extra_binary_references', lazy='dynamic'))),
+ backref=backref('extra_binary_references', lazy='dynamic')),
+ key = relation(BinaryMetadata,
+ collection_class=attribute_mapped_collection('key'))),
extension = validator)
mapper(BinaryACL, self.tbl_binary_acl,
primaryjoin=(self.tbl_source.c.id==self.tbl_dsc_files.c.source)),
suites = relation(Suite, secondary=self.tbl_src_associations,
backref=backref('sources', lazy='dynamic')),
- srcuploaders = relation(SrcUploader)),
+ srcuploaders = relation(SrcUploader),
+ key = relation(SourceMetadata,
+ collection_class=attribute_mapped_collection('key'))),
extension = validator)
mapper(SourceACL, self.tbl_source_acl,
This TestCase checks the metadata handling.
"""
+ def setup_metadata(self):
+ '''
+ Setup the metadata objects.
+ '''
+ self.setup_binaries()
+ self.depends = MetadataKey('Depends')
+ self.session.add(self.depends)
+ self.session.flush()
+ self.bin_hello = self.binary['hello_2.2-1_i386']
+ self.src_hello = self.bin_hello.source
+ self.session.add(self.bin_hello)
+ self.session.add(self.src_hello)
+ self.hello_dep = BinaryMetadata(self.depends, 'foobar (>= 1.0)', self.bin_hello)
+ self.session.add(self.hello_dep)
+ self.recommends = MetadataKey('Recommends')
+ self.bin_hello.key[self.recommends] = BinaryMetadata(self.recommends, 'goodbye')
+ self.build_dep = MetadataKey('Build-Depends')
+ self.hello_bd = SourceMetadata(self.build_dep, 'foobar-dev', self.src_hello)
+ self.session.add(self.hello_bd)
+ self.homepage = MetadataKey('Homepage')
+ self.src_hello.key[self.homepage] = SourceMetadata(self.homepage, 'http://debian.org')
+ self.session.flush()
+
def test_mappers(self):
'''
Tests the mapper configuration.
'''
- self.setup_binaries()
+ self.setup_metadata()
# MetadataKey
- depends = MetadataKey(key = 'Depends')
- self.session.add(depends)
- self.session.flush()
- self.assertTrue(depends.key_id is not None)
+ self.assertTrue(self.depends.key_id is not None)
# BinaryMetadata
- hello_dep = BinaryMetadata(binary = self.binary['hello_2.2-1_i386'],
- key = depends, value = 'foobar (>= 1.0)')
- self.session.add(hello_dep)
- self.session.flush()
- self.assertEqual('hello', hello_dep.binary.package)
- self.assertEqual('Depends', hello_dep.key.key)
- self.assertEqual('foobar (>= 1.0)', hello_dep.value)
+ self.assertEqual('hello', self.hello_dep.binary.package)
+ self.assertEqual('Depends', self.hello_dep.key.key)
+ self.assertEqual('foobar (>= 1.0)', self.hello_dep.value)
# SourceMetadata
- build_dep = MetadataKey(key = 'Build-Depends')
- hello_bd = SourceMetadata(source = self.binary['hello_2.2-1_i386'].source,
- key = build_dep, value = 'foobar-dev')
- self.session.add(hello_bd)
+ self.assertEqual('hello', self.hello_bd.source.source)
+ self.assertEqual('Build-Depends', self.hello_bd.key.key)
+ self.assertEqual('foobar-dev', self.hello_bd.value)
+ # DBBinary relation
+ self.assertEqual(self.hello_dep, self.bin_hello.key[self.depends])
+ self.assertEqual('goodbye', self.bin_hello.key[self.recommends].value)
+ # DBSource relation
+ self.assertEqual(self.hello_bd, self.src_hello.key[self.build_dep])
+ self.assertEqual('http://debian.org', self.src_hello.key[self.homepage].value)
+
+ def test_association_proxy(self):
+ '''
+ Test the association proxies 'metadata' in DBBinary and DBSource.
+ '''
+ self.setup_metadata()
+ # DBBinary association proxy
+ essential = MetadataKey('Essential')
+ self.bin_hello.metadata[essential] = 'yes'
+ self.session.flush()
+ self.assertEqual('yes', self.bin_hello.metadata[essential])
+ self.assertEqual('foobar (>= 1.0)', self.bin_hello.metadata[self.depends])
+ self.assertTrue(self.recommends in self.bin_hello.metadata)
+ # DBSource association proxy
+ std_version = MetadataKey('Standards-Version')
+ self.src_hello.metadata[std_version] = '3.9.1'
self.session.flush()
- self.assertEqual('hello', hello_bd.source.source)
- self.assertEqual('Build-Depends', hello_bd.key.key)
- self.assertEqual('foobar-dev', hello_bd.value)
+ self.assertEqual('3.9.1', self.src_hello.metadata[std_version])
+ self.assertEqual('http://debian.org', self.src_hello.metadata[self.homepage])
+ self.assertTrue(self.depends not in self.src_hello.metadata)
if __name__ == '__main__':
unittest.main()