/
common.py
2030 lines (1750 loc) · 78.3 KB
/
common.py
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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# -*- coding: utf-8 -*-
"""
The module :mod:`odoo.tests.common` provides unittest test cases and a few
helpers and classes to write tests.
"""
import base64
import collections
import importlib
import inspect
import itertools
import json
import logging
import operator
import os
import platform
import re
import requests
import shutil
import signal
import subprocess
import tempfile
import threading
import time
import unittest
import werkzeug.urls
from contextlib import contextmanager
from datetime import datetime, date
from unittest.mock import patch
from decorator import decorator
from lxml import etree, html
from odoo.models import BaseModel
from odoo.osv.expression import normalize_domain
from odoo.tools import float_compare, single_email_re
from odoo.tools.misc import find_in_path
from odoo.tools.safe_eval import safe_eval
try:
from itertools import zip_longest as izip_longest
except ImportError:
from itertools import izip_longest
try:
import websocket
except ImportError:
# chrome headless tests will be skipped
websocket = None
try:
from xmlrpc import client as xmlrpclib
except ImportError:
# pylint: disable=bad-python3-import
import xmlrpclib
import odoo
import pprint
from odoo import api
from odoo.service import security
_logger = logging.getLogger(__name__)
# The odoo library is supposed already configured.
ADDONS_PATH = odoo.tools.config['addons_path']
HOST = '127.0.0.1'
PORT = odoo.tools.config['http_port']
# Useless constant, tests are aware of the content of demo data
ADMIN_USER_ID = odoo.SUPERUSER_ID
def get_db_name():
db = odoo.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'):
return threading.current_thread().dbname
return db
# For backwards-compatibility - get_db_name() should be used instead
DB = get_db_name()
def at_install(flag):
""" Sets the at-install state of a test, the flag is a boolean specifying
whether the test should (``True``) or should not (``False``) run during
module installation.
By default, tests are run right after installing the module, before
starting the installation of the next module.
.. deprecated:: 12.0
``at_install`` is now a flag, you can use :func:`tagged` to
add/remove it, although ``tagged`` only works on test classes
"""
return tagged('at_install' if flag else '-at_install')
def post_install(flag):
""" Sets the post-install state of a test. The flag is a boolean
specifying whether the test should or should not run after a set of
module installations.
By default, tests are *not* run after installation of all modules in the
current installation set.
.. deprecated:: 12.0
``post_install`` is now a flag, you can use :func:`tagged` to
add/remove it, although ``tagged`` only works on test classes
"""
return tagged('post_install' if flag else '-post_install')
def new_test_user(env, login='', groups='base.group_user', context=None, **kwargs):
""" Helper function to create a new test user. It allows to quickly create
users given its login and groups (being a comma separated list of xml ids).
Kwargs are directly propagated to the create to further customize the
created user.
User creation uses a potentially customized environment using the context
parameter allowing to specify a custom context. It can be used to force a
specific behavior and/or simplify record creation. An example is to use
mail-related context keys in mail tests to speedup record creation.
Some specific fields are automatically filled to avoid issues
* groups_id: it is filled using groups function parameter;
* name: "login (groups)" by default as it is required;
* email: it is either the login (if it is a valid email) or a generated
string 'x.x@example.com' (x being the first login letter). This is due
to email being required for most odoo operations;
"""
if not login:
raise ValueError('New users require at least a login')
if not groups:
raise ValueError('New users require at least user groups')
if context is None:
context = {}
groups_id = [(6, 0, [env.ref(g).id for g in groups.split(',')])]
create_values = dict(kwargs, login=login, groups_id=groups_id)
if not create_values.get('name'):
create_values['name'] = '%s (%s)' % (login, groups)
if not create_values.get('email'):
if single_email_re.match(login):
create_values['email'] = login
else:
create_values['email'] = '%s.%s@example.com' % (login[0], login[0])
return env['res.users'].with_context(**context).create(create_values)
# ------------------------------------------------------------
# Main classes
# ------------------------------------------------------------
class TreeCase(unittest.TestCase):
def __init__(self, methodName='runTest'):
super(TreeCase, self).__init__(methodName)
self.addTypeEqualityFunc(etree._Element, self.assertTreesEqual)
self.addTypeEqualityFunc(html.HtmlElement, self.assertTreesEqual)
def assertTreesEqual(self, n1, n2, msg=None):
self.assertIsNotNone(n1, msg)
self.assertIsNotNone(n2, msg)
self.assertEqual(n1.tag, n2.tag, msg)
# Because lxml.attrib is an ordereddict for which order is important
# to equality, even though *we* don't care
self.assertEqual(dict(n1.attrib), dict(n2.attrib), msg)
self.assertEqual((n1.text or u'').strip(), (n2.text or u'').strip(), msg)
self.assertEqual((n1.tail or u'').strip(), (n2.tail or u'').strip(), msg)
for c1, c2 in izip_longest(n1, n2):
self.assertTreesEqual(c1, c2, msg)
class MetaCase(type):
""" Metaclass of test case classes to assign default 'test_tags':
'standard', 'at_install' and the name of the module.
"""
def __init__(cls, name, bases, attrs):
super(MetaCase, cls).__init__(name, bases, attrs)
# assign default test tags
if cls.__module__.startswith('odoo.addons.'):
cls.test_tags = {'standard', 'at_install'}
cls.test_module = cls.__module__.split('.')[2]
cls.test_class = cls.__name__
class BaseCase(TreeCase, MetaCase('DummyCase', (object,), {})):
"""
Subclass of TestCase for common OpenERP-specific code.
This class is abstract and expects self.registry, self.cr and self.uid to be
initialized by subclasses.
"""
longMessage = True # more verbose error message by default: https://www.odoo.com/r/Vmh
warm = True # False during warm-up phase (see :func:`warmup`)
def cursor(self):
return self.registry.cursor()
@property
def uid(self):
""" Get the current uid. """
return self.env.uid
@uid.setter
def uid(self, user):
""" Set the uid by changing the test's environment. """
self.env = self.env(user=user)
def ref(self, xid):
""" Returns database ID for the provided :term:`external identifier`,
shortcut for ``get_object_reference``
:param xid: fully-qualified :term:`external identifier`, in the form
:samp:`{module}.{identifier}`
:raise: ValueError if not found
:returns: registered id
"""
return self.browse_ref(xid).id
def browse_ref(self, xid):
""" Returns a record object for the provided
:term:`external identifier`
:param xid: fully-qualified :term:`external identifier`, in the form
:samp:`{module}.{identifier}`
:raise: ValueError if not found
:returns: :class:`~odoo.models.BaseModel`
"""
assert "." in xid, "this method requires a fully qualified parameter, in the following form: 'module.identifier'"
return self.env.ref(xid)
@contextmanager
def _assertRaises(self, exception):
""" Context manager that clears the environment upon failure. """
with super(BaseCase, self).assertRaises(exception) as cm:
if hasattr(self, 'env'):
with self.env.clear_upon_failure():
yield cm
else:
yield cm
def assertRaises(self, exception, func=None, *args, **kwargs):
if func:
with self._assertRaises(exception):
func(*args, **kwargs)
else:
return self._assertRaises(exception)
@contextmanager
def assertQueryCount(self, default=0, flush=True, **counters):
""" Context manager that counts queries. It may be invoked either with
one value, or with a set of named arguments like ``login=value``::
with self.assertQueryCount(42):
...
with self.assertQueryCount(admin=3, demo=5):
...
The second form is convenient when used with :func:`users`.
"""
if self.warm:
# mock random in order to avoid random bus gc
with self.subTest(), patch('random.random', lambda: 1):
login = self.env.user.login
expected = counters.get(login, default)
if flush:
self.env.user.flush()
count0 = self.cr.sql_log_count
yield
if flush:
self.env.user.flush()
count = self.cr.sql_log_count - count0
if count != expected:
# add some info on caller to allow semi-automatic update of query count
frame, filename, linenum, funcname, lines, index = inspect.stack()[2]
if "/odoo/addons/" in filename:
filename = filename.rsplit("/odoo/addons/", 1)[1]
if count > expected:
msg = "Query count more than expected for user %s: %d > %d in %s at %s:%s"
self.fail(msg % (login, count, expected, funcname, filename, linenum))
else:
logger = logging.getLogger(type(self).__module__)
msg = "Query count less than expected for user %s: %d < %d in %s at %s:%s"
logger.info(msg, login, count, expected, funcname, filename, linenum)
else:
yield
if flush:
self.env.user.flush()
def assertRecordValues(self, records, expected_values):
''' Compare a recordset with a list of dictionaries representing the expected results.
This method performs a comparison element by element based on their index.
Then, the order of the expected values is extremely important.
Note that:
- Comparison between falsy values is supported: False match with None.
- Comparison between monetary field is also treated according the currency's rounding.
- Comparison between x2many field is done by ids. Then, empty expected ids must be [].
- Comparison between many2one field id done by id. Empty comparison can be done using any falsy value.
:param records: The records to compare.
:param expected_values: List of dicts expected to be exactly matched in records
'''
def _compare_candidate(record, candidate):
''' Return True if the candidate matches the given record '''
for field_name in candidate.keys():
record_value = record[field_name]
candidate_value = candidate[field_name]
field = record._fields[field_name]
field_type = field.type
if field_type == 'monetary':
# Compare monetary field.
currency_field_name = record._fields[field_name].currency_field
record_currency = record[currency_field_name]
if record_currency.compare_amounts(candidate_value, record_value)\
if record_currency else candidate_value != record_value:
return False
elif field_type == 'float' and field.get_digits(record.env):
return not float_compare(candidate_value, record_value, precision_digits=field.get_digits(record.env)[1])
elif field_type in ('one2many', 'many2many'):
# Compare x2many relational fields.
# Empty comparison must be an empty list to be True.
if set(record_value.ids) != set(candidate_value):
return False
elif field_type == 'many2one':
# Compare many2one relational fields.
# Every falsy value is allowed to compare with an empty record.
if (record_value or candidate_value) and record_value.id != candidate_value:
return False
elif (candidate_value or record_value) and record_value != candidate_value:
# Compare others fields if not both interpreted as falsy values.
return False
return True
def _repr_field_value(record, field_name):
record_value = record[field_name]
field_type = record._fields[field_name].type
if field_type == 'monetary':
currency_field_name = record._fields[field_name].currency_field
record_currency = record[currency_field_name]
return record_currency and record_currency.round(record_value) or record_value
elif field_type in ('one2many', 'many2many'):
return set(record_value.ids)
elif field_type == 'many2one':
return record_value.id
else:
return record_value
def _format_message(records, expected_values):
''' Return a formatted representation of records/expected_values. '''
all_records_values = [{key: _repr_field_value(record, key) for key in expected_values[0]} for record in records]
msg1 = '\n'.join(pprint.pformat(dic) for dic in all_records_values)
msg2 = '\n'.join(pprint.pformat(dic) for dic in expected_values)
return 'Current values:\n\n%s\n\nExpected values:\n\n%s' % (msg1, msg2)
# if the length or both things to compare is different, we can already tell they're not equal
if len(records) != len(expected_values):
msg = 'Wrong number of records to compare: %d != %d.\n\n' % (len(records), len(expected_values))
self.fail(msg + _format_message(records, expected_values))
candidates = list(expected_values)
for index, record in enumerate(records):
for candidate_index, candidate in enumerate(candidates):
if _compare_candidate(record, candidate):
candidates.pop(candidate_index)
break
else:
msg = 'Record doesn\'t match expected values at index %d.\n\n' % index
self.fail(msg + _format_message(records, expected_values))
def shortDescription(self):
return None
# turns out this thing may not be quite as useful as we thought...
def assertItemsEqual(self, a, b, msg=None):
self.assertCountEqual(a, b, msg=None)
class TransactionCase(BaseCase):
""" TestCase in which each test method is run in its own transaction,
and with its own cursor. The transaction is rolled back and the cursor
is closed after each test.
"""
def setUp(self):
super(TransactionCase, self).setUp()
self.registry = odoo.registry(get_db_name())
#: current transaction's cursor
self.cr = self.cursor()
#: :class:`~odoo.api.Environment` for the current test case
self.env = api.Environment(self.cr, odoo.SUPERUSER_ID, {})
@self.addCleanup
def reset():
# rollback and close the cursor, and reset the environments
self.registry.clear_caches()
self.registry.reset_changes()
self.env.reset()
self.cr.rollback()
self.cr.close()
self.patch(type(self.env['res.partner']), '_get_gravatar_image', lambda *a: False)
def patch(self, obj, key, val):
""" Do the patch ``setattr(obj, key, val)``, and prepare cleanup. """
old = getattr(obj, key)
setattr(obj, key, val)
self.addCleanup(setattr, obj, key, old)
def patch_order(self, model, order):
""" Patch the order of the given model (name), and prepare cleanup. """
self.patch(type(self.env[model]), '_order', order)
class SingleTransactionCase(BaseCase):
""" TestCase in which all test methods are run in the same transaction,
the transaction is started with the first test method and rolled back at
the end of the last.
"""
@classmethod
def setUpClass(cls):
super(SingleTransactionCase, cls).setUpClass()
cls.registry = odoo.registry(get_db_name())
cls.cr = cls.registry.cursor()
cls.env = api.Environment(cls.cr, odoo.SUPERUSER_ID, {})
def setUp(self):
super(SingleTransactionCase, self).setUp()
self.env.user.flush()
@classmethod
def tearDownClass(cls):
# rollback and close the cursor, and reset the environments
cls.registry.clear_caches()
cls.env.reset()
cls.cr.rollback()
cls.cr.close()
super(SingleTransactionCase, cls).tearDownClass()
savepoint_seq = itertools.count()
class SavepointCase(SingleTransactionCase):
""" Similar to :class:`SingleTransactionCase` in that all test methods
are run in a single transaction *but* each test case is run inside a
rollbacked savepoint (sub-transaction).
Useful for test cases containing fast tests but with significant database
setup common to all cases (complex in-db test data): :meth:`~.setUpClass`
can be used to generate db test data once, then all test cases use the
same data without influencing one another but without having to recreate
the test data either.
"""
def setUp(self):
super(SavepointCase, self).setUp()
self._savepoint_id = next(savepoint_seq)
self.cr.execute('SAVEPOINT test_%d' % self._savepoint_id)
def tearDown(self):
self.cr.execute('ROLLBACK TO SAVEPOINT test_%d' % self._savepoint_id)
self.env.clear()
self.registry.clear_caches()
super(SavepointCase, self).tearDown()
class ChromeBrowserException(Exception):
pass
class ChromeBrowser():
""" Helper object to control a Chrome headless process. """
def __init__(self, logger, window_size, test_class):
self._logger = logger
self.test_class = test_class
if websocket is None:
self._logger.warning("websocket-client module is not installed")
raise unittest.SkipTest("websocket-client module is not installed")
self.devtools_port = PORT + 2
self.ws_url = '' # WebSocketUrl
self.ws = None # websocket
self.request_id = 0
self.user_data_dir = tempfile.mkdtemp(suffix='_chrome_odoo')
self.chrome_process = None
otc = odoo.tools.config
self.screenshots_dir = os.path.join(otc['screenshots'], get_db_name(), 'screenshots')
self.screencasts_dir = None
if otc['screencasts']:
if otc['screencasts'] in ('1', 'true', 't'):
self.screencasts_dir = os.path.join(otc['screenshots'], get_db_name(), 'screencasts')
else:
self.screencasts_dir =os.path.join(otc['screencasts'], get_db_name(), 'screencasts')
self.screencast_frames = []
os.makedirs(self.screenshots_dir, exist_ok=True)
self.window_size = window_size
self._chrome_start()
self._find_websocket()
self._logger.info('Websocket url found: %s', self.ws_url)
self._open_websocket()
self._logger.info('Enable chrome headless console log notification')
self._websocket_send('Runtime.enable')
self._logger.info('Chrome headless enable page notifications')
self._websocket_send('Page.enable')
self.sigxcpu_handler = None
if os.name == 'posix':
self.sigxcpu_handler = signal.getsignal(signal.SIGXCPU)
signal.signal(signal.SIGXCPU, self.signal_handler)
def signal_handler(self, sig, frame):
if sig == signal.SIGXCPU:
_logger.info('CPU time limit reached, stopping Chrome and shutting down')
self.stop()
os._exit(0)
def stop(self):
if self.chrome_process is not None:
self._logger.info("Closing chrome headless with pid %s", self.chrome_process.pid)
self._websocket_send('Browser.close')
if self.chrome_process.poll() is None:
self._logger.info("Terminating chrome headless with pid %s", self.chrome_process.pid)
self.chrome_process.terminate()
self.chrome_process.wait()
if self.user_data_dir and os.path.isdir(self.user_data_dir) and self.user_data_dir != '/':
self._logger.info('Removing chrome user profile "%s"', self.user_data_dir)
shutil.rmtree(self.user_data_dir, ignore_errors=True)
# Restore previous signal handler
if self.sigxcpu_handler and os.name == 'posix':
signal.signal(signal.SIGXCPU, self.sigxcpu_handler)
@property
def executable(self):
system = platform.system()
if system == 'Linux':
for bin_ in ['google-chrome', 'chromium', 'chromium-browser']:
try:
return find_in_path(bin_)
except IOError:
continue
elif system == 'Darwin':
bins = [
'/Applications/Google Chrome.app/Contents/MacOS/Google Chrome',
'/Applications/Chromium.app/Contents/MacOS/Chromium',
]
for bin_ in bins:
if os.path.exists(bin_):
return bin_
elif system == 'Windows':
# TODO: handle windows platform: https://stackoverflow.com/a/40674915
pass
raise unittest.SkipTest("Chrome executable not found")
def _chrome_start(self):
if self.chrome_process is not None:
return
switches = {
'--headless': '',
'--enable-logging': 'stderr',
'--no-default-browser-check': '',
'--no-first-run': '',
'--disable-extensions': '',
'--disable-background-networking' : '',
'--disable-background-timer-throttling' : '',
'--disable-backgrounding-occluded-windows': '',
'--disable-renderer-backgrounding' : '',
'--disable-breakpad': '',
'--disable-client-side-phishing-detection': '',
'--disable-crash-reporter': '',
'--disable-default-apps': '',
'--disable-dev-shm-usage': '',
'--disable-device-discovery-notifications': '',
'--disable-namespace-sandbox': '',
'--user-data-dir': self.user_data_dir,
'--disable-translate': '',
'--window-size': self.window_size,
'--remote-debugging-address': HOST,
'--remote-debugging-port': str(self.devtools_port),
'--no-sandbox': '',
}
cmd = [self.executable]
cmd += ['%s=%s' % (k, v) if v else k for k, v in switches.items()]
url = 'about:blank'
cmd.append(url)
self._logger.info('chrome_run executing %s', ' '.join(cmd))
try:
self.chrome_process = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except OSError:
raise unittest.SkipTest("%s not found" % cmd[0])
self._logger.info('Chrome pid: %s', self.chrome_process.pid)
def _find_websocket(self):
version = self._json_command('version')
self._logger.info('Browser version: %s', version['Browser'])
try:
infos = self._json_command('')[0] # Infos about the first tab
except IndexError:
self._logger.warning('No tab found in Chrome')
self.stop()
raise unittest.SkipTest('No tab found in Chrome')
self.ws_url = infos['webSocketDebuggerUrl']
self._logger.info('Chrome headless temporary user profile dir: %s', self.user_data_dir)
def _json_command(self, command, timeout=3):
"""
Inspect dev tools with get
Available commands:
'' : return list of tabs with their id
list (or json/): list tabs
new : open a new tab
activate/ + an id: activate a tab
close/ + and id: close a tab
version : get chrome and dev tools version
protocol : get the full protocol
"""
command = os.path.join('json', command).strip('/')
url = werkzeug.urls.url_join('http://%s:%s/' % (HOST, self.devtools_port), command)
self._logger.info("Issuing json command %s", url)
delay = 0.1
tries = 0
failure_info = None
while tries * delay < timeout:
if self.chrome_process.poll() is not None:
self._logger.error('Chrome crashed at startup with return code', self.chrome_process.returncode)
break
try:
r = requests.get(url, timeout=3)
if r.ok:
self._logger.info("Json command result in %s", tries * delay)
return r.json()
return {'status_code': r.status_code}
except requests.ConnectionError as e:
failure_info = str(e)
time.sleep(delay)
tries+=1
except requests.exceptions.ReadTimeout as e:
failure_info = str(e)
break
self._logger.error('Could not connect to chrome debugger after %s tries, %ss' % (tries, delay))
if failure_info:
self._logger.info(failure_info)
raise unittest.SkipTest("Cannot connect to chrome headless")
def _open_websocket(self):
self.ws = websocket.create_connection(self.ws_url)
if self.ws.getstatus() != 101:
raise unittest.SkipTest("Cannot connect to chrome dev tools")
self.ws.settimeout(0.01)
def _websocket_send(self, method, params=None):
"""
send chrome devtools protocol commands through websocket
"""
sent_id = self.request_id
payload = {
'method': method,
'id': sent_id,
}
if params:
payload.update({'params': params})
self.ws.send(json.dumps(payload))
self.request_id += 1
return sent_id
def _websocket_wait_id(self, awaited_id, timeout=10):
"""
blocking wait for a certain id in a response
warning other messages are discarded
"""
start_time = time.time()
while time.time() - start_time < timeout:
try:
res = json.loads(self.ws.recv())
except websocket.WebSocketTimeoutException:
res = None
if res and res.get('id') == awaited_id:
return res
self._logger.info('timeout exceeded while waiting for id : %d', awaited_id)
return {}
def _websocket_wait_event(self, method, params=None, timeout=10):
"""
blocking wait for a particular event method and eventually a dict of params
"""
start_time = time.time()
while time.time() - start_time < timeout:
try:
res = json.loads(self.ws.recv())
except websocket.WebSocketTimeoutException:
res = None
if res and res.get('method', '') == method:
if params:
if set(params).issubset(set(res.get('params', {}))):
return res
else:
return res
elif res:
self._logger.debug('chrome devtools protocol event: %s', res)
self._logger.info('timeout exceeded while waiting for : %s', method)
def take_screenshot(self, prefix='sc_', suffix=None):
if suffix is None:
suffix = '_%s' % self.test_class
ss_id = self._websocket_send('Page.captureScreenshot')
self._logger.info('Asked for screenshot (id: %s)', ss_id)
res = self._websocket_wait_id(ss_id)
base_png = res.get('result', {}).get('data')
decoded = base64.decodebytes(bytes(base_png.encode('utf-8')))
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S_%f')
fname = '%s%s%s.png' % (prefix, timestamp,suffix)
full_path = os.path.join(self.screenshots_dir, fname)
with open(full_path, 'wb') as f:
f.write(decoded)
self._logger.log(25, 'Screenshot in: %s', full_path)
def _save_screencast(self, prefix='failed'):
# could be encododed with something like that
# ffmpeg -framerate 3 -i frame_%05d.png output.mp4
if not self.screencast_frames:
self._logger.debug('No screencast frames to encode')
return None
for f in self.screencast_frames:
with open(f['file_path'], 'rb') as b64_file:
frame = base64.decodebytes(b64_file.read())
os.unlink(f['file_path'])
f['file_path'] = f['file_path'].replace('.b64', '.png')
with open(f['file_path'], 'wb') as png_file:
png_file.write(frame)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S_%f')
fname = '%s_screencast_%s.mp4' % (prefix, timestamp)
outfile = os.path.join(self.screencasts_dir, fname)
try:
ffmpeg_path = find_in_path('ffmpeg')
except IOError:
ffmpeg_path = None
if ffmpeg_path:
framerate = int(len(self.screencast_frames) / (self.screencast_frames[-1].get('timestamp') - self.screencast_frames[0].get('timestamp')))
r = subprocess.run([ffmpeg_path, '-framerate', str(framerate), '-i', '%s/frame_%%05d.png' % self.screencasts_dir, outfile])
self._logger.log(25, 'Screencast in: %s', outfile)
else:
outfile = outfile.strip('.mp4')
shutil.move(self.screencasts_frames_dir, outfile)
self._logger.log(25, 'Screencast frames in: %s', outfile)
def start_screencast(self):
if self.screencasts_dir:
os.makedirs(self.screencasts_dir, exist_ok=True)
self.screencasts_frames_dir = os.path.join(self.screencasts_dir, 'frames')
os.makedirs(self.screencasts_frames_dir, exist_ok=True)
self._websocket_send('Page.startScreencast', params={'maxWidth': 1024, 'maxHeight': 576})
def set_cookie(self, name, value, path, domain):
params = {'name': name, 'value': value, 'path': path, 'domain': domain}
_id = self._websocket_send('Network.setCookie', params=params)
return self._websocket_wait_id(_id)
def delete_cookie(self, name, **kwargs):
params = {kw:kwargs[kw] for kw in kwargs if kw in ['url', 'domain', 'path']}
params.update({'name': name})
_id = self._websocket_send('Network.deleteCookies', params=params)
return self._websocket_wait_id(_id)
def _wait_ready(self, ready_code, timeout=60):
self._logger.info('Evaluate ready code "%s"', ready_code)
awaited_result = {'result': {'type': 'boolean', 'value': True}}
ready_id = self._websocket_send('Runtime.evaluate', params={'expression': ready_code})
last_bad_res = ''
start_time = time.time()
tdiff = time.time() - start_time
has_exceeded = False
while tdiff < timeout:
try:
res = json.loads(self.ws.recv())
except websocket.WebSocketTimeoutException:
res = None
if res and res.get('id') == ready_id:
if res.get('result') == awaited_result:
if has_exceeded:
self._logger.info('The ready code tooks too much time : %s', tdiff)
return True
else:
last_bad_res = res
ready_id = self._websocket_send('Runtime.evaluate', params={'expression': ready_code})
tdiff = time.time() - start_time
if tdiff >= 2 and not has_exceeded:
has_exceeded = True
self.take_screenshot(prefix='sc_failed_ready_')
self._logger.info('Ready code last try result: %s', last_bad_res or res)
return False
def _wait_code_ok(self, code, timeout):
self._logger.info('Evaluate test code "%s"', code)
code_id = self._websocket_send('Runtime.evaluate', params={'expression': code})
start_time = time.time()
logged_error = False
nb_frame = 0
while time.time() - start_time < timeout:
try:
res = json.loads(self.ws.recv())
except websocket.WebSocketTimeoutException:
res = None
if res and res.get('id', -1) == code_id:
self._logger.info('Code start result: %s', res)
if res.get('result', {}).get('result').get('subtype', '') == 'error':
raise ChromeBrowserException("Running code returned an error: %s" % res)
elif res and res.get('method') == 'Runtime.exceptionThrown':
exception_details = res.get('params', {}).get('exceptionDetails', {})
self.take_screenshot()
self._save_screencast()
raise ChromeBrowserException(exception_details)
elif res and res.get('method') == 'Runtime.consoleAPICalled' and res.get('params', {}).get('type') in ('log', 'error', 'trace'):
logs = res.get('params', {}).get('args')
log_type = res.get('params', {}).get('type')
content = " ".join([str(log.get('value', '')) for log in logs])
if log_type == 'error':
self.take_screenshot()
self._save_screencast()
raise ChromeBrowserException(content)
else:
self._logger.info('console log: %s', content)
if 'test successful' in content:
return True
elif res and res.get('method') == 'Page.screencastFrame':
session_id = res.get('params').get('sessionId')
self._websocket_send('Page.screencastFrameAck', params={'sessionId': int(session_id)})
outfile = os.path.join(self.screencasts_frames_dir, 'frame_%05d.b64' % nb_frame)
frame = res.get('params')
with open(outfile, 'w') as f:
f.write(frame.get('data'))
nb_frame += 1
self.screencast_frames.append({
'file_path': outfile,
'timestamp': frame.get('metadata').get('timestamp')
})
elif res:
self._logger.debug('chrome devtools protocol event: %s', res)
self.take_screenshot()
raise ChromeBrowserException('Script timeout exceeded : %s' % (time.time() - start_time))
def navigate_to(self, url, wait_stop=False):
self._logger.info('Navigating to: "%s"', url)
nav_id = self._websocket_send('Page.navigate', params={'url': url})
nav_result = self._websocket_wait_id(nav_id)
self._logger.info("Navigation result: %s", nav_result)
frame_id = nav_result.get('result', {}).get('frameId', '')
if wait_stop and frame_id:
self._logger.info('Waiting for frame "%s" to stop loading', frame_id)
self._websocket_wait_event('Page.frameStoppedLoading', params={'frameId': frame_id})
def clear(self):
self._websocket_send('Page.stopScreencast')
if self.screencasts_dir and os.path.isdir(self.screencasts_frames_dir):
shutil.rmtree(self.screencasts_frames_dir)
self.screencast_frames = []
sl_id = self._websocket_send('Page.stopLoading')
self._websocket_wait_id(sl_id)
self._logger.info('Deleting cookies and clearing local storage')
dc_id = self._websocket_send('Network.clearBrowserCache')
self._websocket_wait_id(dc_id)
dc_id = self._websocket_send('Network.clearBrowserCookies')
self._websocket_wait_id(dc_id)
cl_id = self._websocket_send('Runtime.evaluate', params={'expression': 'localStorage.clear()'})
self._websocket_wait_id(cl_id)
self.navigate_to('about:blank', wait_stop=True)
class HttpCase(TransactionCase):
""" Transactional HTTP TestCase with url_open and Chrome headless helpers.
"""
registry_test_mode = True
browser = None
browser_size = '1366x768'
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')
cls = type(self)
cls._logger = logging.getLogger('%s.%s' % (cls.__module__, cls.__name__))
@classmethod
def start_browser(cls):
# start browser on demand
if cls.browser is None:
cls.browser = ChromeBrowser(cls._logger, cls.browser_size, cls.__name__)
@classmethod
def tearDownClass(cls):
if cls.browser:
cls.browser.stop()
cls.browser = None
super(HttpCase, cls).tearDownClass()
def setUp(self):
super(HttpCase, self).setUp()
if self.registry_test_mode:
self.registry.enter_test_mode(self.cr)
self.addCleanup(self.registry.leave_test_mode)
# setup a magic session_id that will be rollbacked
self.session = odoo.http.root.session_store.new()
self.session_id = self.session.sid
self.session.db = get_db_name()
odoo.http.root.session_store.save(self.session)
# setup an url opener helper
self.opener = requests.Session()
self.opener.cookies['session_id'] = self.session_id
def url_open(self, url, data=None, files=None, timeout=10, headers=None):
self.env['base'].flush()
if url.startswith('/'):
url = "http://%s:%s%s" % (HOST, PORT, url)
if data or files:
return self.opener.post(url, data=data, files=files, timeout=timeout, headers=headers)
return self.opener.get(url, timeout=timeout, headers=headers)
def _wait_remaining_requests(self, timeout=10):
def get_http_request_threads():
return [t for t in threading.enumerate() if t.name.startswith('odoo.service.http.request.')]
start_time = time.time()
request_threads = get_http_request_threads()
self._logger.info('waiting for threads: %s', request_threads)
for thread in request_threads:
thread.join(timeout - (time.time() - start_time))
request_threads = get_http_request_threads()
for thread in request_threads:
self._logger.info("Stop waiting for thread %s handling request for url %s",
thread.name, getattr(thread, 'url', '<UNKNOWN>'))
if request_threads:
self._logger.info('remaining requests')
odoo.tools.misc.dumpstacks()
def authenticate(self, user, password):
# stay non-authenticated
if user is None:
return
db = get_db_name()
uid = self.registry['res.users'].authenticate(db, user, password, None)
env = api.Environment(self.cr, uid, {})
# self.session.authenticate(db, user, password, uid=uid)
# OpenERPSession.authenticate accesses the current request, which we
# don't have, so reimplement it manually...
session = self.session
session.db = db
session.uid = uid
session.login = user
session.session_token = uid and security.compute_session_token(session, env)
session.context = env['res.users'].context_get() or {}
session.context['uid'] = uid
session._fix_lang(session.context)
odoo.http.root.session_store.save(session)
if self.browser:
self._logger.info('Setting session cookie in browser')
self.browser.set_cookie('session_id', self.session_id, '/', HOST)
def browser_js(self, url_path, code, ready='', login=None, timeout=60, **kw):
""" Test js code running in the browser
- optionnally log as 'login'
- load page given by url_path
- wait for ready object to be available
- eval(code) inside the page
To signal success test do:
console.log('test successful')