Source code for destral.openerp

import logging
import time

from osconf import config_from_environment
from destral.utils import update_config

logger = logging.getLogger('destral.openerp')
"""Default user id

[docs]def patched_pool_jobs(*args, **kwargs): """Patch pool jobs """ logger.debug('Patched ir.cron') return False
[docs]class OpenERPService(object): """OpenERP Service. """ def __init__(self, **kwargs): """Creates a new OpenERP service. :param \**kwargs: keyword arguments passed to the config """ config = config_from_environment('OPENERP', [], **kwargs) import netsvc import tools update_config(tools.config, **config) if hasattr(tools.config, 'parse'): tools.config.parse() from tools import config as default_config self.config = update_config(default_config, **config) import pooler import workflow self.pooler = pooler self.db = None self.pool = None if 'db_name' in config: self.db_name = config['db_name'] # Stop the cron netsvc.Agent.quit()
[docs] def create_database(self, template=True): """Creates a new database. :param template: use a template (name must be `base`) (default True) """ db_name = 'test_' + str(int(time.time())) import sql_db conn = sql_db.db_connect('postgres') cursor = conn.cursor() try:'Creating database %s', db_name) cursor.autocommit(True) if template: cursor.execute('CREATE DATABASE {} WITH TEMPLATE base'.format( db_name )) else: cursor.execute('CREATE DATABASE {}'.format(db_name)) return db_name finally: cursor.close() sql_db.close_db('postgres')
[docs] def drop_database(self): """Drop database from `self.db_name` """ import sql_db sql_db.close_db(self.db_name) conn = sql_db.db_connect('template1') cursor = conn.cursor() try:'Droping database %s', self.db_name) cursor.autocommit(True) cursor.execute('DROP DATABASE ' + self.db_name) finally: cursor.close()
@property def db_name(self): """Database name. """ return self.config['db_name'] @db_name.setter def db_name(self, value): self.config['db_name'] = value if value: self.db, self.pool = self.pooler.get_db_and_pool(self.db_name) else: self.db = None self.pool = None
[docs] def install_module(self, module): """Installs a module :param module: Module to install """'Installing module %s', module) import pooler from destral.transaction import Transaction module_obj = self.pool.get('ir.module.module') with Transaction().start(self.config['db_name']) as txn: module_obj.update_list(txn.cursor, txn.user) module_ids = txn.cursor, DEFAULT_USER, [('name', '=', module)], ) assert module_ids, "Module %s not found" % module module_obj.button_install(txn.cursor, DEFAULT_USER, module_ids) pool = pooler.get_pool(txn.cursor.dbname) mod_obj = pool.get('ir.module.module') ids =, txn.user, [ ('state', 'in', ['to upgrade', 'to remove', 'to install']) ]) unmet_packages = [] mod_dep_obj = pool.get('ir.module.module.dependency') for mod in mod_obj.browse(txn.cursor, txn.user, ids): deps =, txn.user, [ ('module_id', '=', ]) for dep_mod in mod_dep_obj.browse(txn.cursor, txn.user, deps): if dep_mod.state in ('unknown', 'uninstalled'): unmet_packages.append(, txn.user, ids) txn.cursor.commit() self.db, self.pool = pooler.restart_pool( self.config['db_name'], update_module=True )
def enable_admin(self, password='admin'): from destral.transaction import Transaction with Transaction().start(self.config['db_name']) as txn: user_obj = self.pool.get('res.users') user_ids =, txn.user, [ ('login', '=', 'admin'), ('password', '=', False) ], context={'active_test': False}) if user_ids: user_obj.write(txn.cursor, txn.user, user_ids, { 'active': True, 'password': password }) txn.cursor.commit() 'User admin enabled with password: %s on %s', password, txn.cursor.dbname)