#
##############################################################################
-""" OpenERP core library..
-
-"""
-
-import sys
+""" OpenERP core library."""
+#----------------------------------------------------------
+# Running mode flags (gevent, prefork)
+#----------------------------------------------------------
# Is the server running with gevent.
+import sys
evented = False
if sys.modules.get("gevent") is not None:
evented = True
+# Is the server running in pefork mode (e.g. behind Gunicorn).
+# If this is True, the processes have to communicate some events,
+# e.g. database update or cache invalidation. Each process has also
+# its own copy of the data structure and we don't need to care about
+# locks between threads.
+multi_process = False
+
+#----------------------------------------------------------
+# libc UTC hack
+#----------------------------------------------------------
# Make sure the OpenERP server runs in UTC. This is especially necessary
# under Windows as under Linux it seems the real import of time is
# sufficiently deferred so that setting the TZ environment variable
del os
del time
+#----------------------------------------------------------
+# Shortcuts
+#----------------------------------------------------------
# The hard-coded super-user id (a.k.a. administrator, or root user).
SUPERUSER_ID = 1
+def registry(database_name):
+ """
+ Return the model registry for the given database. If the registry does not
+ exist yet, it is created on the fly.
+ """
+ return modules.registry.RegistryManager.get(database_name)
+
+#----------------------------------------------------------
+# Imports
+#----------------------------------------------------------
import addons
import cli
import conf
import sql_db
import tools
import workflow
-# backward compatilbility
-# TODO: This is for the web addons, can be removed later.
-wsgi = service
-wsgi.register_wsgi_handler = wsgi.wsgi_server.register_wsgi_handler
-# Is the server running in multi-process mode (e.g. behind Gunicorn).
-# If this is True, the processes have to communicate some events,
-# e.g. database update or cache invalidation. Each process has also
-# its own copy of the data structure and we don't need to care about
-# locks between threads.
-multi_process = False
-
-def registry(database_name):
- """
- Return the model registry for the given database. If the registry does not
- exist yet, it is created on the fly.
- """
- return modules.registry.RegistryManager.get(database_name)
# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
def wsgi_postload():
global root
root = Root()
- openerp.wsgi.register_wsgi_handler(root)
+ openerp.service.wsgi_server.register_wsgi_handler(root)
# vim:et:ts=4:sw=4:
--- /dev/null
+# -*- coding: utf-8 -*-
+"""
+The module :mod:`openerp.tests.common` provides unittest2 test cases and a few
+helpers and classes to write tests.
+
+"""
+import json
+import os
+import select
+import subprocess
+import threading
+import time
+import unittest2
+import uuid
+import xmlrpclib
+import logging
+import sys
+
+import openerp
+# backward compatbility
+common = sys.modules['openerp.tests.common'] = sys.modules['openerp.tests']
+
+_logger = logging.getLogger(__name__)
+
+# The openerp library is supposed already configured.
+ADDONS_PATH = openerp.tools.config['addons_path']
+HOST = '127.0.0.1'
+PORT = openerp.tools.config['xmlrpc_port']
+DB = openerp.tools.config['db_name']
+
+# If the database name is not provided on the command-line,
+# use the one on the thread (which means if it is provided on
+# the command-line, this will break when installing another
+# database from XML-RPC).
+if not DB and hasattr(threading.current_thread(), 'dbname'):
+ DB = threading.current_thread().dbname
+
+ADMIN_USER = 'admin'
+ADMIN_USER_ID = openerp.SUPERUSER_ID
+ADMIN_PASSWORD = 'admin'
+
+HTTP_SESSION = {}
+
+class BaseCase(unittest2.TestCase):
+ """
+ Subclass of TestCase for common OpenERP-specific code.
+
+ This class is abstract and expects self.cr and self.uid to be initialized by subclasses.
+ """
+
+ @classmethod
+ def cursor(self):
+ return openerp.modules.registry.RegistryManager.get(DB).db.cursor()
+
+ @classmethod
+ def registry(self, model):
+ return openerp.modules.registry.RegistryManager.get(DB)[model]
+
+ @classmethod
+ def ref(self, xid):
+ """ Returns database ID corresponding to a given identifier.
+
+ :param xid: fully-qualified record identifier, in the form ``module.identifier``
+ :raise: ValueError if not found
+ """
+ assert "." in xid, "this method requires a fully qualified parameter, in the following form: 'module.identifier'"
+ module, xid = xid.split('.')
+ _, id = self.registry('ir.model.data').get_object_reference(self.cr, self.uid, module, xid)
+ return id
+
+ @classmethod
+ def browse_ref(self, xid):
+ """ Returns a browsable record for the given identifier.
+
+ :param xid: fully-qualified record identifier, in the form ``module.identifier``
+ :raise: ValueError if not found
+ """
+ assert "." in xid, "this method requires a fully qualified parameter, in the following form: 'module.identifier'"
+ module, xid = xid.split('.')
+ return self.registry('ir.model.data').get_object(self.cr, self.uid, module, xid)
+
+
+class TransactionCase(BaseCase):
+ """
+ Subclass of BaseCase with a single transaction, rolled-back at the end of
+ each test (method).
+ """
+
+ def setUp(self):
+ # Store cr and uid in class variables, to allow ref() and browse_ref to be BaseCase @classmethods
+ # and still access them
+ TransactionCase.cr = self.cursor()
+ TransactionCase.uid = openerp.SUPERUSER_ID
+
+ def tearDown(self):
+ self.cr.rollback()
+ self.cr.close()
+
+
+class SingleTransactionCase(BaseCase):
+ """
+ Subclass of BaseCase with a single transaction for the whole class,
+ rolled-back after all the tests.
+ """
+
+ @classmethod
+ def setUpClass(cls):
+ cls.cr = cls.cursor()
+ cls.uid = openerp.SUPERUSER_ID
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.cr.rollback()
+ cls.cr.close()
+
+
+class HttpCase(SingleTransactionCase):
+ """ Transactionnal HTTP TestCase with a phantomjs helper.
+ """
+
+ def __init__(self, methodName='runTest'):
+ super(HttpCase, self).__init__(methodName)
+ # v8 api with correct xmlrpc exception handling.
+ self.xmlrpc_url = url_8 = 'http://%s:%d/xmlrpc/2/' % (HOST, PORT)
+ self.xmlrpc_common = xmlrpclib.ServerProxy(url_8 + 'common')
+ self.xmlrpc_db = xmlrpclib.ServerProxy(url_8 + 'db')
+ self.xmlrpc_object = xmlrpclib.ServerProxy(url_8 + 'object')
+
+ @classmethod
+ def setUpClass(cls):
+ super(HttpCase, cls).setUpClass()
+ cls.session_id = uuid.uuid4().hex
+ HTTP_SESSION[cls.session_id] = cls.cr
+
+ @classmethod
+ def tearDownClass(cls):
+ del HTTP_SESSION[cls.session_id]
+ super(HttpCase, cls).tearDownClass()
+
+ def phantomjs(self, jsfile, timeout=30, options=None):
+ """ Phantomjs Test protocol.
+
+ Use console.log in phantomjs to output test results evrey line must be
+ a one line JSON message using the following format:
+
+ - for a success: { "event": "success", "message": "Log message" }
+ - for an error: { "event": "error", "message": "Short error description" }
+
+ if a non json parsable line is received the helper will raise an
+ exception, the output buffer will be printed and phantom will be
+ killed
+
+ """
+ self.timeout = timeout
+ self.options = {
+ 'timeout' : timeout,
+ 'port': PORT,
+ 'db': DB,
+ 'user': ADMIN_USER,
+ 'password': ADMIN_PASSWORD,
+ 'session_id': self.session_id,
+ }
+ if options:
+ self.options.update(options)
+
+ self.ignore_filters = [
+ # Ignore phantomjs warnings
+ "*** WARNING:",
+ # Fixes an issue with PhantomJS 1.9.2 on OS X 10.9 (Mavericks)
+ # cf. https://github.com/ariya/phantomjs/issues/11418
+ "CoreText performance note",
+ ]
+
+ cmd = ['phantomjs', jsfile, json.dumps(self.options)]
+ _logger.info('executing %s', cmd)
+ try:
+ phantom = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+ except OSError:
+ _logger.info("phantomjs not found, test %s skipped", jsfile)
+
+ try:
+ t0 = time.time()
+ buf = ''
+ while 1:
+ if time.time() > t0 + self.timeout:
+ raise Exception("Phantom js timeout (%ss)" % self.timeout)
+
+ ready, _, _ = select.select([phantom.stdout], [], [], 0.5)
+ if ready:
+ s = phantom.stdout.read(4096)
+ if s:
+ buf += s
+ else:
+ break
+
+ # process lines
+ if '\n' in buf:
+ line, buf = buf.split('\n', 1)
+ if line not in self.ignore_filters:
+ try:
+ line_json = json.loads(line)
+ if line_json.get('event') == 'success':
+ _logger.info(line_json.get('message','ok'))
+ continue
+ elif line_json.get('event') == 'error':
+ err = line_json.get('message','error')
+ _logger.info(err)
+ else:
+ err = line + buf
+ except ValueError:
+ err = line + buf
+ raise Exception(err)
+ finally:
+ # kill phantomjs if phantom.exit() wasn't called in the test
+ if phantom.poll() is None:
+ phantom.terminate()
+
+
+# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
+++ /dev/null
-# -*- coding: utf-8 -*-
-"""
-Tests for the OpenERP library.
-
-This module groups a few sub-modules containing unittest2 test cases.
-
-Tests can be explicitely added to the `fast_suite` or `checks` lists or not.
-See the :ref:`test-framework` section in the :ref:`features` list.
-"""
-
-import common
-from common import *
-
-
-# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
+++ /dev/null
-# -*- coding: utf-8 -*-
-"""
-The module :mod:`openerp.tests.common` provides a few helpers and classes to write
-tests.
-"""
-import json
-import os
-import select
-import subprocess
-import threading
-import time
-import unittest2
-import uuid
-import xmlrpclib
-import logging
-
-import openerp
-
-_logger = logging.getLogger(__name__)
-
-# The openerp library is supposed already configured.
-ADDONS_PATH = openerp.tools.config['addons_path']
-HOST = '127.0.0.1'
-PORT = openerp.tools.config['xmlrpc_port']
-DB = openerp.tools.config['db_name']
-
-# If the database name is not provided on the command-line,
-# use the one on the thread (which means if it is provided on
-# the command-line, this will break when installing another
-# database from XML-RPC).
-if not DB and hasattr(threading.current_thread(), 'dbname'):
- DB = threading.current_thread().dbname
-
-ADMIN_USER = 'admin'
-ADMIN_USER_ID = openerp.SUPERUSER_ID
-ADMIN_PASSWORD = 'admin'
-
-HTTP_SESSION = {}
-
-class BaseCase(unittest2.TestCase):
- """
- Subclass of TestCase for common OpenERP-specific code.
-
- This class is abstract and expects self.cr and self.uid to be initialized by subclasses.
- """
-
- @classmethod
- def cursor(self):
- return openerp.modules.registry.RegistryManager.get(DB).db.cursor()
-
- @classmethod
- def registry(self, model):
- return openerp.modules.registry.RegistryManager.get(DB)[model]
-
- @classmethod
- def ref(self, xid):
- """ Returns database ID corresponding to a given identifier.
-
- :param xid: fully-qualified record identifier, in the form ``module.identifier``
- :raise: ValueError if not found
- """
- assert "." in xid, "this method requires a fully qualified parameter, in the following form: 'module.identifier'"
- module, xid = xid.split('.')
- _, id = self.registry('ir.model.data').get_object_reference(self.cr, self.uid, module, xid)
- return id
-
- @classmethod
- def browse_ref(self, xid):
- """ Returns a browsable record for the given identifier.
-
- :param xid: fully-qualified record identifier, in the form ``module.identifier``
- :raise: ValueError if not found
- """
- assert "." in xid, "this method requires a fully qualified parameter, in the following form: 'module.identifier'"
- module, xid = xid.split('.')
- return self.registry('ir.model.data').get_object(self.cr, self.uid, module, xid)
-
-
-class TransactionCase(BaseCase):
- """
- Subclass of BaseCase with a single transaction, rolled-back at the end of
- each test (method).
- """
-
- def setUp(self):
- # Store cr and uid in class variables, to allow ref() and browse_ref to be BaseCase @classmethods
- # and still access them
- TransactionCase.cr = self.cursor()
- TransactionCase.uid = openerp.SUPERUSER_ID
-
- def tearDown(self):
- self.cr.rollback()
- self.cr.close()
-
-
-class SingleTransactionCase(BaseCase):
- """
- Subclass of BaseCase with a single transaction for the whole class,
- rolled-back after all the tests.
- """
-
- @classmethod
- def setUpClass(cls):
- cls.cr = cls.cursor()
- cls.uid = openerp.SUPERUSER_ID
-
- @classmethod
- def tearDownClass(cls):
- cls.cr.rollback()
- cls.cr.close()
-
-
-class HttpCase(SingleTransactionCase):
- """ Transactionnal HTTP TestCase with a phantomjs helper.
- """
-
- def __init__(self, methodName='runTest'):
- super(HttpCase, self).__init__(methodName)
- # v8 api with correct xmlrpc exception handling.
- self.xmlrpc_url = url_8 = 'http://%s:%d/xmlrpc/2/' % (HOST, PORT)
- self.xmlrpc_common = xmlrpclib.ServerProxy(url_8 + 'common')
- self.xmlrpc_db = xmlrpclib.ServerProxy(url_8 + 'db')
- self.xmlrpc_object = xmlrpclib.ServerProxy(url_8 + 'object')
-
- @classmethod
- def setUpClass(cls):
- super(HttpCase, cls).setUpClass()
- cls.session_id = uuid.uuid4().hex
- HTTP_SESSION[cls.session_id] = cls.cr
-
- @classmethod
- def tearDownClass(cls):
- del HTTP_SESSION[cls.session_id]
- super(HttpCase, cls).tearDownClass()
-
- def phantomjs(self, jsfile, timeout=30, options=None):
- """ Phantomjs Test protocol.
-
- Use console.log in phantomjs to output test results evrey line must be
- a one line JSON message using the following format:
-
- - for a success: { "event": "success", "message": "Log message" }
- - for an error: { "event": "error", "message": "Short error description" }
-
- if a non json parsable line is received the helper will raise an
- exception, the output buffer will be printed and phantom will be
- killed
-
- """
- self.timeout = timeout
- self.options = {
- 'timeout' : timeout,
- 'port': PORT,
- 'db': DB,
- 'user': ADMIN_USER,
- 'password': ADMIN_PASSWORD,
- 'session_id': self.session_id,
- }
- if options:
- self.options.update(options)
-
- self.ignore_filters = [
- # Ignore phantomjs warnings
- "*** WARNING:",
- # Fixes an issue with PhantomJS 1.9.2 on OS X 10.9 (Mavericks)
- # cf. https://github.com/ariya/phantomjs/issues/11418
- "CoreText performance note",
- ]
-
- cmd = ['phantomjs', jsfile, json.dumps(self.options)]
- _logger.info('executing %s', cmd)
- try:
- phantom = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
- except OSError:
- _logger.info("phantomjs not found, test %s skipped", jsfile)
-
- try:
- t0 = time.time()
- buf = ''
- while 1:
- if time.time() > t0 + self.timeout:
- raise Exception("Phantom js timeout (%ss)" % self.timeout)
-
- ready, _, _ = select.select([phantom.stdout], [], [], 0.5)
- if ready:
- s = phantom.stdout.read(4096)
- if s:
- buf += s
- else:
- break
-
- # process lines
- if '\n' in buf:
- line, buf = buf.split('\n', 1)
- if line not in self.ignore_filters:
- try:
- line_json = json.loads(line)
- if line_json.get('event') == 'success':
- _logger.info(line_json.get('message','ok'))
- continue
- elif line_json.get('event') == 'error':
- err = line_json.get('message','error')
- _logger.info(err)
- else:
- err = line + buf
- except ValueError:
- err = line + buf
- raise Exception(err)
- finally:
- # kill phantomjs if phantom.exit() wasn't called in the test
- if phantom.poll() is None:
- phantom.terminate()
-
-
-# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
+++ /dev/null
-import unittest2
-
-import openerp
-
-
-class test_domain_normalization(unittest2.TestCase):
- def test_normalize_domain(self):
- expression = openerp.osv.expression
- norm_domain = domain = ['&', (1, '=', 1), ('a', '=', 'b')]
- assert norm_domain == expression.normalize_domain(domain), "Normalized domains should be left untouched"
- domain = [('x', 'in', ['y', 'z']), ('a.v', '=', 'e'), '|', '|', ('a', '=', 'b'), '!', ('c', '>', 'd'), ('e', '!=', 'f'), ('g', '=', 'h')]
- norm_domain = ['&', '&', '&'] + domain
- assert norm_domain == expression.normalize_domain(domain), "Non-normalized domains should be properly normalized"
projects / odoo / odoo.git / commitdiff