~jfb-tempo-consulting/unifield-wm/sync-env-py3

#!/usr/bin/env python2

# -*- coding: utf-8 -*-

"""

  (C) 2012 OpenERP - All rights reserved

"""

#Load config file

import config

from config import coordo_count, project_count, hq_count, default_oc

import sys

import os

import shutil

import time

import uuid

import argparse

from subprocess import call

import base64

import csv

assert hq_count > 0, "You must have at least one HQ!"

from scripts.common import client, db_instance, Synchro, check_lp_update, get_revno_from_path

if sys.version_info >= (2, 7):

    import unittest

else:

    # Needed for setUpClass and skipIf methods

    import unittest27 as unittest

bool_configuration_only = False

bool_creation_only = False

master_dir = '/'.join(os.path.realpath(__file__).split('/')[0:-1]+['master_dump'])

master_prefix_name = 'msf_profile_sync_so'

dir_to_dump = os.path.join(config.dump_dir, time.strftime('%Y%m%d%H%M'))

def get_file_from_source(filename):

    if filename:

        last = filename.split('/')[-1]

        newfile = os.path.expanduser('~/unifield-server/bin/addons/msf_profile/user_rights/%s' % last)

        if os.path.exists(newfile):

            return newfile

    return filename

def warn(*messages):

    sys.stderr.write(" ".join(messages)+"\n")

# Fake TestCase to enable/disable quickly some tests

class creation_only(unittest.TestCase):

    pass

class configuration_only(unittest.TestCase):

    pass

class skip_all(unittest.TestCase):

    pass

def get_users_from_file(filename):

    LOGIN_COLUMN_INDEX=1

    PASSWD_COLUMN_INDEX=2

    FOR_HQ_COLUMN_INDEX=3

    FOR_COORDO_COLUMN_INDEX=4

    FOR_PROJECT_COLUMN_INDEX=5

    FIRST_GROUP_COLUMN_INDEX=6

    SELECTION_CHAR='X'

    users = []

    with open(filename, 'rb') as csvfile:

        reader = csv.reader(csvfile, delimiter=';')

        header=False

        groups = []

        for row in reader:

            if not header:

                header=True

                for index in range(FIRST_GROUP_COLUMN_INDEX, len(row)):

                    groups.append((index, row[index]))

            else:

                user_groups = []

                for grp in groups:

                    if row[grp[0]] == SELECTION_CHAR:

                        user_groups.append(grp[1])

                data = {

                    'login': row[LOGIN_COLUMN_INDEX],

                    'passwd': row[PASSWD_COLUMN_INDEX],

                    'for_hq': True if row[FOR_HQ_COLUMN_INDEX] == SELECTION_CHAR else False,

                    'for_co': True if row[FOR_COORDO_COLUMN_INDEX] == SELECTION_CHAR else False,

                    'for_pr': True if row[FOR_PROJECT_COLUMN_INDEX] == SELECTION_CHAR else False,

                    'groups': user_groups

                }

                users.append(data)

    return users

# Determin skip flags if needed

skipDrop = True

skipDumpDbs = False

skipBranchesUpdate = True

if __name__ == '__main__':

    parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)

    parser.add_argument("--nodrop", "-n", action='store_true', default=False, help="Don't drop existing db")

    parser.add_argument("--nodump", action='store_true', default=False, help="Disable dbs dump at the end")

    parser.add_argument("--log-to-file", action='store_true', default=False, help="Log the unittest")

    parser.add_argument("--update-code", action='store_true', default=False, help="Update the code, restart servers and create db if needed")

    parser.add_argument('unit_test_option', nargs='*', help='Tests to start: server_creation hq01_creation coordo01_creation dump_all  ...')

    o = parser.parse_args()

    if o.nodump and 'dump_all' not in o.unit_test_option:

        skipDumpDbs = True

    elif o.unit_test_option and 'dump_all' not in o.unit_test_option:

        o.unit_test_option.append('dump_all')

    if o.update_code or 'update_branches' in  o.unit_test_option:

        skipBranchesUpdate = False

        if o.unit_test_option and 'update_branches' not in o.unit_test_option:

            o.unit_test_option.insert(0, 'update_branches')

    sys.argv = [sys.argv[0]] + o.unit_test_option

    skipDrop = o.nodrop

    bool_creation_only = bool('creation_only' in sys.argv) or bool('skip_all' in sys.argv)

    bool_configuration_only = bool('configuration_only' in sys.argv) or bool('skip_all' in sys.argv)

else:

    bool_skip_all = bool(__name__+'.skip_all' in sys.argv)

    if bool_skip_all:

        bool_creation_only = True

        bool_configuration_only = True

    else:

        bool_creation_only = bool(__name__+'.creation_only' in sys.argv)

        bool_configuration_only = bool(__name__+'.configuration_only' in sys.argv)

skipCreation = bool_configuration_only

skipModules = bool_configuration_only

skipSyncSo = bool_configuration_only

skipModuleUpdate = bool_configuration_only

skipUniUser = bool_configuration_only

skipMasterCreation = False

skipGroups = bool_creation_only

skipPropInstance = bool_creation_only

skipConfig = bool_creation_only

skipRegister = bool_creation_only

skipSync = bool_creation_only

skipSync = bool_creation_only

skipModuleData = bool_creation_only

skipPartner = bool_creation_only

skipManualConfig = bool_creation_only

skipOpenPeriod = bool_creation_only

skipLoadUACFile = bool_creation_only

skipCreateUsers = bool_creation_only

skipLoadExtraFiles = bool_creation_only

# eval cond during the run

def skip_test_real_eval(cond, reason):

    def deco(fun):

        def wrap(*a, **b):

            if eval(cond):

                raise unittest.SkipTest(reason)

            return fun(*a, **b)

        return wrap

    return deco

class update_branches(unittest.TestCase):

    @unittest.skipIf(skipBranchesUpdate, "Branches update deactivated")

    def test_01_update_branch(self):

        to_up = check_lp_update(True)

        if 'unifield-web' in to_up:

            if not config.web_restart_cmd:

                raise self.fail('web_restart_cmd not define in config.py')

            call(config.web_restart_cmd, shell=True)

            to_up.remove('unifield-web')

        if to_up:

            if not config.server_restart_cmd:

                raise self.fail('server_restart_cmd not define in config.py')

            call(config.server_restart_cmd, shell=True)

            time.sleep(5)

        if not to_up:

            raise self.fail('No new code to pull')

# Base of database creation

class db_creation(object):

    #ignore_wizard = ['sale.price.setup'] # Fixed in unifield-wm > SP5

    ignore_wizard = ['msf_button_access_rights.view_config_wizard_install']

    base_wizards = {

        'base.setup.config' : {

            'button' : 'config',

        },

        'res.config.view' : {

            'name' : "auto_init",

            'view' : 'extended',

        },

        'sale.price.setup' : {

            'sale_price' : 0.10,

        },

        'stock.location.configuration.wizard' : {

            'location_type' : 'internal',

            'location_usage' : 'stock',

            'location_name' : 'Test Location',

            'button' : 'action_stop',

        },

        'currency.setup' : {

            'functional_id' : config.default_currency,

        } 

    }

    db = None

    parent = None

    @property

    def parent_name(self):

        return self.parent.db.name if self.parent else None

    @classmethod

    def getNameFormat(cls):

        return  {

            'db': config.prefix,

            'ind': cls.index,

            'pind': cls.parent and cls.parent.index or '','ppind': cls.parent and cls.parent.parent and cls.parent.parent.index or ''

        }

    @classmethod

    def setUpClass(cls):

        if cls.db is None and hasattr(cls, 'index'):

            if cls.parent is not None:

                cls.parent.setUpClass()

            name = cls.name_format % cls.getNameFormat()

            if hasattr(config, 'sync_user_admin') and config.sync_user_admin:

                sync_user = 'admin'

            else:

                sync_user = name

            cls.db = db_instance(

                server=client,

                name=name,

                synchro={

                    'protocol' : 'xmlrpc',

                    'host' : config.server_host,

                    'port' : config.server_port,

                    'database' : Synchro.name,

                    'login' : sync_user,

                    'password' : sync_user,

                    'timeout': 600,

                    'netrpc_retry': 10,

                    'xmlrpc_retry': 10,

                },

            )

            last_sync.test_cases.append(cls)

    def setUp(self):

        if self.db is None:

            self.fail("Bad use of class")

    @unittest.skipIf(skipDrop, "Drop DB deactivated")

    def test_00_drop(self):

        self.db.connect('admin')

        self.db.drop()

    @skip_test_real_eval("skipCreation", "Creation desactivated")

    def test_01_create_db(self):

        self.db.connect('admin')

        self.db.create_db(config.admin_password)

        self.db.wait()

        self.db.user('admin').addGroups('Useability / Extended View')

    @skip_test_real_eval("skipModules", "Modules installation desactivated")

    def test_02_base_install(self):

        self.db.connect('admin')

        self.db.module('msf_profile').install().do()

    @skip_test_real_eval("skipSyncSo", "Modules installation desactivated")

    def test_04_sync_so_install(self):

        self.db.connect('admin')

        self.db.module('sync_so').install().do()

        # disable automatic backup

        backup_ids = self.db.get('ir.model').search([('model', '=', 'backup.config')])

        if backup_ids:

            self.db.get('backup.config').write([1], {

                'beforemanualsync': False,

                'beforeautomaticsync': False,

                'aftermanualsync': False,

                'afterautomaticsync': False,

                'scheduledbackup': False

            })

    @skip_test_real_eval("skipUniUser", "UniField user creation desactivated")

    def test_05_unifield_user_creation(self):

        self.db.connect('admin')

        if not hasattr(config, 'load_uac_file') or not config.load_uac_file:

            self.db.user('unifield').add('admin').addGroups('Sync / User', 'Purchase / User')

    def configure(self):

        # We did rather start on msf_instance.setup...

        # Reason: For an unknown reason, this wizard is set as 'done' automatically after run any first wizard

        #model = 'base.setup.installer'

        model = 'msf_instance.setup'

        while model != 'ir.ui.menu':

            try:

                # skip account.installer if no parent_name providen (typically: HQ instance)

                if model in self.ignore_wizard or \

                   (model == 'account.installer' and self.parent_name is not None) or \

                   (model == 'msf_instance.setup' and self.db is Synchro):

                    proxy = self.db.get(model)

                    answer = proxy.action_skip([])

                elif model == 'msf_instance.setup':

                    instance_id = self.db.search_data('msf.instance', [('instance','=',self.db.name)])[0]

                    self.db.get('res.company').write([1], {'instance_id': instance_id})

                    answer = self.db.wizard(model, {'instance_id': instance_id}).action_next()

                else:

                    data = dict(self.base_wizards.get(model, {}))

                    button = data.pop('button', 'action_next')

                    answer = getattr(self.db.wizard(model, data), button)()

                model = answer.get('res_model', None)

            except:

                warn("DEBUG: db=%s, model=%s" % (self.db.name, model))

                raise

    @classmethod

    def sync(cls, db=None):

        if db is None: db = cls.db

        db.connect('admin')

        db.get('sync.client.sync_server_connection').connect()

        if not db.get('sync.client.entity').sync():

            monitor = db.get('sync.monitor')

            ids = monitor.search([], 0, 1, '"end" desc')

            raise Exception('Synchronization process of database "%s" failed!\n%s' % (db.db_name,monitor.read(ids, ['error'])[0]['error']))

    # Create Cost Center and Proprietary Instance for Test Cases

    def make_prop_instance(self, hq, prop_instance=None, mission=None):

        hq.connect('admin')

        # Get 2 cost centers: the top one and the normal one

        cost_center_id = False

        top_cost_center_id = False

        mission_suffix = 'OC'

        if mission and mission.db is hq:

            # coordo

            mission_suffix = "%02d" % self.index

            top_data = {

                'name' : "HT%d" % (self.index),

                'code' : "HT%d" % (self.index),

                'category' : 'OC',

                'type' : 'view',

                'parent_id' : hq.search_data('account.analytic.account', {'Code':'OC'})[0],

            }

            top_cost_center_id = hq.get('account.analytic.account').create(top_data)

            data = {

                'name' : "HT%d01" % (self.index),

                'code' : "HT%d01" % (self.index),

                'category' : 'OC',

                'type' : 'normal',

                'parent_id' : top_cost_center_id,

                'date_start': '2013-01-01',

            }

            cost_center_id = hq.get('account.analytic.account').create(data)

        elif self.db is not hq:

            # project

            mission_suffix = "%02d" % mission.index

            parent_cost_center_id = hq.search_data('account.analytic.account', {'Code':"HT%d" % (mission.index)})[0]

            data = {

                'name' : "HT%d%d1" % (mission.index, self.index),

                'code' : "HT%d%d1" % (mission.index, self.index),

                'category' : 'OC',

                'type' : 'normal',

                'parent_id' : parent_cost_center_id,

                'date_start': '2013-01-01',

            }

            top_cost_center_id = hq.get('account.analytic.account').create(data)

        data = {

            'code' : self.db.name,

            'name' : self.db.name,

            'instance' : self.db.name,

            'mission' : '%s_MISSION_%s' % (config.prefix, mission_suffix),

            'state' : 'active',

        }

        if prop_instance is not None:

            data.update(prop_instance)

        if not hq.test('msf.instance', data):

            instance_id = hq.get('msf.instance').create(data)

            if self.db is not hq:

                # Create/update cost center lines as needed by level

                if mission.db is hq:

                    # Coordo: add cost center lines to the instance, tick both

                    top_line_data = {

                        'instance_id' : instance_id,

                        'cost_center_id' : top_cost_center_id,

                        'is_target' : True,

                        'is_top_cost_center' : True,

                        'is_po_fo_cost_center' : False,

                    }

                    hq.get('account.target.costcenter').create(top_line_data)

                    line_data = {

                        'instance_id' : instance_id,

                        'cost_center_id' : cost_center_id,

                        'is_target' : True,

                        'is_top_cost_center' : False,

                        'is_po_fo_cost_center' : True,

                    }

                    hq.get('account.target.costcenter').create(line_data)

                else:

                    # Project: add cost center lines to parent coordo instance, tick them in instance

                    top_line_data = {

                        'instance_id' : data['parent_id'],

                        'cost_center_id' : top_cost_center_id,

                        'is_target' : False,

                        'is_top_cost_center' : False,

                        'is_po_fo_cost_center' : False,

                    }

                    hq.get('account.target.costcenter').create(top_line_data)

                    project_target_ids = hq.search_data('account.target.costcenter', {'instance_id' : instance_id, 'cost_center_id' : top_cost_center_id})

                    hq.write('account.target.costcenter', project_target_ids, {'is_target': True, 'is_top_cost_center': True, 'is_po_fo_cost_center' : True})

                self.sync(hq)

    def add_to_group(self, group_name, group_type, oc=default_oc):

        Synchro.connect('admin')

        entity_ids = Synchro.get('sync.server.entity').search([('name','=',self.db.name)])

        assert len(entity_ids) == 1, "The entity must exists!"

        # Make groups

        group = Synchro.get('sync.server.entity_group')

        # Make or update OC group

        group_ids = group.search([('name','=',group_name)])

        if group_ids:

            group.write(group_ids, {'entity_ids' : [(4,entity_ids[0])]})

        else:

            Type = Synchro.get('sync.server.group_type')

            type_ids = Type.search([('name', '=', group_type)])

            if not type_ids:

                type_ids = [Type.create({'name':group_type})]

            group.create({

                'name' : group_name,

                'type_id' : type_ids[0],

                'entity_ids' : [(6,0,entity_ids)],

                'oc': oc

            })

    @classmethod

    def dump_db(self, path, name=None):

        if self.db is None:

            self.setUpClass()

        if not os.path.exists(path):

            os.makedirs(path)

        if name is None:

            self.db.connect()

            name =self.db.db_name

        bckfile = os.path.join(path, '%s.dump' % name)

        orig_bck = bckfile

        i = 0

        while os.path.exists(bckfile):

            i += 1

            bckfile = os.path.join(path, '%s_%s.dump' % (name, i))

        if i:

            shutil.move(orig_bck, bckfile)

        self.db.dump_db_file(orig_bck)

    def restore_db(self):

        dump = os.path.join(master_dir, "%s.dump" % (master_prefix_name,) ) #self.db.name)

        self.db.connect('admin')

        self.db.restore_db_file(self.db.name, dump)

        # wait process

        time.sleep(10)

# Run a last sync after all synchronization

class last_sync(unittest.TestCase):

    test_cases = []

    def test_50_last_synchronization(self):

        if not self.test_cases:

            self.skipTest("No database to update")

        for i in [0,1]:

            for tc in self.test_cases:

                assert issubclass(tc, db_creation), "The object %s is not of type db_creation!"

                tc.sync()

class activate_inter_partner(unittest.TestCase):

    def test_99_activate_inter_partner(self):

        all_projects = []

        for tc in test_cases:

            if issubclass(tc, projectn_creation):

                all_projects.append(tc.db.name)

        for tc in test_cases:

            if issubclass(tc, (coordon_creation, projectn_creation)):

                db = tc.db

                db.connect('admin')

                p_obj = db.get('res.partner')

                exclude_name = [db.name]

                same_mission_ids = p_obj.search([('partner_type', '=', 'internal')])

                for p in p_obj.read(same_mission_ids, ['name']):

                    exclude_name.append(p['name'])

                exclude_name += all_projects

                partner_ids = p_obj.search([('partner_type', 'in', ['section', 'intermission']), ('active', '=', False), ('name', 'not in', exclude_name)])

                if partner_ids:

                    p_obj.write(partner_ids, {'active': True})

class dump_all(unittest.TestCase):

    @unittest.skipIf(skipDumpDbs, "DBs dump directory creation deactivated")

    def test_00_create_dump_dir(self):

        if not os.path.exists(dir_to_dump):

            os.makedirs(dir_to_dump)

    @unittest.skipIf(skipDumpDbs, "DBs dump deactivated")

    def test_10_dump_all(self):

        for tc in test_cases:

            if issubclass(tc, db_creation):

                tc.dump_db(dir_to_dump)

    @unittest.skipIf(skipDumpDbs, "DBs dump deactivated")

    def test_20_dump_branch_info(self):

        info = {}

        for ad in config.addons:

            src_path = os.path.join(config.source_path, ad)

            if not os.path.exists(src_path):

                raise self.fail('%s does not exist ! Did you set source_path in config.py ?' % src_path)

            info[ad] = get_revno_from_path(src_path)

        f = open(os.path.join(dir_to_dump, 'info.txt'), 'w')

        for mod, data in info.items():

            f.write("%s_url=%s\n" % (mod, data['lpurl']))

            f.write("%s_revno=%s\n" % (mod, data['revno']))

        f.close()

# Specific Sync Server creation

class server_creation(db_creation, unittest.TestCase):

    db = Synchro

    def test_02_install_lang(self):

        self.db.connect('admin')

        lang = False

        if hasattr(config, 'lang'):

            lang = config.lang

        if lang:

            if self.db.get('sync.client.entity'):

                call(config.server_restart_cmd, shell=True)

                time.sleep(5)

            lang_obj = self.db.get('res.lang')

            lang_id = lang_obj.search([('code', '=', lang)]) 

            mod_obj = self.db.get('ir.module.module')

            if lang_id:

                lang_obj.write(lang_id, {'translatable': True})

                mod_ids = mod_obj.search([('state', '=', 'installed')])

                mod_obj.button_update_translations(mod_ids, lang)

    @unittest.skipIf(skipMasterCreation, "Master dump creation desactivated") 

    def test_03_dump_master(self):

        self.dump_db(master_dir, master_prefix_name)

    @unittest.skipIf(skipModuleUpdate, "update_server installation desactivated")

    def test_10_install_update_server(self):

        self.db.connect('admin')

        self.db.module('update_server').install().do()

    @unittest.skipIf(skipModuleData, "Data module installation desactivated")

    def test_10_install_data_server(self):

        self.db.connect('admin')

        self.db.module('sync_remote_warehouse_server').install().do()

        self.db.module('msf_sync_data_server').install().do()

    @unittest.skipIf(skipConfig, "Modules configuration desactivated")

    def test_30_configuration_wizards(self):