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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
#!/usr/bin/env python2
# -*- coding: utf-8 -*-

#Load common script tests
from scripts.common import Synchro, HQ, Coordo, Project

#Load config file
import config

#Load OpenERP Client Library
import openerplib
from tests.openerplib import *

#Other stuffs
from datetime import datetime
from random import choice

#Load tests procedures
if sys.version_info >= (2, 7):
    import unittest
else:
    # Needed for setUpClass and skipIf methods
    import unittest27 as unittest

try:
    import ipdb as pdb
except:
    import pdb

class finance(unittest.TestCase):

    models = ('res.currency','res.currency.table','res.currency.rate')

    def sync(self, db=None):
        if db is None: db = self.db
        if not db.get('sync.client.entity').sync():
            monitor = db.get('sync.monitor')
            ids = monitor.search([], 0, 1, '"end" desc')
            self.fail('Synchronization process of database "%s" failed!\n%s' % (db.db_name,monitor.read(ids, ['error'])[0]['error']))

    def clean(self):
        for db in (HQ, Coordo, Project):
            db.clean('res.currency.rate', self.rate_data)
            db.clean('res.currency', self.currency_data, active='both', domain=['|','!',('currency_table_id','=',False),('currency_table_id','=',False)])
            db.clean('res.currency.table', self.hq_data)
        for db in (Coordo, Project):
            db.clean('res.currency.table', self.coordo_data)

    def setUp(self):
        Synchro.connect('admin')
        HQ.connect('admin')
        Coordo.connect('admin')
        Project.connect('admin')

    def test_20_currency(self):
        # Make data
        self.id = datetime.now().strftime('%m%d%H%M')
        self.hq_data = {
            'code':'TEST_%s' % self.id,
            'name':'TEST_%s' % self.id,
        }
        self.coordo_data = {
            'code':'TEST_%s_X' % self.id,
        }
        self.currency_data = {
            'accuracy' : 4,
            'currency_name' : 'TEST_%s' % self.id,
            'name' : 'TEST_%s' % self.id,
            'rounding' : 0.01,
            'symbol' : 'T%s' % self.id,
        }
        self.rate_data = {
            'name' : datetime.now().strftime('%Y-%m-%d'),
            'rate' : '0.9999',
        }
        self.clean()
        # Activate rule
        #Synchro.desactivate('sync_server.sync_rule', [])
        n = Synchro.activate('sync_server.sync_rule', [('model_id','in',self.models)])
        #self.assertTrue(n >= len(self.models), "Cannot enable sync rules!")
        # Create & Synchronize
        cur = dict(self.currency_data)
        rate = dict(self.rate_data)
        cur['currency_table_id'] = HQ.get('res.currency.table').create(self.hq_data)
        self.assertTrue(cur['currency_table_id'], "Cannot create currency table!")
        rate['currency_id'] = HQ.get('res.currency').create(cur)
        self.assertTrue(rate['currency_id'], "Cannot create currency!")
        rate_id = HQ.get('res.currency.rate').create(rate)
        self.assertTrue(rate_id, "Cannot create currency rate!")
        HQ.get('res.currency').write(rate['currency_id'], {'active':1})
        for db in (HQ, Coordo, Project,):
            self.sync(db)
        # Check
        for db in (HQ, Coordo, Project,):
            self.assertTrue(db.test('res.currency.table', self.hq_data),
                "The currency table created in HQ is not present in database %s! data=%s" % (db.db_name,self.hq_data))
            self.assertTrue(db.test('res.currency', self.currency_data, active='both', domain=['|','!',('currency_table_id','=',False),('currency_table_id','=',False)]),
                "The currency created in HQ is not present in database %s! data=%s" % (db.db_name,self.currency_data,))
            self.assertTrue(db.test('res.currency.rate', self.rate_data),
                "The currency rate created in HQ is not present in database %s! data=%s" % (db.db_name,self.rate_data))
        # Modify & Synchronize
        n = Coordo.write('res.currency.table', self.hq_data, self.coordo_data)
        self.assertTrue(n == 1,
            "Cannot write coordo data! hq_data=%s, coordo_data=%s" % (self.hq_data, self.coordo_data))
        for db in (Coordo, HQ, Project,):
            self.sync(db)
        # Check
        self.assertFalse(HQ.test('res.currency.table', self.coordo_data),
            "The currency table created in Coordo has been modified in database HQ!")
        self.assertTrue(Project.test('res.currency.table', self.coordo_data),
            "The currency table created in Coordo has not been properly modified in database Project!")

test_cases = (finance,)

def load_tests(loader, tests, pattern):
    suite = unittest.TestSuite()
    for test_class in test_cases:
        tests = loader.loadTestsFromTestCase(test_class)
        suite.addTests(tests)
    return suite

if __name__ == '__main__':
    unittest.main(failfast=True, verbosity=2)