bin/tools/yaml_test.py

   1 # -*- encoding: utf-8 -*-
   2 import types
   3 import yaml
   4 import time # used to eval time.strftime expressions
   5 import logging
   6
   7 import pooler
   8 import netsvc
   9
  10 from config import config
  11
  12 logger_channel = 'tests'
  13
  14 class YamlImportException(Exception):
  15     pass
  16
  17 class YamlImportAbortion(Exception):
  18     pass
  19
  20 class YamlTag(object):
  21     """Superclass for constructors of custom tags defined in yaml file."""
  22
  23     def __init__(self, **kwargs):
  24         self.__dict__.update(kwargs)
  25     def __getitem__(self, key):
  26         return getattr(self, key)
  27     def __getattr__(self, attr):
  28         return None
  29     def __repr__(self):
  30         return "<%s %s>" % (self.__class__.__name__, sorted(self.__dict__.items()))
  31
  32 class Assert(YamlTag):
  33     def __init__(self, model, id, severity=netsvc.LOG_ERROR, string="NONAME", **kwargs):
  34         self.model = model
  35         self.id = id
  36         self.severity = severity
  37         self.string = string
  38         super(Assert, self).__init__(**kwargs)
  39
  40 class Record(YamlTag):
  41     def __init__(self, model, id, **kwargs):
  42         self.model = model
  43         self.id = id
  44         super(Record, self).__init__(**kwargs)
  45
  46 class Python(YamlTag):
  47     def __init__(self, model, severity=netsvc.LOG_ERROR, name="NONAME", **kwargs):
  48         self.model= model
  49         self.severity = severity
  50         self.name = name
  51         super(Python, self).__init__(**kwargs)
  52
  53 class Workflow(YamlTag):
  54     def __init__(self, model, action, **kwargs):
  55         self.model = model
  56         self.action = action
  57         super(Workflow, self).__init__(**kwargs)
  58
  59 class Context(YamlTag):
  60     def __init__(self, **kwargs):
  61         super(Context, self).__init__(**kwargs)
  62
  63 class Eval(YamlTag):
  64     def __init__(self, expression):
  65         self.expression = expression
  66         super(Eval, self).__init__()
  67
  68 class Ref(YamlTag):
  69     def __init__(self, xmlid):
  70         self.xmlid = xmlid
  71         super(Ref, self).__init__()
  72
  73 def assert_constructor(loader, node):
  74     kwargs = loader.construct_mapping(node)
  75     return Assert(**kwargs)
  76
  77 def record_constructor(loader, node):
  78     kwargs = loader.construct_mapping(node)
  79     return Record(**kwargs)
  80
  81 def python_constructor(loader, node):
  82     kwargs = loader.construct_mapping(node)
  83     return Python(**kwargs)
  84
  85 def workflow_constructor(loader, node):
  86     kwargs = loader.construct_mapping(node)
  87     return Workflow(**kwargs)
  88
  89 def context_constructor(loader, node):
  90     kwargs = loader.construct_mapping(node)
  91     return Context(**kwargs)
  92
  93 def eval_constructor(loader, node):
  94     expression = loader.construct_scalar(node)
  95     return Eval(expression)
  96
  97 def ref_constructor(loader, node):
  98     xmlid = loader.construct_scalar(node)
  99     return Ref(xmlid)
 100
 101 yaml.add_constructor(u"!assert", assert_constructor)
 102 yaml.add_constructor(u"!record", record_constructor)
 103 yaml.add_constructor(u"!python", python_constructor)
 104 yaml.add_constructor(u"!workflow", workflow_constructor)
 105 yaml.add_constructor(u"!context", context_constructor)
 106 yaml.add_constructor(u"!eval", eval_constructor)
 107 yaml.add_constructor(u"!ref", ref_constructor)
 108
 109 def _is_yaml_mapping(node, tag_constructor):
 110     value = isinstance(node, types.DictionaryType) \
 111         and len(node.keys()) == 1 \
 112         and isinstance(node.keys()[0], tag_constructor)
 113     return value
 114
 115 def is_comment(node):
 116     return isinstance(node, types.StringTypes)
 117
 118 def is_assert(node):
 119     return _is_yaml_mapping(node, Assert)
 120
 121 def is_record(node):
 122     return _is_yaml_mapping(node, Record)
 123
 124 def is_python(node):
 125     return _is_yaml_mapping(node, Python)
 126
 127 def is_workflow(node):
 128     return isinstance(node, Workflow) \
 129         or _is_yaml_mapping(node, Workflow)
 130
 131 def is_context(node):
 132     return isinstance(node, Context)
 133
 134 def is_eval(node):
 135     return isinstance(node, Eval)
 136
 137 def is_ref(node):
 138     return isinstance(node, Ref)
 139
 140
 141 class TestReport(object):
 142     def __init__(self):
 143         self._report = {}
 144
 145     def record(self, success, severity):
 146         """Records the result of an assertion for the failed/success count.
 147            Returns success."""
 148
 149         if severity in self._report:
 150             self._report[severity][success] += 1
 151         else:
 152             self._report[severity] = {success: 1, not success: 0}
 153         return success
 154
 155     def __str__(self):
 156         res = []
 157         res.append('\nAssertions report:\nLevel\tsuccess\tfailed')
 158         success = failure = 0
 159         for severity in self._report:
 160             res.append("%s\t%s\t%s" % (severity, self._report[severity][True], self._report[severity][False]))
 161             success += self._report[severity][True]
 162             failure += self._report[severity][False]
 163         res.append("total\t%s\t%s" % (success, failure))
 164         res.append("end of report (%s assertion(s) checked)" % success + failure)
 165         return "\n".join(res)
 166
 167
 168 class YamlInterpreter(object):
 169     def __init__(self, cr, module, id_map, mode, filename, noupdate=False):
 170         self.cr = cr
 171         self.module = module
 172         self.id_map = id_map
 173         self.mode = mode
 174         self.filename = filename
 175         self.assert_report = TestReport()
 176         self.noupdate = noupdate
 177         self.logger = netsvc.Logger()
 178         self.pool = pooler.get_pool(cr.dbname)
 179         self.uid = 1
 180         self.context = {}
 181
 182     def _ref(self):
 183         return lambda xml_id: self.get_id(xml_id)
 184
 185     def get_model(self, model_name):
 186         model = self.pool.get(model_name)
 187         assert model, "The model %s does not exist." % (model_name,)
 188         return model
 189
 190     def validate_xml_id(self, xml_id):
 191         id = xml_id
 192         if '.' in xml_id:
 193             module, id = xml_id.split('.', 1)
 194             assert '.' not in id, "The ID reference '%s' must contains maximum one dot.\n" \
 195                                   "It is used to refer to other modules ID, in the form: module.record_id" \
 196                                   % (xml_id,)
 197             if module != self.module:
 198                 module_count = self.pool.get('ir.module.module').search_count(self.cr, self.uid, \
 199                         ['&', ('name', '=', module), ('state', 'in', ['installed'])])
 200                 assert module_count == 1, """The ID "%s" refers to an uninstalled module""" % (xml_id,)
 201         if len(id) > 64: # TODO where does 64 come from (DB is 128)? should be a constant or loaded form DB
 202             self.logger.notifyChannel(logger_channel, netsvc.LOG_ERROR, 'id: %s is to long (max: 64)' % (id,))
 203
 204     def get_id(self, xml_id):
 205         if isinstance(xml_id, types.IntType):
 206             id = xml_id
 207         elif xml_id in self.id_map:
 208             id = self.id_map[xml_id]
 209         else:
 210             if '.' in xml_id:
 211                 module, xml_id = xml_id.split('.', 1)
 212             else:
 213                 module = self.module
 214             ir_id = self.pool.get('ir.model.data')._get_id(self.cr, self.uid, module, xml_id)
 215             obj = self.pool.get('ir.model.data').read(self.cr, self.uid, ir_id, ['res_id'])
 216             id = int(obj['res_id'])
 217             self.id_map[xml_id] = id
 218         return id
 219
 220     def process_comment(self, node):
 221         return node
 222
 223     def _log_assert_failure(self, severity, msg, *args):
 224         self.assert_report.record(False, severity)
 225         self.logger.notifyChannel(logger_channel, severity, msg % (args))
 226         if getattr(logging, severity.upper()) >= config['assert_exit_level']:
 227             raise YamlImportAbortion('Severe assertion failure (%s), aborting.' % (severity,))
 228         return
 229
 230     def _get_assertion_id(self, assertion):
 231         if assertion.id:
 232             ids = [self.get_id(assertion.id)]
 233         elif assertion.search:
 234             q = eval(assertion.search, {'ref': self._ref})
 235             ids = self.pool.get(assertion.model).search(self.cr, self.uid, q, context=assertion.context)
 236         if not ids:
 237             raise YamlImportException('Nothing to assert: you must give either an id or a search criteria.')
 238         return ids
 239
 240     def process_assert(self, node):
 241         assertion, expressions = node.items()[0]
 242
 243         model = self.get_model(assertion.model)
 244         ids = self._get_assertion_id(assertion)
 245         if assertion.count and len(ids) != assertion.count:
 246             msg = 'assertion "%s" failed!\n'   \
 247                   ' Incorrect search count:\n' \
 248                   ' expected count: %d\n'      \
 249                   ' obtained count: %d\n'
 250             args = (assertion.string, assertion.count, len(ids))
 251             self._log_assert_failure(assertion.severity, msg, *args)
 252         else:
 253             test_context = {'ref': self._ref, '_ref': self._ref} # added '_ref' so that record['ref'] is possible
 254             for id in ids:
 255                 record = model.browse(self.cr, self.uid, id, assertion.context)
 256                 for test in expressions.get('test', ''):
 257                     success = eval(test, test_context, record)
 258                     if not success:
 259                         msg = 'Assertion "%s" FAILED\ntest: %s\n'
 260                         args = (assertion.string, test)
 261                         self._log_assert_failure(assertion.severity, msg, *args)
 262                         return
 263             else: # all tests were successful for this assertion tag (no break)
 264                 self.assert_report.record(True, assertion.severity)
 265
 266     def process_record(self, node):
 267         record, fields = node.items()[0]
 268
 269         self.validate_xml_id(record.id)
 270         model = self.get_model(record.model)
 271         record_dict = self._create_record(model, fields)
 272         id = self.pool.get('ir.model.data')._update(self.cr, self.uid, record.model, \
 273                 self.module, record_dict, record.id, mode=self.mode)
 274         self.id_map[record.id] = int(id)
 275         if config.get('import_partial', False):
 276             self.cr.commit()
 277
 278     def _create_record(self, model, fields):
 279         record_dict = {}
 280         for field_name, expression in fields.items():
 281             field_value = self._eval_field(model, field_name, expression)
 282             record_dict[field_name] = field_value
 283         return record_dict
 284
 285     def _eval_field(self, model, field_name, expression):
 286         column = model._columns[field_name]
 287         if column._type == "many2one":
 288             value = self.get_id(expression)
 289         elif column._type == "one2many":
 290             other_model = self.get_model(column._obj)
 291             value = [(0, 0, self._create_record(other_model, fields)) for fields in expression]
 292         elif column._type == "many2many":
 293             ids = [self.get_id(xml_id) for xml_id in expression]
 294             value= [(6, 0, ids)]
 295         else: # scalar field
 296             if isinstance(expression, Eval):
 297                 value = eval(expression.expression)
 298             else:
 299                 value = expression
 300             # raise YamlImportException('Unsupported column "%s" or value %s:%s' % (field_name, type(expression), expression))
 301         return value
 302
 303     def process_context(self, node):
 304         self.context = node.__dict__
 305         if 'uid' in node.__dict__:
 306             self.uid = self.get_id(node.__dict__['uid'])
 307
 308     def process_python(self, node):
 309         def log(msg):
 310             self.logger.notifyChannel(logger_channel, netsvc.LOG_TEST, msg)
 311         python, statements = node.items()[0]
 312         model = self.get_model(python.model)
 313         statements = statements.replace("\r\n", "\n").replace("\n", ";")
 314         code_context = {'self': model, 'cr': self.cr, 'uid': self.uid, 'log': log, 'context': self.context}
 315         try:
 316             code = compile(statements, self.filename, 'exec')
 317             eval(code, code_context)
 318         except AssertionError, e:
 319             msg = 'Assertion "%s" FAILED in Python code.'
 320             args = (python.name,)
 321             self._log_assert_failure(python.severity, msg, *args)
 322             return
 323         except Exception, e:
 324             raise YamlImportAbortion(e)
 325         else:
 326             self.assert_report.record(True, python.severity)
 327
 328     def process_workflow(self, node):
 329         workflow, values = node.items()[0]
 330         model = self.get_model(workflow.model)
 331         if workflow.ref:
 332             id = self.get_id(workflow.ref)
 333         else:
 334             if not values:
 335                 raise YamlImportException('You must define a child node if you do not give a ref.')
 336             if not len(values) == 1:
 337                 raise YamlImportException('Only one child node is accepted (%d given).' % len(values))
 338             value = values[0]
 339             if not 'model' in value and not 'eval' in value:
 340                 raise YamlImportException('You must provide a model and an expression to evaluate for the workflow.')
 341             value_model = self.get_model(value['model'])
 342             local_context = {'ref': self._ref, '_ref': self._ref}
 343             local_context['obj'] = lambda x: value_model.browse(self.cr, self.uid, x, context=self.context)
 344             local_context.update(self.id_map)
 345             id = eval(value['eval'], local_context)
 346
 347         if workflow.uid is not None:
 348             uid = workflow.uid
 349         else:
 350             uid = self.uid
 351         wf_service = netsvc.LocalService("workflow")
 352         wf_service.trg_validate(uid, model, id, workflow.action, self.cr)
 353
 354     def process(self, yaml_string):
 355         """Processes a Yaml string. Custom tags are interpreted by 'process_' instance methods."""
 356
 357         is_preceded_by_comment = False
 358         for node in yaml.load(yaml_string):
 359             is_preceded_by_comment = self._log(node, is_preceded_by_comment)
 360             try:
 361                 self._process_node(node)
 362             except YamlImportException, e:
 363                 self.logger.notifyChannel(logger_channel, netsvc.LOG_ERROR, e)
 364             except YamlImportAbortion, e:
 365                 self.logger.notifyChannel(logger_channel, netsvc.LOG_ERROR, e)
 366                 return
 367
 368     def _process_node(self, node):
 369         if is_comment(node):
 370             self.process_comment(node)
 371         elif is_assert(node):
 372             self.process_assert(node)
 373         elif is_record(node):
 374             self.process_record(node)
 375         elif is_python(node):
 376             self.process_python(node)
 377         elif is_context(node):
 378             self.process_context(node)
 379         elif is_workflow(node):
 380             if isinstance(node, types.DictionaryType):
 381                 self.process_workflow(node)
 382             else:
 383                 self.process_workflow({node: []})
 384         else:
 385             raise YamlImportException("Can not process YAML block: %s" % node)
 386
 387     def _log(self, node, is_preceded_by_comment):
 388         if is_comment(node):
 389             is_preceded_by_comment = True
 390             self.logger.notifyChannel(logger_channel, netsvc.LOG_TEST, node)
 391         elif not is_preceded_by_comment:
 392             if isinstance(node, types.DictionaryType):
 393                 k, v = node.items()[0]
 394                 msg = "Creating %s\n with %s" % (k, v)
 395                 self.logger.notifyChannel(logger_channel, netsvc.LOG_TEST, msg)
 396             else:
 397                 self.logger.notifyChannel(logger_channel, netsvc.LOG_TEST, node)
 398         else:
 399             is_preceded_by_comment = False
 400         return is_preceded_by_comment
 401
 402 def yaml_import(cr, module, yamlfile, idref=None, mode='init', noupdate=False, report=None):
 403     if idref is None:
 404         idref = {}
 405     yaml_string = yamlfile.read()
 406     yaml_interpreter = YamlInterpreter(cr, module, idref, mode, filename=yamlfile.name, noupdate=noupdate)
 407     yaml_interpreter.process(yaml_string)
 408
 409 # keeps convention of convert.py
 410 convert_yaml_import = yaml_import
 411
 412 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: