aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat (limited to 'overlord/dbbase.py')
-rw-r--r--overlord/dbbase.py276
1 files changed, 276 insertions, 0 deletions
diff --git a/overlord/dbbase.py b/overlord/dbbase.py
new file mode 100644
index 0000000..7821aea
--- /dev/null
+++ b/overlord/dbbase.py
@@ -0,0 +1,276 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+#################################################################################
+# overlord OVERLAY HANDLER
+#################################################################################
+# File: overlay.py
+#
+# Access to an xml list of overlays
+#
+# Copyright:
+# (c) 2005 - 2009 Gunnar Wrobel
+# (c) 2009 Sebastian Pipping
+# (c) 2009 Christian Groschupp
+# Distributed under the terms of the GNU General Public License v2
+#
+# Author(s):
+# Gunnar Wrobel <wrobel@gentoo.org>
+# Sebastian Pipping <sebastian@pipping.org>
+# Christian Groschupp <christian@groschupp.org>
+#
+'''Main handler for overlays.'''
+
+__version__ = "$Id: overlay.py 273 2006-12-30 15:54:50Z wrobel $"
+
+#===============================================================================
+#
+# Dependencies
+#
+#-------------------------------------------------------------------------------
+
+import sys, os, os.path
+import xml
+import xml.etree.ElementTree as ET # Python 2.5
+
+#from overlord.debug import OUT
+from overlord.utils import indent
+from overlord.overlays.overlay import Overlay
+
+#===============================================================================
+#
+# Class UnknownOverlayException
+#
+#-------------------------------------------------------------------------------
+
+class UnknownOverlayException(Exception):
+ def __init__(self, repo_name):
+ message = 'Overlay "%s" does not exist.' % repo_name
+ super(UnknownOverlayException, self).__init__(message)
+
+#===============================================================================
+#
+# Class BrokenOverlayCatalog
+#
+#-------------------------------------------------------------------------------
+
+class BrokenOverlayCatalog(ValueError):
+ def __init__(self, origin, expat_error, hint=None):
+ if hint == None:
+ hint = ''
+ else:
+ hint = '\nHint: %s' % hint
+
+ super(BrokenOverlayCatalog, self).__init__(
+ 'XML parsing failed for "%(origin)s" (line %(line)d, column %(column)d)%(hint)s' % \
+ {'line':expat_error.lineno, 'column':expat_error.offset + 1, 'origin':origin, 'hint':hint})
+
+
+#===============================================================================
+#
+# Class DbBase
+#
+#-------------------------------------------------------------------------------
+
+class DbBase:
+ ''' Handle a list of overlays.'''
+
+ def __init__(self, paths, config, ignore = 0, quiet = False, ignore_init_read_errors=False):
+
+ self.config = config
+ self.quiet = quiet
+ self.paths = paths
+ self.ignore = ignore
+ self.output = config['output']
+
+ self.overlays = {}
+
+ self.output.debug('Initializing overlay list handler', 8)
+
+ for path in self.paths:
+ if not os.path.exists(path):
+ continue
+
+ try:
+ self.read_file(path)
+ except Exception, error:
+ if not ignore_init_read_errors: raise error
+
+ def __eq__(self, other):
+ for key in set(self.overlays.keys() + other.overlays.keys()):
+ if self.overlays[key] != other.overlays[key]:
+ return False
+ return True
+
+ def __ne__(self, other):
+ return not self.__eq__(other)
+
+ def read_file(self, path):
+ '''Read the overlay definition file.'''
+
+ try:
+ document = open(path, 'r').read()
+
+ except Exception, error:
+ raise IOError('Failed to read the overlay list at ("'
+ + path + '")!\nError was:\n' + str(error))
+
+ self.read(document, origin=path)
+
+ def _broken_catalog_hint(self):
+ this_function_name = sys._getframe().f_code.co_name
+ raise NotImplementedError('Method "%s.%s" not implemented' % \
+ (self.__class__.__name__, this_function_name))
+
+ def read(self, text, origin):
+ '''
+ Read an xml list of overlays (adding to and potentially overwriting existing entries)
+
+ >>> here = os.path.dirname(os.path.realpath(__file__))
+ >>> config = {'svn_command': '/usr/bin/svn', 'rsync_command':'/usr/bin/rsync'}
+ >>> a = DbBase([here + '/tests/testfiles/global-overlays.xml', ], config)
+ >>> a.overlays.keys()
+ [u'wrobel', u'wrobel-stable']
+
+ >>> list(a.overlays['wrobel-stable'].source_uris())
+ [u'rsync://gunnarwrobel.de/wrobel-stable']
+ '''
+ try:
+ document = ET.fromstring(text)
+ except xml.parsers.expat.ExpatError, error:
+ raise BrokenOverlayCatalog(origin, error, self._broken_catalog_hint())
+
+ overlays = document.findall('overlay') + \
+ document.findall('repo')
+
+ for overlay in overlays:
+ self.output.debug('Parsing overlay entry', 8)
+ try:
+ ovl = Overlay(overlay, self.config, self.ignore, self.quiet)
+ except Exception, error:
+ self.output.warn(str(error), 3)
+ else:
+ self.overlays[ovl.name] = ovl
+
+
+ def write(self, path):
+ '''
+ Write the list of overlays to a file.
+
+ >>> write = os.tmpnam()
+ >>> here = os.path.dirname(os.path.realpath(__file__))
+ >>> config = {'svn_command': '/usr/bin/svn', 'rsync_command':'/usr/bin/rsync'}
+ >>> a = DbBase([here + '/tests/testfiles/global-overlays.xml', ], config)
+ >>> b = DbBase([write,], dict())
+ >>> b.overlays['wrobel-stable'] = a.overlays['wrobel-stable']
+ >>> b.write(write)
+ >>> c = DbBase([write,], dict())
+ >>> c.overlays.keys()
+ [u'wrobel-stable']
+
+ >>> os.unlink(write)
+ '''
+
+ tree = ET.Element('repositories', version="1.0")
+ tree[:] = [e.to_xml() for e in self.overlays.values()]
+ indent(tree)
+ tree = ET.ElementTree(tree)
+ try:
+ f = open(path, 'w')
+ f.write("""\
+<?xml version="1.0" encoding="UTF-8"?>
+""")
+ tree.write(f, encoding='utf-8')
+ f.close()
+ except Exception, error:
+ raise Exception('Failed to write to local overlays file: '
+ + path + '\nError was:\n' + str(error))
+
+ def select(self, overlay):
+ '''
+ Select an overlay from the list.
+
+ >>> here = os.path.dirname(os.path.realpath(__file__))
+ >>> config = {'svn_command': '/usr/bin/svn', 'rsync_command':'/usr/bin/rsync'}
+ >>> a = DbBase([here + '/tests/testfiles/global-overlays.xml', ], config)
+ >>> list(a.select('wrobel-stable').source_uris())
+ [u'rsync://gunnarwrobel.de/wrobel-stable']
+ '''
+ if not overlay in self.overlays.keys():
+ raise UnknownOverlayException(overlay)
+ return self.overlays[overlay]
+
+ def list(self, repos=None, verbose = False, width = 0):
+ '''
+ List all overlays.
+
+ >>> here = os.path.dirname(os.path.realpath(__file__))
+ >>> config = {'svn_command': '/usr/bin/svn', 'rsync_command':'/usr/bin/rsync'}
+ >>> a = DbBase([here + '/tests/testfiles/global-overlays.xml', ], config)
+ >>> for i in a.list(True):
+ ... print i[0]
+ wrobel
+ ~~~~~~
+ Source : https://overlays.gentoo.org/svn/dev/wrobel
+ Contact : nobody@gentoo.org
+ Type : Subversion; Priority: 10
+ Quality : experimental
+ <BLANKLINE>
+ Description:
+ Test
+ <BLANKLINE>
+ wrobel-stable
+ ~~~~~~~~~~~~~
+ Source : rsync://gunnarwrobel.de/wrobel-stable
+ Contact : nobody@gentoo.org
+ Type : Rsync; Priority: 50
+ Quality : experimental
+ <BLANKLINE>
+ Description:
+ A collection of ebuilds from Gunnar Wrobel [wrobel@gentoo.org].
+ <BLANKLINE>
+
+ >>> for i in a.list(False, 80):
+ ... print i[0]
+ wrobel [Subversion] (https://o.g.o/svn/dev/wrobel )
+ wrobel-stable [Rsync ] (rsync://gunnarwrobel.de/wrobel-stable)
+ '''
+ result = []
+
+ selection = [overlay for (a, overlay) in self.overlays.items()]
+ if repos:
+ selection = [overlay for overlay in selection if overlay.name in repos]
+
+ for overlay in selection:
+ if verbose:
+ result.append((str(overlay), overlay.is_supported(),
+ overlay.is_official()))
+ else:
+ result.append((overlay.short_list(width), overlay.is_supported(),
+ overlay.is_official()))
+
+ result = sorted(result)
+
+ return result
+
+ def list_ids(self):
+ """returns a list of the overlay names
+ """
+ return sorted(self.overlays)
+
+
+#===============================================================================
+#
+# Testing
+#
+#-------------------------------------------------------------------------------
+
+if __name__ == '__main__':
+ import doctest
+
+ # Ignore warnings here. We are just testing
+ from warnings import filterwarnings, resetwarnings
+ filterwarnings('ignore')
+
+ doctest.testmod(sys.modules[__name__])
+
+ resetwarnings()