# -*- encoding: utf-8 -*- # Part of Odoo. See LICENSE file for full copyright and licensing details. import logging import operator import re import secrets from io import BytesIO from unittest.mock import patch import requests import odoo from odoo.modules.registry import Registry from odoo.tests.common import BaseCase, HttpCase, tagged from odoo.tools import config class TestDatabaseManager(HttpCase): def test_database_manager(self): if not config['list_db']: return res = self.url_open('/web/database/manager') self.assertEqual(res.status_code, 200) # check that basic existing db actions are present self.assertIn('.o_database_backup', res.text) self.assertIn('.o_database_duplicate', res.text) self.assertIn('.o_database_delete', res.text) # check that basic db actions are present self.assertIn('.o_database_create', res.text) self.assertIn('.o_database_restore', res.text) @tagged('-at_install', 'post_install', '-standard', 'database_operations') class TestDatabaseOperations(BaseCase): def setUp(self): self.password = secrets.token_hex() # monkey-patch password verification self.verify_admin_password_patcher = patch( 'odoo.tools.config.verify_admin_password', self.password.__eq__, ) self.startPatcher(self.verify_admin_password_patcher) self.db_name = config['db_name'] self.assertTrue(self.db_name) # monkey-patch db-filter self.addCleanup(operator.setitem, config, 'dbfilter', config['dbfilter']) config['dbfilter'] = self.db_name + '.*' self.base_databases = self.list_dbs_filtered() self.session = requests.Session() self.session.get(self.url('/web/database/manager')) def tearDown(self): self.assertEqual( self.list_dbs_filtered(), self.base_databases, 'No database should have been created or removed at the end of this test', ) def list_dbs_filtered(self): return set(db for db in odoo.service.db.list_dbs(True) if re.match(config['dbfilter'], db)) def url(self, path): return HttpCase.base_url() + path def assertDbs(self, dbs): self.assertEqual(self.list_dbs_filtered() - self.base_databases, set(dbs)) def url_open_drop(self, dbname): res = self.session.post(self.url('/web/database/drop'), data={ 'master_pwd': self.password, 'name': dbname, }, allow_redirects=False) res.raise_for_status() return res def test_database_creation(self): # check verify_admin_password patch self.assertTrue(odoo.tools.config.verify_admin_password(self.password)) # create a database test_db_name = self.db_name + '-test-database-creation' self.assertNotIn(test_db_name, self.list_dbs_filtered()) res = self.session.post(self.url('/web/database/create'), data={ 'master_pwd': self.password, 'name': test_db_name, 'login': 'admin', 'password': 'admin', 'lang': 'en_US', 'phone': '', }, allow_redirects=False) self.assertEqual(res.status_code, 303) self.assertIn('/odoo', res.headers['Location']) self.assertDbs([test_db_name]) # delete the created database res = self.url_open_drop(test_db_name) self.assertEqual(res.status_code, 303) self.assertIn('/web/database/manager', res.headers['Location']) self.assertDbs([]) def test_database_duplicate(self): # duplicate this database test_db_name = self.db_name + '-test-database-duplicate' self.assertNotIn(test_db_name, self.list_dbs_filtered()) res = self.session.post(self.url('/web/database/duplicate'), data={ 'master_pwd': self.password, 'name': self.db_name, 'new_name': test_db_name, }, allow_redirects=False) self.assertEqual(res.status_code, 303) self.assertIn('/web/database/manager', res.headers['Location']) self.assertDbs([test_db_name]) # delete the created database res = self.url_open_drop(test_db_name) self.assertIn('/web/database/manager', res.headers['Location']) self.assertDbs([]) def test_database_restore(self): test_db_name = self.db_name + '-test-database-restore' self.assertNotIn(test_db_name, self.list_dbs_filtered()) # backup the current database inside a temporary zip file res = self.session.post( self.url('/web/database/backup'), data={ 'master_pwd': self.password, 'name': self.db_name, }, allow_redirects=False, stream=True, ) res.raise_for_status() datetime_pattern = r'\d\d\d\d-\d\d-\d\d_\d\d-\d\d-\d\d' self.assertRegex( res.headers.get('Content-Disposition'), fr"attachment; filename\*=UTF-8''{self.db_name}_{datetime_pattern}\.zip" ) backup_file = BytesIO() backup_file.write(res.content) self.assertGreater(backup_file.tell(), 0, "The backup seems corrupted") # upload the backup under a new name (create a duplicate) with self.subTest(DEFAULT_MAX_CONTENT_LENGTH=None), \ patch.object(odoo.http, 'DEFAULT_MAX_CONTENT_LENGTH', None): backup_file.seek(0) self.session.post( self.url('/web/database/restore'), data={ 'master_pwd': self.password, 'name': test_db_name, 'copy': True, }, files={ 'backup_file': backup_file, }, allow_redirects=False ).raise_for_status() self.assertDbs([test_db_name]) self.url_open_drop(test_db_name) # upload the backup again, this time simulating that the file is # too large under the default size limit, the default size limit # shouldn't apply to /web/database URLs with self.subTest(DEFAULT_MAX_CONTENT_LENGTH=1024), \ patch.object(odoo.http, 'DEFAULT_MAX_CONTENT_LENGTH', 1024): backup_file.seek(0) self.session.post( self.url('/web/database/restore'), data={ 'master_pwd': self.password, 'name': test_db_name, 'copy': True, }, files={ 'backup_file': backup_file, }, allow_redirects=False ).raise_for_status() self.assertDbs([test_db_name]) self.url_open_drop(test_db_name) def test_database_http_registries(self): # This test is about dropping a connection inside one worker and # make sure that the other workers behave correctly. # # Setup # # duplicate this database test_db_name = self.db_name + '-test-database-duplicate' res = self.session.post(self.url('/web/database/duplicate'), data={ 'master_pwd': self.password, 'name': self.db_name, 'new_name': test_db_name, }, allow_redirects=False) # get a registry and a cursor on that new database registry = Registry(test_db_name) cr = registry.cursor() self.assertIn(test_db_name, Registry.registries) # delete the created database but keep the cursor with patch('odoo.sql_db.close_db') as close_db: res = self.url_open_drop(test_db_name) close_db.assert_called_once_with(test_db_name) # simulate that some customers were connected to that dropped db session_store = odoo.http.root.session_store session = session_store.new() session.update(odoo.http.get_default_session(), db=test_db_name) session.context['lang'] = odoo.http.DEFAULT_LANG self.session.cookies['session_id'] = session.sid # make it possible to inject the registry back patcher = patch.dict(Registry.registries.d, {test_db_name: registry}) registries = patcher.start() self.addCleanup(patcher.stop) # # Tests # # The other worker doesn't have a registry in its LRU cache for # that session database. with self.subTest(msg="Registry.init() fails"): session_store.save(session) registries.pop('test_db_name', None) with self.assertLogs('odoo.sql_db', logging.INFO) as capture: res = self.session.get(self.url('/web/health')) self.assertEqual(res.status_code, 200) self.assertEqual(session_store.get(session.sid)['db'], None) self.assertEqual(capture.output, [ "INFO:odoo.sql_db:Connection to the database failed", ]) # The other worker has a registry in its LRU cache for that # session database. But it doesn't have a connection to the sql # database. with self.subTest(msg="Registry.cursor() fails"): session_store.save(session) registries[test_db_name] = registry with self.assertLogs('odoo.sql_db', logging.INFO) as capture, \ patch.object(Registry, '__new__', return_value=registry): res = self.session.get(self.url('/web/health')) self.assertEqual(res.status_code, 200) self.assertEqual(session_store.get(session.sid)['db'], None) self.assertEqual(capture.output, [ "INFO:odoo.sql_db:Connection to the database failed", ]) # The other worker has a registry in its LRU cache for that # session database. It also has a (now broken) connection to the # sql database. with self.subTest(msg="Registry.check_signaling() fails"): session_store.save(session) registries[test_db_name] = registry with self.assertLogs('odoo.sql_db', logging.ERROR) as capture, \ patch.object(Registry, '__new__', return_value=registry), \ patch.object(Registry, 'cursor', return_value=cr): res = self.session.get(self.url('/web/health')) self.assertEqual(res.status_code, 200) self.assertEqual(session_store.get(session.sid)['db'], None) self.maxDiff = None self.assertRegex(capture.output[0], ( r"^ERROR:odoo\.sql_db:bad query:(?s:.*?)" r"ERROR: terminating connection due to administrator command\s+" r"server closed the connection unexpectedly\s+" r"This probably means the server terminated abnormally\s+" r"before or while processing the request\.$" ))