diff --git a/odoorpc/db.py b/odoorpc/db.py index 028b7a5..4683031 100644 --- a/odoorpc/db.py +++ b/odoorpc/db.py @@ -7,7 +7,6 @@ import sys from odoorpc import error -from odoorpc.rpc import HTTPError from odoorpc.tools import v from odoorpc.rpc.jsonrpclib import Secret, Bloat @@ -22,7 +21,7 @@ def encode2bytes(data): # Python >= 3 else: - import xml.etree.ElementTree as ET + from html.parser import HTMLParser import requests @@ -32,14 +31,25 @@ def encode2bytes(data): def raise_for_status(response): if response.status_code < 200 or response.status_code > 299: - raise HTTPError("%s - %s " % (response.status_code, response.reason)) + raise error.RPCError("%s - %s " % (response.status_code, response.reason)) # In case of error, Odoo returns it as an HTML page with HTTP 200 if "text/html" in response.headers["Content-Type"]: - clean_body = response.text.replace("&", "&") - html = ET.fromstring(clean_body) - div_error = html.findall(".//div[@class='alert alert-danger']") - if div_error: - raise error.RPCError(div_error[0].text) + parser = AlertHTMLParser() + parser.feed(response.text) + + +class AlertHTMLParser(HTMLParser): + # NOTE HTMLParser is more tolerant regarding invalid markups + # than xml.etree.ElementTree. + def handle_starttag(self, tag, attrs): + attrs_ = dict(attrs) + self.alert_detected = tag == "div" and "alert alert-danger" in attrs_.get( + "class" + ) + + def handle_data(self, data): + if hasattr(self, "alert_detected") and self.alert_detected and data.strip(): + raise error.RPCError(data) class DB(object): @@ -131,7 +141,7 @@ def dump(self, password, db, format_="zip"): :raise: :class:`odoorpc.error.RPCError` (access denied / wrong database) :raise: `urllib.error.URLError` (connection error) """ - if v(self._odoo.version)[0] >= 19: + if v(self._odoo.version)[0] >= 15: url = "/web/database/backup" full_url = self._odoo._connector._proxy_http._get_full_url(url) response = requests.request( @@ -180,7 +190,7 @@ def change_password(self, password, new_password): :raise: :class:`odoorpc.error.RPCError` (access denied) :raise: `urllib.error.URLError` (connection error) """ - if v(self._odoo.version)[0] >= 19: + if v(self._odoo.version)[0] >= 15: url = "/web/database/change_password" full_url = self._odoo._connector._proxy_http._get_full_url(url) response = requests.request( @@ -241,7 +251,7 @@ def create( :raise: :class:`odoorpc.error.RPCError` (access denied) :raise: `urllib.error.URLError` (connection error) """ - if v(self._odoo.version)[0] >= 19: + if v(self._odoo.version)[0] >= 15: url = "/web/database/create" full_url = self._odoo._connector._proxy_http._get_full_url(url) response = requests.request( @@ -290,7 +300,7 @@ def drop(self, password, db): if self._odoo._env and self._odoo._env.db == db: # Remove the existing session to avoid HTTP session error self._odoo.logout() - if v(self._odoo.version)[0] >= 19: + if v(self._odoo.version)[0] >= 15: url = "/web/database/drop" full_url = self._odoo._connector._proxy_http._get_full_url(url) response = requests.request( @@ -333,18 +343,20 @@ def duplicate(self, password, db, new_db, neutralize_database=False): :raise: :class:`odoorpc.error.RPCError` (access denied / wrong database) :raise: `urllib.error.URLError` (connection error) """ - if v(self._odoo.version)[0] >= 19: + if v(self._odoo.version)[0] >= 15: url = "/web/database/duplicate" full_url = self._odoo._connector._proxy_http._get_full_url(url) + data = { + "master_pwd": Secret(password), + "name": db, + "new_name": new_db, + } + if v(self._odoo.version)[0] >= 16: + data["neutralize_database"] = neutralize_database response = requests.request( method="POST", url=full_url, - data={ - "master_pwd": Secret(password), - "name": db, - "new_name": new_db, - "neutralize_database": neutralize_database, - }, + data=data, ) return raise_for_status(response) args = [Secret(password), db, new_db] @@ -387,6 +399,8 @@ def restore(self, password, db, dump, copy=False, neutralize_database=False): The `dump` file object can be obtained with the :func:`dump ` method. If `copy` is set to `True`, the restored database will have a new UUID. + If `neutralize_database` is set to `True`, the duplicated database will + be neutralized (available from Odoo 16+). >>> odoo.db.restore('super_admin_passwd', 'test', dump_file) # doctest: +SKIP @@ -416,18 +430,20 @@ def restore(self, password, db, dump, copy=False, neutralize_database=False): """ if dump.closed: raise error.InternalError("Dump file closed") - if v(self._odoo.version)[0] >= 19: + if v(self._odoo.version)[0] >= 15: url = "/web/database/restore" full_url = self._odoo._connector._proxy_http._get_full_url(url) + data = { + "master_pwd": Secret(password), + "name": db, + "copy": copy, + } + if v(self._odoo.version)[0] >= 16: + data["neutralize_database"] = neutralize_database response = requests.request( method="POST", url=full_url, - data={ - "master_pwd": Secret(password), - "name": db, - "copy": copy, - "neutralize_database": neutralize_database, - }, + data=data, files={"backup_file": ("dump", dump)}, ) return raise_for_status(response) diff --git a/odoorpc/tests/__init__.py b/odoorpc/tests/__init__.py index a5ca51a..25c2a70 100644 --- a/odoorpc/tests/__init__.py +++ b/odoorpc/tests/__init__.py @@ -37,9 +37,9 @@ def setUpClass(cls): port=cls.env["port"], version=cls.env["version"], ) - if v(cls.odoo.version)[0] >= 19: + if v(cls.odoo.version)[0] >= 15: # NOTE: cannot use for test purpose default 'admin' password (lowercase) - # with Odoo 19.0+ and /web/database/ HTTP controllers as they accept + # with Odoo 15.0+ and /web/database/ HTTP controllers as they accept # by default the default password. cls.env["super_pwd"] = "Admin" # Create the database diff --git a/odoorpc/tests/test_db.py b/odoorpc/tests/test_db.py index 02ccb79..08c4e38 100644 --- a/odoorpc/tests/test_db.py +++ b/odoorpc/tests/test_db.py @@ -76,11 +76,7 @@ def test_db_drop(self): self.assertTrue(res) def test_db_drop_wrong_database(self): - res = self.odoo.db.drop(self.env["super_pwd"], "wrong_database") - if v(self.odoo.version)[0] >= 19: - self.assertTrue(res) - else: - self.assertFalse(res) + self.odoo.db.drop(self.env["super_pwd"], "wrong_database") def test_db_drop_wrong_password(self): date = datetime.strftime(datetime.today(), "%Y%m%d_%Hh%Mm%S")