Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
assert(schema.name == 'Job')
assert(schema.category == 'jobs')
assert(schema.allItems() == list(dd.items()))
cc = (schema.componentItems() + schema.simpleItems()).sort()
cc2 = list(dd.items()).sort()
assert(cc == cc2)
for i in schema.allItems():
assert(schema[i[0]] == schema.getItem(i[0]))
assert(schema['id'].isA(SimpleItem))
assert(schema['application'].isA(ComponentItem))
assert(schema['inputbox'].isA(ComponentItem))
assert(schema['inputbox'].isA(FileItem))
assert(schema['id']['protected'])
assert(not schema['id']['comparable'])
assert(schema['id']['type'] == 'string')
logger.info(schema['application']['category'] + ' ' + schema['application']['defvalue'])
schema2 = copy.deepcopy(schema)
assert(schema2 is not schema)
assert(schema.datadict is not schema2.datadict)
assert(schema._pluginclass is schema2._pluginclass)
for i in schema.allItems():
assert(schema.getItem(i[0]) is not schema2.getItem(i[0]))
def _check_type(self, val, name, enableGangaList=True):
if enableGangaList:
from GangaCore.GPIDev.Lib.GangaList.GangaList import GangaList
else:
GangaList = list
# type checking does not make too much sense for Component Items because component items are
# always checked at the object (_impl) level for category compatibility.
if self.isA(ComponentItem):
return
validTypes = self._meta['typelist']
# setting typelist explicitly to None results in disabling the type
# checking completely
if validTypes is None:
return
if self._meta['sequence']:
from GangaCore.GPIDev.Base.Proxy import isType
if not isType(self._meta['defvalue'], (list, tuple, GangaList)):
raise SchemaError('Attribute "%s" defined as a sequence but defvalue is not a list.' % name)
if not isType(val, (GangaList, tuple, list)):
raise TypeMismatchError('Attribute "%s" expects a list.' % name)
# Files are important and common enough to merit a special support for
# defining their metaproperties
class FileItem(ComponentItem):
__slots__=list()
def __init__(self, **kwds):
super(FileItem, self).__init__('files')
self._update(kwds)
def _describe(self):
return "'files' object," + Item._describe(self)
class GangaFileItem(ComponentItem):
__slots__ = list()
def __init__(self, **kwds):
super(GangaFileItem, self).__init__('gangafiles')
self._update(kwds)
def _describe(self):
return "'gangafiles' object," + Item._describe(self)
if __name__ == '__main__':
# a simple test
dd = {
def prehook(name, x):
errmsg = "Cannot set %s=%s in [%s]: " % (name, repr(x), config.name)
try:
item = self.getItem(name)
except Exception as x:
raise GangaCore.Utility.Config.ConfigError(errmsg + str(x))
if item.isA(ComponentItem):
if not isinstance(x, str) and not x is None:
raise GangaCore.Utility.Config.ConfigError(errmsg + "only strings and None allowed as a default value of Component Item.")
try:
self._getDefaultValueInternal(name, x, True)
except Exception as err:
logger.info("Unexpected error: %s", err)
raise
if item['protected'] or item['hidden']:
raise GangaCore.Utility.Config.ConfigError(errmsg + "protected or hidden property")
return x
def __init__(self, **kwds):
super(GangaFileItem, self).__init__('gangafiles')
self._update(kwds)
def _describe(self):
return "'gangafiles' object," + Item._describe(self)
if __name__ == '__main__':
# a simple test
dd = {
'application': ComponentItem(category='applications'),
'backend': ComponentItem(category='backends'),
'name': SimpleItem('', comparable=0),
'workdir': SimpleItem(defvalue=None, type='string', transient=1, protected=1, comparable=0),
'status': SimpleItem(defvalue='new', protected=1, comparable=0),
'id': SimpleItem(defvalue=None, type='string', protected=1, comparable=0),
'inputbox': FileItem(defvalue=[], sequence=1),
'outputbox': FileItem(defvalue=[], sequence=1),
'overriden_copyable': SimpleItem(defvalue=None, protected=1, copyable=1),
'plain_copyable': SimpleItem(defvalue=None, copyable=0)
}
schema = Schema(Version(1, 0), dd)
# NOT a public interface: emulate the Ganga Plugin object for test purposes
# Note that pclass MUST be a new-style class in order to support deepcopy
class pclass(object):
_category = 'jobs'
def __init__(self, category, optional=0, load_default=1, **kwds):
super(ComponentItem, self).__init__()
kwds['category'] = category
kwds['optional'] = optional
kwds['load_default'] = load_default
#kwds['getter'] = getter
self._update(kwds, ComponentItem._forced)
try:
assert(implies(self['defvalue'] is None and not self['load_default'], self['optional']))
except AssertionError:
raise SchemaError("ComponentItem has no defvalue, load_default or requirement to be optional")
try:
assert(implies(self['getter'], self['transient'] and self['defvalue'] is None and self['protected'] and not self['sequence'] and not self['copyable']))
except AssertionError:
raise SchemaError("There is no getter, transient flag or defvalue and the ComponentItem is protected, not a sequence or not copyable. This is not supported")
useDefVal = True
if useDefVal:
# hidden, protected and sequence values are not represented in config
defvalue = item['defvalue']
from GangaCore.GPIDev.Base.Proxy import isProxy
if isProxy(defvalue):
raise GangaException("(2)Proxy found where is shouldn't be in the Config" % stored_attr_key)
## Just in case a developer puts the proxied object into the default value!
_found_attrs[stored_attr_key] = defvalue
# in the checking mode, use the provided value instead
if check is True:
defvalue = val
if isinstance(item, ComponentItem):
# FIXME: limited support for initializing non-empty sequences (i.e.
# apps => ['DaVinci','Executable'] is NOT correctly initialized)
if not item['sequence']:
if defvalue is None:
if not item['load_default']:
try:
assert(item['optional'])
except AssertionError:
raise SchemaError("This item '%s' is not a sequence, doesn't have a load_default and is not optional. This is unsupported!" % type(item))
return None
# if a defvalue of a component item is an object (not string) just process it as for SimpleItems (useful for FileItems)
# otherwise do a lookup via plugin registry
def componentItems(self):
return self._filter(ComponentItem)
def __init__(self, category, optional=0, load_default=1, **kwds):
super(ComponentItem, self).__init__()
kwds['category'] = category
kwds['optional'] = optional
kwds['load_default'] = load_default
#kwds['getter'] = getter
self._update(kwds, ComponentItem._forced)
try:
assert(implies(self['defvalue'] is None and not self['load_default'], self['optional']))
except AssertionError:
raise SchemaError("ComponentItem has no defvalue, load_default or requirement to be optional")
try:
assert(implies(self['getter'], self['transient'] and self['defvalue'] is None and self['protected'] and not self['sequence'] and not self['copyable']))
except AssertionError:
raise SchemaError("There is no getter, transient flag or defvalue and the ComponentItem is protected, not a sequence or not copyable. This is not supported")
if typelist == defaultValue:
if type(defvalue) == dict:
typelist = []
else:
typelist = [type(defvalue)]
kwds['defvalue'] = defvalue
kwds['typelist'] = typelist
self._update(kwds)
def _describe(self):
return 'shared property,' + Item._describe(self)
# Files are important and common enough to merit a special support for
# defining their metaproperties
class FileItem(ComponentItem):
__slots__=list()
def __init__(self, **kwds):
super(FileItem, self).__init__('files')
self._update(kwds)
def _describe(self):
return "'files' object," + Item._describe(self)
class GangaFileItem(ComponentItem):
__slots__ = list()
def __init__(self, **kwds):