From 83429df27d0a3caf7db49ca43ee266ab7a10f5f1 Mon Sep 17 00:00:00 2001 From: Chris Busillo Date: Sun, 10 May 2026 11:01:44 -0400 Subject: [PATCH 1/2] Add website bootstrap payload support --- docker/scripts/run_odoo_data_workflows.py | 163 ++++++++++++++++- docker/scripts/run_odoo_startup.py | 161 +++++++++++++++- docs/tooling/tenant-overlay.md | 5 + docs/tooling/workspace-cli.md | 6 + odoo_devkit/local_runtime.py | 192 +++++++++++++++++++- tests/test_docker_script_override_guards.py | 6 + tests/test_runtime.py | 112 ++++++++++++ 7 files changed, 639 insertions(+), 6 deletions(-) diff --git a/docker/scripts/run_odoo_data_workflows.py b/docker/scripts/run_odoo_data_workflows.py index a0277c8..98b2b0b 100644 --- a/docker/scripts/run_odoo_data_workflows.py +++ b/docker/scripts/run_odoo_data_workflows.py @@ -1118,26 +1118,185 @@ def apply_environment_overrides(self) -> None: "db": self.local.db_name, } script = textwrap.dedent(""" +import base64 +import binascii import json import os +from pathlib import Path from odoo import api, SUPERUSER_ID from odoo.modules.registry import Registry payload = json.loads('__PAYLOAD__') + +def _load_instance_override_payload(): + encoded_payload = os.environ.get('ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64', '').strip() + if not encoded_payload: + return None + try: + decoded_payload = base64.b64decode(encoded_payload, validate=True) + parsed_payload = json.loads(decoded_payload.decode('utf-8')) + except (ValueError, UnicodeDecodeError, binascii.Error, json.JSONDecodeError) as error: + raise RuntimeError('Invalid Odoo instance override payload.') from error + if not isinstance(parsed_payload, dict): + raise RuntimeError('Odoo instance override payload must decode to an object.') + return parsed_payload + +def _payload_has_launchplane_settings(parsed_payload): + if not parsed_payload: + return False + return bool(parsed_payload.get('config_parameters') or parsed_payload.get('addon_settings')) + +def _field_values(record, values): + return {key: value for key, value in values.items() if key in record._fields} + +def _write_existing_fields(record, values): + filtered_values = _field_values(record, values) + if filtered_values: + record.sudo().write(filtered_values) + +def _module_is_installed(env, module_name): + normalized_module_name = str(module_name or '').strip() + if not normalized_module_name: + return False + module = env['ir.module.module'].sudo().search( + [('name', '=', normalized_module_name), ('state', '=', 'installed')], + limit=1, + ) + return bool(module) + +def _resolve_bootstrap_logo_path(raw_logo_path): + logo_path = str(raw_logo_path or '').strip() + if not logo_path: + return None + candidate_paths = [] + candidate = Path(logo_path) + if candidate.is_absolute(): + candidate_paths.append(candidate) + else: + candidate_paths.append(Path('/opt/project') / logo_path) + candidate_paths.append(Path('/opt/project/addons') / logo_path) + for candidate_path in candidate_paths: + if candidate_path.is_file(): + return candidate_path + formatted_candidates = ', '.join(str(candidate_path) for candidate_path in candidate_paths) + raise RuntimeError(f'Website bootstrap logo file not found: {formatted_candidates}') + +def _find_website_page(env, website, *, xmlid, url): + page = None + if xmlid: + candidate = env.ref(xmlid, raise_if_not_found=False) + if candidate and candidate._name == 'website.page': + page = candidate.sudo() + if page: + return page + if not url: + return None + page_domain = [('url', '=', url)] + if 'website_id' in env['website.page']._fields: + page_domain = ['&', ('url', '=', url), '|', ('website_id', '=', False), ('website_id', '=', website.id)] + return env['website.page'].sudo().search(page_domain, order='website_id desc,id', limit=1) + +def _verify_route(env, website, route_payload, *, fallback_module): + route_url = str(route_payload.get('url') or '').strip() + if not route_url: + return None + module_name = str(route_payload.get('module') or fallback_module or '').strip() + page = _find_website_page(env, website, xmlid='', url=route_url) + if page: + if bool(route_payload.get('published', True)): + _write_existing_fields(page, {'is_published': True, 'website_published': True}) + return page + if module_name: + if not _module_is_installed(env, module_name): + raise RuntimeError(f"Website bootstrap route {route_url!r} requires module {module_name!r}, but it is not installed.") + print(f"Website bootstrap route {route_url} is delegated to installed module {module_name}.") + return None + ir_http = env['ir.http'].sudo() + match = getattr(ir_http, '_match', None) + if callable(match): + try: + match(route_url) + return None + except Exception as error: + raise RuntimeError(f"Website bootstrap route {route_url!r} is not routable.") from error + print(f'Website bootstrap route verification skipped for {route_url}: ir.http._match unavailable.') + return None + +def _apply_website_bootstrap(env, parsed_payload): + if not parsed_payload: + return + website_payload = parsed_payload.get('website_bootstrap') + if not isinstance(website_payload, dict) or not website_payload: + return + if 'website' not in env.registry: + raise RuntimeError('Website bootstrap supplied, but the website module is not installed.') + + website_model = env['website'].sudo() + website = website_model.search([], order='id', limit=1) + if not website: + create_values = _field_values(website_model, {'name': str(website_payload.get('name') or 'Website').strip() or 'Website'}) + website = website_model.create(create_values or {'name': str(website_payload.get('name') or 'Website').strip() or 'Website'}) + + website_values = {} + website_name = str(website_payload.get('name') or '').strip() + if website_name: + website_values['name'] = website_name + canonical_url = str(website_payload.get('canonical_url') or '').strip() + if canonical_url: + env['ir.config_parameter'].sudo().set_param('web.base.url', canonical_url) + env['ir.config_parameter'].sudo().set_param('web.base.url.freeze', 'True') + website_values['domain'] = canonical_url + default_lang = str(website_payload.get('default_lang') or '').strip() + if default_lang and 'default_lang_id' in website._fields: + lang = env['res.lang'].sudo().search([('code', '=', default_lang)], limit=1) + if lang: + website_values['default_lang_id'] = lang.id + logo_path = _resolve_bootstrap_logo_path(website_payload.get('logo_path')) + if logo_path is not None and 'logo' in website._fields: + website_values['logo'] = base64.b64encode(logo_path.read_bytes()).decode('ascii') + _write_existing_fields(website, website_values) + + homepage_url = str(website_payload.get('homepage_url') or '').strip() + primary_page_xmlid = str(website_payload.get('primary_page_xmlid') or '').strip() + homepage_page = _find_website_page(env, website, xmlid=primary_page_xmlid, url=homepage_url) + if homepage_page: + page_values = {'is_published': True, 'website_published': True} + if 'website_id' in homepage_page._fields: + page_values['website_id'] = website.id + _write_existing_fields(homepage_page, page_values) + _write_existing_fields(website, {'homepage_id': homepage_page.id}) + elif primary_page_xmlid: + raise RuntimeError(f"Website bootstrap primary page XML ID not found: {primary_page_xmlid}") + + routes_source = website_payload.get('routes_source') if isinstance(website_payload.get('routes_source'), dict) else {} + fallback_module = str(routes_source.get('module') or '').strip() + if homepage_url and not homepage_page: + _verify_route(env, website, {'url': homepage_url, 'module': fallback_module, 'published': True}, fallback_module=fallback_module) + for route_payload in website_payload.get('routes') or []: + if isinstance(route_payload, dict): + route_page = _verify_route(env, website, route_payload, fallback_module=fallback_module) + if route_page and bool(route_payload.get('homepage')): + _write_existing_fields(website, {'homepage_id': route_page.id}) + + print('website_bootstrap_applied=true') + registry = Registry(payload['db']) with registry.cursor() as cr: env = api.Environment(cr, SUPERUSER_ID, {}) - typed_override_payload_present = bool(os.environ.get('ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64', '').strip()) + instance_override_payload = _load_instance_override_payload() + typed_override_payload_present = instance_override_payload is not None if 'launchplane.settings' in env.registry: env['launchplane.settings'].sudo().apply_from_env() cr.commit() - elif typed_override_payload_present: + elif _payload_has_launchplane_settings(instance_override_payload): raise RuntimeError( 'Launchplane supplied ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64, ' 'but launchplane.settings is not installed.' ) else: print('Launchplane settings addon not installed; skipping settings apply.') + _apply_website_bootstrap(env, instance_override_payload) + cr.commit() """).replace("__PAYLOAD__", json.dumps(payload)) try: diff --git a/docker/scripts/run_odoo_startup.py b/docker/scripts/run_odoo_startup.py index b4fe8dd..c3b408f 100644 --- a/docker/scripts/run_odoo_startup.py +++ b/docker/scripts/run_odoo_startup.py @@ -401,16 +401,173 @@ def _apply_admin_password_if_configured(settings: StartupSettings) -> None: def _apply_environment_overrides_if_available(settings: StartupSettings) -> None: script = """ +import base64 +import binascii +import json import os +from pathlib import Path -typed_override_payload_present = bool(os.environ.get('ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64', '').strip()) +def _load_instance_override_payload(): + encoded_payload = os.environ.get('ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64', '').strip() + if not encoded_payload: + return None + try: + decoded_payload = base64.b64decode(encoded_payload, validate=True) + parsed_payload = json.loads(decoded_payload.decode('utf-8')) + except (ValueError, UnicodeDecodeError, binascii.Error, json.JSONDecodeError) as error: + raise RuntimeError('Invalid Odoo instance override payload.') from error + if not isinstance(parsed_payload, dict): + raise RuntimeError('Odoo instance override payload must decode to an object.') + return parsed_payload + +def _payload_has_launchplane_settings(parsed_payload): + if not parsed_payload: + return False + return bool(parsed_payload.get('config_parameters') or parsed_payload.get('addon_settings')) + +def _field_values(record, values): + return {key: value for key, value in values.items() if key in record._fields} + +def _write_existing_fields(record, values): + filtered_values = _field_values(record, values) + if filtered_values: + record.sudo().write(filtered_values) + +def _module_is_installed(module_name): + normalized_module_name = str(module_name or '').strip() + if not normalized_module_name: + return False + module = env['ir.module.module'].sudo().search( + [('name', '=', normalized_module_name), ('state', '=', 'installed')], + limit=1, + ) + return bool(module) + +def _resolve_bootstrap_logo_path(raw_logo_path): + logo_path = str(raw_logo_path or '').strip() + if not logo_path: + return None + candidate_paths = [] + candidate = Path(logo_path) + if candidate.is_absolute(): + candidate_paths.append(candidate) + else: + candidate_paths.append(Path('/opt/project') / logo_path) + candidate_paths.append(Path('/opt/project/addons') / logo_path) + for candidate_path in candidate_paths: + if candidate_path.is_file(): + return candidate_path + formatted_candidates = ', '.join(str(candidate_path) for candidate_path in candidate_paths) + raise RuntimeError(f'Website bootstrap logo file not found: {formatted_candidates}') + +def _find_website_page(website, *, xmlid, url): + page = None + if xmlid: + candidate = env.ref(xmlid, raise_if_not_found=False) + if candidate and candidate._name == 'website.page': + page = candidate.sudo() + if page: + return page + if not url: + return None + page_domain = [('url', '=', url)] + if 'website_id' in env['website.page']._fields: + page_domain = ['&', ('url', '=', url), '|', ('website_id', '=', False), ('website_id', '=', website.id)] + return env['website.page'].sudo().search(page_domain, order='website_id desc,id', limit=1) + +def _verify_route(website, route_payload, *, fallback_module): + route_url = str(route_payload.get('url') or '').strip() + if not route_url: + return None + module_name = str(route_payload.get('module') or fallback_module or '').strip() + page = _find_website_page(website, xmlid='', url=route_url) + if page: + if bool(route_payload.get('published', True)): + _write_existing_fields(page, {'is_published': True, 'website_published': True}) + return page + if module_name: + if not _module_is_installed(module_name): + raise RuntimeError(f"Website bootstrap route {route_url!r} requires module {module_name!r}, but it is not installed.") + print(f"Website bootstrap route {route_url} is delegated to installed module {module_name}.") + return None + match = getattr(env['ir.http'].sudo(), '_match', None) + if callable(match): + try: + match(route_url) + return None + except Exception as error: + raise RuntimeError(f"Website bootstrap route {route_url!r} is not routable.") from error + print(f'Website bootstrap route verification skipped for {route_url}: ir.http._match unavailable.') + return None + +def _apply_website_bootstrap(parsed_payload): + if not parsed_payload: + return + website_payload = parsed_payload.get('website_bootstrap') + if not isinstance(website_payload, dict) or not website_payload: + return + if 'website' not in env.registry: + raise RuntimeError('Website bootstrap supplied, but the website module is not installed.') + + website_model = env['website'].sudo() + website = website_model.search([], order='id', limit=1) + if not website: + create_values = _field_values(website_model, {'name': str(website_payload.get('name') or 'Website').strip() or 'Website'}) + website = website_model.create(create_values or {'name': str(website_payload.get('name') or 'Website').strip() or 'Website'}) + + website_values = {} + website_name = str(website_payload.get('name') or '').strip() + if website_name: + website_values['name'] = website_name + canonical_url = str(website_payload.get('canonical_url') or '').strip() + if canonical_url: + env['ir.config_parameter'].sudo().set_param('web.base.url', canonical_url) + env['ir.config_parameter'].sudo().set_param('web.base.url.freeze', 'True') + website_values['domain'] = canonical_url + default_lang = str(website_payload.get('default_lang') or '').strip() + if default_lang and 'default_lang_id' in website._fields: + lang = env['res.lang'].sudo().search([('code', '=', default_lang)], limit=1) + if lang: + website_values['default_lang_id'] = lang.id + logo_path = _resolve_bootstrap_logo_path(website_payload.get('logo_path')) + if logo_path is not None and 'logo' in website._fields: + website_values['logo'] = base64.b64encode(logo_path.read_bytes()).decode('ascii') + _write_existing_fields(website, website_values) + + homepage_url = str(website_payload.get('homepage_url') or '').strip() + primary_page_xmlid = str(website_payload.get('primary_page_xmlid') or '').strip() + homepage_page = _find_website_page(website, xmlid=primary_page_xmlid, url=homepage_url) + if homepage_page: + page_values = {'is_published': True, 'website_published': True} + if 'website_id' in homepage_page._fields: + page_values['website_id'] = website.id + _write_existing_fields(homepage_page, page_values) + _write_existing_fields(website, {'homepage_id': homepage_page.id}) + elif primary_page_xmlid: + raise RuntimeError(f"Website bootstrap primary page XML ID not found: {primary_page_xmlid}") + + routes_source = website_payload.get('routes_source') if isinstance(website_payload.get('routes_source'), dict) else {} + fallback_module = str(routes_source.get('module') or '').strip() + if homepage_url and not homepage_page: + _verify_route(website, {'url': homepage_url, 'module': fallback_module, 'published': True}, fallback_module=fallback_module) + for route_payload in website_payload.get('routes') or []: + if isinstance(route_payload, dict): + route_page = _verify_route(website, route_payload, fallback_module=fallback_module) + if route_page and bool(route_payload.get('homepage')): + _write_existing_fields(website, {'homepage_id': route_page.id}) + + print('website_bootstrap_applied=true') + +instance_override_payload = _load_instance_override_payload() +typed_override_payload_present = instance_override_payload is not None if 'launchplane.settings' in env.registry: env['launchplane.settings'].sudo().apply_from_env() -elif typed_override_payload_present: +elif _payload_has_launchplane_settings(instance_override_payload): raise RuntimeError( 'Launchplane supplied ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64, ' 'but launchplane.settings is not installed.' ) +_apply_website_bootstrap(instance_override_payload) env.cr.commit() print('launchplane_settings_applied=true') """ diff --git a/docs/tooling/tenant-overlay.md b/docs/tooling/tenant-overlay.md index c835d5d..3b97cda 100644 --- a/docs/tooling/tenant-overlay.md +++ b/docs/tooling/tenant-overlay.md @@ -28,6 +28,8 @@ When - tenant-specific docs - tracked `workspace.toml` - tracked `artifact-inputs.toml` for runtime and publish-time source inputs +- tracked `website-bootstrap.toml` when the tenant needs Launchplane/devkit to + rebuild public website identity, canonical URL, homepage route, or logo state - tenant-owned code ## Tenant Root Should Not Contain @@ -57,6 +59,9 @@ When - The scaffold includes a repo-owned `artifact-inputs.toml` beside `workspace.toml` so source selection lives in the tenant repo instead of depending on implicit runtime defaults. +- Website bootstrap intent, when present, also lives beside `workspace.toml` as + `website-bootstrap.toml`. Devkit consumes that file during runtime selection + and data workflows apply the resulting typed payload after module install. - Release actions for remote environments still belong in `launchplane`, not in tenant-root `platform runtime` commands. - The generated `Workspace Sync` and `Workspace Status` entrypoints call the diff --git a/docs/tooling/workspace-cli.md b/docs/tooling/workspace-cli.md index 7512636..6e1a3eb 100644 --- a/docs/tooling/workspace-cli.md +++ b/docs/tooling/workspace-cli.md @@ -190,6 +190,12 @@ Notes `launchplane_settings`. `config_parameters` tables write Odoo `ir.config_parameter` keys, while `addon_settings.` tables write supported addon settings such as `authentik_sso` values. +- When a tenant repo contains `website-bootstrap.toml` beside `workspace.toml`, + runtime selection also folds that non-secret website intent into the same + typed payload. The bootstrap contract can add install modules, select the + lane canonical URL, identify a homepage page or controller route, and point at + a repo-local logo asset. Data workflows and startup apply that state + idempotently after modules are installed, without hard-coded tenant defaults. - Legacy setting-shaped inputs such as `ENV_OVERRIDE_CONFIG_PARAM__*`, `ENV_OVERRIDE_AUTHENTIK__*`, and `ENV_OVERRIDE_SHOPIFY__*` are still accepted as a compatibility input and converted into the same typed payload, but they diff --git a/odoo_devkit/local_runtime.py b/odoo_devkit/local_runtime.py index 8165095..fafd1e0 100644 --- a/odoo_devkit/local_runtime.py +++ b/odoo_devkit/local_runtime.py @@ -157,6 +157,7 @@ "ODOO_KEY", "ODOO_ADMIN_LOGIN", "ODOO_ADMIN_PASSWORD", + ODOO_INSTANCE_OVERRIDES_PAYLOAD_ENV_KEY, "ODOO_DATA_WORKFLOW_LOCK_FILE", "ODOO_UPSTREAM_HOST", "ODOO_UPSTREAM_USER", @@ -194,6 +195,31 @@ class OdooOverrideDefinition: addon_settings: dict[str, ScalarMap] +@dataclass(frozen=True) +class WebsiteBootstrapRouteDefinition: + name: str + url: str + module: str | None + published: bool + homepage: bool + + +@dataclass(frozen=True) +class WebsiteBootstrapDefinition: + tenant: str + install_modules: tuple[str, ...] + name: str + default_lang: str | None + homepage_url: str | None + primary_page_xmlid: str | None + logo_path: str | None + logo_alt: str | None + canonical_urls: dict[str, str] + pages_source: dict[str, object] + routes_source: dict[str, object] + routes: tuple[WebsiteBootstrapRouteDefinition, ...] + + @dataclass(frozen=True) class InstanceDefinition: database: str | None @@ -259,6 +285,7 @@ class RuntimeSelection: effective_source_selectors: tuple[str, ...] effective_runtime_env: dict[str, str] effective_odoo_overrides: OdooOverrideDefinition + website_bootstrap: WebsiteBootstrapDefinition | None @dataclass(frozen=True) @@ -756,12 +783,14 @@ def load_runtime_context( manifest=manifest, stack_definition=loaded_stack.stack_definition, ) + website_bootstrap = load_website_bootstrap_definition(manifest=manifest) runtime_selection = resolve_runtime_selection( stack_definition=effective_stack_definition, artifact_inputs_definition=artifact_inputs_definition, context_name=manifest.runtime.context, instance_name=manifest.runtime.instance, repo_root=runtime_repo_path, + website_bootstrap=website_bootstrap, ) runtime_env_file = runtime_env_file_for_scope( repo_root=runtime_repo_path, @@ -809,6 +838,81 @@ def resolve_manifest_runtime_stack_definition(*, manifest: WorkspaceManifest, st ) +def load_website_bootstrap_definition(*, manifest: WorkspaceManifest) -> WebsiteBootstrapDefinition | None: + tenant_repo_path = manifest.tenant_repo.resolve_path(manifest_directory=manifest.manifest_directory) + candidate_paths: list[Path] = [] + if tenant_repo_path is not None: + candidate_paths.append(tenant_repo_path / "website-bootstrap.toml") + manifest_candidate_path = manifest.manifest_directory / "website-bootstrap.toml" + if manifest_candidate_path not in candidate_paths: + candidate_paths.append(manifest_candidate_path) + + bootstrap_path = next((candidate_path for candidate_path in candidate_paths if candidate_path.exists()), None) + if bootstrap_path is None: + return None + + try: + payload = tomllib.loads(bootstrap_path.read_text(encoding="utf-8")) + except (OSError, tomllib.TOMLDecodeError) as error: + raise RuntimeCommandError(f"Invalid website bootstrap file {bootstrap_path}: {error}") from error + return parse_website_bootstrap_definition(payload, bootstrap_path=bootstrap_path, context_name=manifest.runtime.context) + + +def parse_website_bootstrap_definition( + payload: dict[str, object], + *, + bootstrap_path: Path, + context_name: str, +) -> WebsiteBootstrapDefinition: + schema_version = _read_required_int(payload, "schema_version") + if schema_version != 1: + raise RuntimeCommandError(f"Unsupported website bootstrap schema_version in {bootstrap_path}: {schema_version}") + tenant = _read_required_string(payload, "tenant", scope="website-bootstrap") + if tenant != context_name: + raise RuntimeCommandError( + f"Website bootstrap tenant {tenant!r} does not match runtime context {context_name!r} in {bootstrap_path}." + ) + odoo_table = _read_optional_table(payload, "odoo", scope="website-bootstrap") + website_table = _read_required_table(payload, "website", scope="website-bootstrap") + routes: list[WebsiteBootstrapRouteDefinition] = [] + raw_routes = website_table.get("routes") + if raw_routes is not None: + if not isinstance(raw_routes, list): + raise RuntimeCommandError("Expected website-bootstrap.website.routes to be an array of tables when present") + for index, raw_route in enumerate(raw_routes): + route_table = _ensure_table(raw_route, scope=f"website-bootstrap.website.routes[{index}]") + routes.append( + WebsiteBootstrapRouteDefinition( + name=_read_optional_string(route_table, "name", scope=f"website-bootstrap.website.routes[{index}]") or "", + url=_read_required_string(route_table, "url", scope=f"website-bootstrap.website.routes[{index}]"), + module=_read_optional_string(route_table, "module", scope=f"website-bootstrap.website.routes[{index}]"), + published=_read_optional_bool(route_table, "published", default=True), + homepage=_read_optional_bool(route_table, "homepage", default=False), + ) + ) + return WebsiteBootstrapDefinition( + tenant=tenant, + install_modules=_read_optional_string_tuple(odoo_table, "install_modules", scope="website-bootstrap.odoo"), + name=_read_required_string(website_table, "name", scope="website-bootstrap.website"), + default_lang=_read_optional_string(website_table, "default_lang", scope="website-bootstrap.website"), + homepage_url=_read_optional_string(website_table, "homepage_url", scope="website-bootstrap.website"), + primary_page_xmlid=_read_optional_string(website_table, "primary_page_xmlid", scope="website-bootstrap.website"), + logo_path=_read_optional_string(website_table, "logo_path", scope="website-bootstrap.website"), + logo_alt=_read_optional_string(website_table, "logo_alt", scope="website-bootstrap.website"), + canonical_urls={ + key.strip(): value.strip() + for key, value in _read_optional_string_map( + website_table, + "canonical_urls", + scope="website-bootstrap.website", + ).items() + }, + pages_source=_read_optional_table(website_table, "pages_source", scope="website-bootstrap.website"), + routes_source=_read_optional_table(website_table, "routes_source", scope="website-bootstrap.website"), + routes=tuple(routes), + ) + + def resolve_manifest_container_addons_paths(*, manifest: WorkspaceManifest) -> tuple[str, ...]: resolved_paths: list[str] = [] seen_paths: set[str] = set() @@ -1351,6 +1455,7 @@ def resolve_runtime_selection( context_name: str, instance_name: str, repo_root: Path, + website_bootstrap: WebsiteBootstrapDefinition | None = None, ) -> RuntimeSelection: context_definition = stack_definition.contexts.get(context_name) if context_definition is None: @@ -1365,6 +1470,8 @@ def resolve_runtime_selection( effective_install_modules = merge_effective_modules( context_definition=context_definition, instance_definition=instance_definition ) + if website_bootstrap is not None: + effective_install_modules = dedupe_module_names((*effective_install_modules, *website_bootstrap.install_modules)) effective_source_repositories = resolve_runtime_source_repositories( artifact_inputs_definition=artifact_inputs_definition, context_name=context_name, @@ -1409,6 +1516,7 @@ def resolve_runtime_selection( effective_source_selectors=effective_source_selectors, effective_runtime_env=effective_runtime_env, effective_odoo_overrides=effective_odoo_overrides, + website_bootstrap=website_bootstrap, ) @@ -1420,6 +1528,16 @@ def merge_effective_modules(*, context_definition: ContextDefinition, instance_d return tuple(effective_install_modules) +def dedupe_module_names(module_names: Iterable[str]) -> tuple[str, ...]: + effective_module_names: list[str] = [] + for module_name in module_names: + normalized_module_name = module_name.strip() + if not normalized_module_name or normalized_module_name in effective_module_names: + continue + effective_module_names.append(normalized_module_name) + return tuple(effective_module_names) + + def resolve_runtime_source_repositories( *, artifact_inputs_definition: ArtifactInputsDefinition | None, @@ -1620,6 +1738,7 @@ def build_runtime_env_values( context_name=runtime_selection.context_name, instance_name=runtime_selection.instance_name, odoo_overrides=runtime_selection.effective_odoo_overrides, + website_bootstrap=runtime_selection.website_bootstrap, ) return runtime_values @@ -1630,12 +1749,14 @@ def apply_typed_odoo_instance_override_payload( context_name: str, instance_name: str, odoo_overrides: OdooOverrideDefinition | None = None, + website_bootstrap: WebsiteBootstrapDefinition | None = None, ) -> None: payload = build_typed_odoo_instance_override_payload( runtime_values=runtime_values, context_name=context_name, instance_name=instance_name, odoo_overrides=odoo_overrides, + website_bootstrap=website_bootstrap, ) if payload is None: return @@ -1656,6 +1777,7 @@ def build_typed_odoo_instance_override_payload( context_name: str, instance_name: str, odoo_overrides: OdooOverrideDefinition | None = None, + website_bootstrap: WebsiteBootstrapDefinition | None = None, ) -> dict[str, object] | None: config_parameters: list[dict[str, object]] = [] addon_settings: list[dict[str, object]] = [] @@ -1720,15 +1842,56 @@ def build_typed_odoo_instance_override_payload( "value": {"source": "literal", "value": runtime_value}, } ) - if not config_parameters and not addon_settings: + website_bootstrap_payload = render_website_bootstrap_payload( + website_bootstrap=website_bootstrap, + instance_name=instance_name, + ) + if not config_parameters and not addon_settings and website_bootstrap_payload is None: return None - return { + payload: dict[str, object] = { "schema_version": 1, "context": context_name, "instance": instance_name, "config_parameters": config_parameters, "addon_settings": addon_settings, } + if website_bootstrap_payload is not None: + payload["website_bootstrap"] = website_bootstrap_payload + return payload + + +def render_website_bootstrap_payload( + *, + website_bootstrap: WebsiteBootstrapDefinition | None, + instance_name: str, +) -> dict[str, object] | None: + if website_bootstrap is None: + return None + canonical_url = website_bootstrap.canonical_urls.get(instance_name, "").strip() + routes = [ + { + "name": route.name, + "url": route.url, + "module": route.module or "", + "published": route.published, + "homepage": route.homepage, + } + for route in website_bootstrap.routes + ] + payload: dict[str, object] = { + "tenant": website_bootstrap.tenant, + "name": website_bootstrap.name, + "default_lang": website_bootstrap.default_lang or "", + "homepage_url": website_bootstrap.homepage_url or "", + "primary_page_xmlid": website_bootstrap.primary_page_xmlid or "", + "logo_path": website_bootstrap.logo_path or "", + "logo_alt": website_bootstrap.logo_alt or "", + "canonical_url": canonical_url, + "pages_source": website_bootstrap.pages_source, + "routes_source": website_bootstrap.routes_source, + "routes": routes, + } + return payload def apply_publish_artifact_input_manifest( @@ -3016,6 +3179,31 @@ def _read_optional_scalar_map(source: dict[str, object], key: str, *, scope: str return scalar_map +def _read_optional_string_map(source: dict[str, object], key: str, *, scope: str) -> dict[str, str]: + value = source.get(key) + if value is None: + return {} + if not isinstance(value, dict): + raise RuntimeCommandError(f"Expected {scope}.{key} to be a table when present") + string_map: dict[str, str] = {} + for raw_key, raw_value in value.items(): + if not isinstance(raw_key, str) or not raw_key.strip(): + raise RuntimeCommandError(f"Expected {scope}.{key} keys to be non-empty strings") + if not isinstance(raw_value, str): + raise RuntimeCommandError(f"Expected {scope}.{key}.{raw_key} to be a string") + string_map[raw_key] = raw_value + return string_map + + +def _read_optional_bool(source: dict[str, object], key: str, *, default: bool) -> bool: + value = source.get(key) + if value is None: + return default + if not isinstance(value, bool): + raise RuntimeCommandError(f"Expected {key} to be a boolean when present") + return value + + def _read_optional_odoo_override_definition(source: dict[str, object], *, scope: str) -> OdooOverrideDefinition: override_table = _read_optional_table(source, "odoo_overrides", scope=scope) if not override_table: diff --git a/tests/test_docker_script_override_guards.py b/tests/test_docker_script_override_guards.py index fd32c41..8918629 100644 --- a/tests/test_docker_script_override_guards.py +++ b/tests/test_docker_script_override_guards.py @@ -9,9 +9,12 @@ def test_data_workflow_fails_when_typed_override_payload_has_no_consumer(self) - script = (REPO_ROOT / "docker/scripts/run_odoo_data_workflows.py").read_text(encoding="utf-8") self.assertIn("typed_override_payload_present", script) + self.assertIn("_payload_has_launchplane_settings", script) self.assertIn("ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64", script) self.assertIn("launchplane.settings", script) self.assertIn("but launchplane.settings is not installed", script) + self.assertIn("_apply_website_bootstrap", script) + self.assertIn("website_bootstrap_applied=true", script) self.assertNotIn("environment.overrides", script) self.assertNotIn("authentik.sso.config", script) @@ -19,9 +22,12 @@ def test_startup_fails_when_typed_override_payload_has_no_consumer(self) -> None script = (REPO_ROOT / "docker/scripts/run_odoo_startup.py").read_text(encoding="utf-8") self.assertIn("typed_override_payload_present", script) + self.assertIn("_payload_has_launchplane_settings", script) self.assertIn("ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64", script) self.assertIn("launchplane.settings", script) self.assertIn("but launchplane.settings is not installed", script) + self.assertIn("_apply_website_bootstrap", script) + self.assertIn("website_bootstrap_applied=true", script) self.assertNotIn("environment.overrides", script) self.assertNotIn("authentik.sso.config", script) diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 26b946a..0b2c7b4 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -237,6 +237,60 @@ def test_typed_odoo_instance_override_payload_from_stack_overrides(self) -> None ) self.assertEqual(runtime_values["ENV_OVERRIDE_DISABLE_CRON"], "1") + def test_typed_odoo_instance_override_payload_includes_website_bootstrap(self) -> None: + runtime_values: dict[str, str] = {} + website_bootstrap = local_runtime.WebsiteBootstrapDefinition( + tenant="opw", + install_modules=("opw_custom",), + name="OPW", + default_lang="en_US", + homepage_url="/shop", + primary_page_xmlid=None, + logo_path="addons/opw_custom/static/description/icon.png", + logo_alt="OPW", + canonical_urls={"local": "https://opw-local.example.com"}, + pages_source={}, + routes_source={"kind": "controller", "module": "website_sale", "homepage_url": "/shop"}, + routes=( + local_runtime.WebsiteBootstrapRouteDefinition( + name="Shop", + url="/shop", + module="website_sale", + published=True, + homepage=True, + ), + ), + ) + + local_runtime.apply_typed_odoo_instance_override_payload( + runtime_values=runtime_values, + context_name="opw", + instance_name="local", + website_bootstrap=website_bootstrap, + ) + + encoded_payload = runtime_values[local_runtime.ODOO_INSTANCE_OVERRIDES_PAYLOAD_ENV_KEY] + payload = json.loads(base64.b64decode(encoded_payload).decode("utf-8")) + + self.assertEqual(payload["config_parameters"], []) + self.assertEqual(payload["addon_settings"], []) + self.assertEqual(payload["website_bootstrap"]["canonical_url"], "https://opw-local.example.com") + self.assertEqual(payload["website_bootstrap"]["homepage_url"], "/shop") + self.assertEqual(payload["website_bootstrap"]["routes"][0]["module"], "website_sale") + + def test_data_workflow_script_environment_keeps_typed_payload(self) -> None: + environment = { + local_runtime.ODOO_INSTANCE_OVERRIDES_PAYLOAD_ENV_KEY: "encoded-payload", + "ODOO_DB_NAME": "opw", + "UNRELATED": "value", + } + + filtered_environment = local_runtime.data_workflow_script_environment(environment) + + self.assertEqual(filtered_environment[local_runtime.ODOO_INSTANCE_OVERRIDES_PAYLOAD_ENV_KEY], "encoded-payload") + self.assertEqual(filtered_environment["ODOO_DB_NAME"], "opw") + self.assertNotIn("UNRELATED", filtered_environment) + def test_typed_odoo_instance_override_payload_rejects_stack_and_legacy_setting_mix(self) -> None: runtime_values = { "ENV_OVERRIDE_CONFIG_PARAM__WEB__BASE__URL": "https://opw-local.example.com", @@ -817,6 +871,64 @@ def test_native_runtime_select_prefers_manifest_mounts_over_runtime_repo_default runtime_env_text, ) + def test_native_runtime_select_includes_website_bootstrap_payload(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + temp_root = Path(temporary_directory) + tenant_repo_path = temp_root / "tenant-repo" + runtime_repo_path = temp_root / "runtime-repo" + tenant_repo_path.mkdir(parents=True, exist_ok=True) + (tenant_repo_path / "addons").mkdir(parents=True, exist_ok=True) + self._write_runtime_repo(runtime_repo_path) + (tenant_repo_path / "website-bootstrap.toml").write_text( + """ +schema_version = 1 +tenant = "opw" + +[odoo] +install_modules = ["opw_custom", "website_sale"] + +[website] +name = "OPW" +default_lang = "en_US" +homepage_url = "/shop" +logo_path = "addons/opw_custom/static/description/icon.png" +logo_alt = "OPW" + +[website.routes_source] +kind = "controller" +module = "website_sale" +homepage_url = "/shop" + +[website.canonical_urls] +local = "https://opw-local.example.com" +testing = "https://opw-testing.example.com" + +[[website.routes]] +name = "Shop" +url = "/shop" +module = "website_sale" +published = true +homepage = true +""".strip() + + "\n", + encoding="utf-8", + ) + manifest_path = self._write_manifest(tenant_repo_path=tenant_repo_path, runtime_repo_path=runtime_repo_path) + + manifest = load_workspace_manifest(manifest_path) + + with contextlib.redirect_stdout(io.StringIO()): + exit_code = run_native_runtime_select(manifest=manifest) + + self.assertEqual(exit_code, 0) + runtime_values = local_runtime.parse_env_file(runtime_repo_path / ".platform" / "env" / "opw.local.env") + self.assertEqual(runtime_values["ODOO_INSTALL_MODULES"], "opw_custom,website_sale") + encoded_payload = runtime_values[local_runtime.ODOO_INSTANCE_OVERRIDES_PAYLOAD_ENV_KEY] + payload = json.loads(base64.b64decode(encoded_payload).decode("utf-8")) + self.assertEqual(payload["website_bootstrap"]["canonical_url"], "https://opw-local.example.com") + self.assertEqual(payload["website_bootstrap"]["routes_source"]["module"], "website_sale") + self.assertEqual(payload["website_bootstrap"]["routes"][0]["url"], "/shop") + def test_native_runtime_up_runs_compose_up_without_build_when_disabled(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: temp_root = Path(temporary_directory) From f5f63ba27e74b815a9fbb0154bb9ff0422f2d6f9 Mon Sep 17 00:00:00 2001 From: Chris Busillo Date: Sun, 10 May 2026 11:16:46 -0400 Subject: [PATCH 2/2] Share website bootstrap apply helper --- docker/scripts/odoo_website_bootstrap.py | 176 ++++++++++++++++++++ docker/scripts/run_odoo_data_workflows.py | 167 +------------------ docker/scripts/run_odoo_startup.py | 167 +------------------ tests/test_docker_script_override_guards.py | 20 ++- 4 files changed, 206 insertions(+), 324 deletions(-) create mode 100644 docker/scripts/odoo_website_bootstrap.py diff --git a/docker/scripts/odoo_website_bootstrap.py b/docker/scripts/odoo_website_bootstrap.py new file mode 100644 index 0000000..e93e9d3 --- /dev/null +++ b/docker/scripts/odoo_website_bootstrap.py @@ -0,0 +1,176 @@ +import base64 +import binascii +import json +import os +from pathlib import Path +from typing import Any + +ODOO_INSTANCE_OVERRIDES_PAYLOAD_ENV_KEY = "ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64" + + +def load_instance_override_payload() -> dict[str, object] | None: + encoded_payload = os.environ.get(ODOO_INSTANCE_OVERRIDES_PAYLOAD_ENV_KEY, "").strip() + if not encoded_payload: + return None + try: + decoded_payload = base64.b64decode(encoded_payload, validate=True) + parsed_payload = json.loads(decoded_payload.decode("utf-8")) + except (ValueError, UnicodeDecodeError, binascii.Error, json.JSONDecodeError) as error: + raise RuntimeError("Invalid Odoo instance override payload.") from error + if not isinstance(parsed_payload, dict): + raise RuntimeError("Odoo instance override payload must decode to an object.") + return parsed_payload + + +def payload_has_launchplane_settings(parsed_payload: dict[str, object] | None) -> bool: + if not parsed_payload: + return False + return bool(parsed_payload.get("config_parameters") or parsed_payload.get("addon_settings")) + + +def _field_values(record: Any, values: dict[str, object]) -> dict[str, object]: + return {key: value for key, value in values.items() if key in record._fields} + + +def _write_existing_fields(record: Any, values: dict[str, object]) -> None: + filtered_values = _field_values(record, values) + if filtered_values: + record.sudo().write(filtered_values) + + +def _module_is_installed(env: Any, module_name: object) -> bool: + normalized_module_name = str(module_name or "").strip() + if not normalized_module_name: + return False + module = ( + env["ir.module.module"] + .sudo() + .search( + [("name", "=", normalized_module_name), ("state", "=", "installed")], + limit=1, + ) + ) + return bool(module) + + +def _resolve_bootstrap_logo_path(raw_logo_path: object) -> Path | None: + logo_path = str(raw_logo_path or "").strip() + if not logo_path: + return None + candidate_paths: list[Path] = [] + candidate = Path(logo_path) + if candidate.is_absolute(): + candidate_paths.append(candidate) + else: + candidate_paths.append(Path("/opt/project") / logo_path) + candidate_paths.append(Path("/opt/project/addons") / logo_path) + for candidate_path in candidate_paths: + if candidate_path.is_file(): + return candidate_path + formatted_candidates = ", ".join(str(candidate_path) for candidate_path in candidate_paths) + raise RuntimeError(f"Website bootstrap logo file not found: {formatted_candidates}") + + +def _find_website_page(env: Any, website: Any, *, xmlid: str, url: str) -> Any | None: + page = None + if xmlid: + candidate = env.ref(xmlid, raise_if_not_found=False) + if candidate and candidate._name == "website.page": + page = candidate.sudo() + if page: + return page + if not url: + return None + page_domain: list[Any] = [("url", "=", url)] + if "website_id" in env["website.page"]._fields: + page_domain = ["&", ("url", "=", url), "|", ("website_id", "=", False), ("website_id", "=", website.id)] + return env["website.page"].sudo().search(page_domain, order="website_id desc,id", limit=1) + + +def _verify_route(env: Any, website: Any, route_payload: dict[str, object], *, fallback_module: str) -> Any | None: + route_url = str(route_payload.get("url") or "").strip() + if not route_url: + return None + module_name = str(route_payload.get("module") or fallback_module or "").strip() + page = _find_website_page(env, website, xmlid="", url=route_url) + if page: + if bool(route_payload.get("published", True)): + _write_existing_fields(page, {"is_published": True, "website_published": True}) + return page + if module_name: + if not _module_is_installed(env, module_name): + raise RuntimeError(f"Website bootstrap route {route_url!r} requires module {module_name!r}, but it is not installed.") + print(f"Website bootstrap route {route_url} is delegated to installed module {module_name}.") + return None + match = getattr(env["ir.http"].sudo(), "_match", None) + if callable(match): + try: + match(route_url) + return None + except Exception as error: + raise RuntimeError(f"Website bootstrap route {route_url!r} is not routable.") from error + print(f"Website bootstrap route verification skipped for {route_url}: ir.http._match unavailable.") + return None + + +def apply_website_bootstrap(env: Any, parsed_payload: dict[str, object] | None) -> None: + if not parsed_payload: + return + website_payload = parsed_payload.get("website_bootstrap") + if not isinstance(website_payload, dict) or not website_payload: + return + if "website" not in env.registry: + raise RuntimeError("Website bootstrap supplied, but the website module is not installed.") + + website_model = env["website"].sudo() + website = website_model.search([], order="id", limit=1) + if not website: + default_name = str(website_payload.get("name") or "Website").strip() or "Website" + create_values = _field_values(website_model, {"name": default_name}) + website = website_model.create(create_values or {"name": default_name}) + + website_values: dict[str, object] = {} + website_name = str(website_payload.get("name") or "").strip() + if website_name: + website_values["name"] = website_name + canonical_url = str(website_payload.get("canonical_url") or "").strip() + if canonical_url: + env["ir.config_parameter"].sudo().set_param("web.base.url", canonical_url) + env["ir.config_parameter"].sudo().set_param("web.base.url.freeze", "True") + website_values["domain"] = canonical_url + default_lang = str(website_payload.get("default_lang") or "").strip() + if default_lang and "default_lang_id" in website._fields: + lang = env["res.lang"].sudo().search([("code", "=", default_lang)], limit=1) + if lang: + website_values["default_lang_id"] = lang.id + logo_path = _resolve_bootstrap_logo_path(website_payload.get("logo_path")) + if logo_path is not None and "logo" in website._fields: + website_values["logo"] = base64.b64encode(logo_path.read_bytes()).decode("ascii") + _write_existing_fields(website, website_values) + + homepage_url = str(website_payload.get("homepage_url") or "").strip() + primary_page_xmlid = str(website_payload.get("primary_page_xmlid") or "").strip() + homepage_page = _find_website_page(env, website, xmlid=primary_page_xmlid, url=homepage_url) + if homepage_page: + page_values: dict[str, object] = {"is_published": True, "website_published": True} + if "website_id" in homepage_page._fields: + page_values["website_id"] = website.id + _write_existing_fields(homepage_page, page_values) + _write_existing_fields(website, {"homepage_id": homepage_page.id}) + elif primary_page_xmlid: + raise RuntimeError(f"Website bootstrap primary page XML ID not found: {primary_page_xmlid}") + + raw_routes_source = website_payload.get("routes_source") + routes_source = raw_routes_source if isinstance(raw_routes_source, dict) else {} + fallback_module = str(routes_source.get("module") or "").strip() + if homepage_url and not homepage_page: + _verify_route( + env, website, {"url": homepage_url, "module": fallback_module, "published": True}, fallback_module=fallback_module + ) + for route_payload in website_payload.get("routes") or []: + if isinstance(route_payload, dict): + route_page = _verify_route(env, website, route_payload, fallback_module=fallback_module) + if route_page and bool(route_payload.get("homepage")): + _write_existing_fields(website, {"homepage_id": route_page.id}) + + print("website_bootstrap_applied=true") diff --git a/docker/scripts/run_odoo_data_workflows.py b/docker/scripts/run_odoo_data_workflows.py index 98b2b0b..7e2579a 100644 --- a/docker/scripts/run_odoo_data_workflows.py +++ b/docker/scripts/run_odoo_data_workflows.py @@ -1118,184 +1118,33 @@ def apply_environment_overrides(self) -> None: "db": self.local.db_name, } script = textwrap.dedent(""" -import base64 -import binascii import json -import os -from pathlib import Path from odoo import api, SUPERUSER_ID from odoo.modules.registry import Registry +from odoo_website_bootstrap import ( + apply_website_bootstrap, + load_instance_override_payload, + payload_has_launchplane_settings, +) payload = json.loads('__PAYLOAD__') -def _load_instance_override_payload(): - encoded_payload = os.environ.get('ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64', '').strip() - if not encoded_payload: - return None - try: - decoded_payload = base64.b64decode(encoded_payload, validate=True) - parsed_payload = json.loads(decoded_payload.decode('utf-8')) - except (ValueError, UnicodeDecodeError, binascii.Error, json.JSONDecodeError) as error: - raise RuntimeError('Invalid Odoo instance override payload.') from error - if not isinstance(parsed_payload, dict): - raise RuntimeError('Odoo instance override payload must decode to an object.') - return parsed_payload - -def _payload_has_launchplane_settings(parsed_payload): - if not parsed_payload: - return False - return bool(parsed_payload.get('config_parameters') or parsed_payload.get('addon_settings')) - -def _field_values(record, values): - return {key: value for key, value in values.items() if key in record._fields} - -def _write_existing_fields(record, values): - filtered_values = _field_values(record, values) - if filtered_values: - record.sudo().write(filtered_values) - -def _module_is_installed(env, module_name): - normalized_module_name = str(module_name or '').strip() - if not normalized_module_name: - return False - module = env['ir.module.module'].sudo().search( - [('name', '=', normalized_module_name), ('state', '=', 'installed')], - limit=1, - ) - return bool(module) - -def _resolve_bootstrap_logo_path(raw_logo_path): - logo_path = str(raw_logo_path or '').strip() - if not logo_path: - return None - candidate_paths = [] - candidate = Path(logo_path) - if candidate.is_absolute(): - candidate_paths.append(candidate) - else: - candidate_paths.append(Path('/opt/project') / logo_path) - candidate_paths.append(Path('/opt/project/addons') / logo_path) - for candidate_path in candidate_paths: - if candidate_path.is_file(): - return candidate_path - formatted_candidates = ', '.join(str(candidate_path) for candidate_path in candidate_paths) - raise RuntimeError(f'Website bootstrap logo file not found: {formatted_candidates}') - -def _find_website_page(env, website, *, xmlid, url): - page = None - if xmlid: - candidate = env.ref(xmlid, raise_if_not_found=False) - if candidate and candidate._name == 'website.page': - page = candidate.sudo() - if page: - return page - if not url: - return None - page_domain = [('url', '=', url)] - if 'website_id' in env['website.page']._fields: - page_domain = ['&', ('url', '=', url), '|', ('website_id', '=', False), ('website_id', '=', website.id)] - return env['website.page'].sudo().search(page_domain, order='website_id desc,id', limit=1) - -def _verify_route(env, website, route_payload, *, fallback_module): - route_url = str(route_payload.get('url') or '').strip() - if not route_url: - return None - module_name = str(route_payload.get('module') or fallback_module or '').strip() - page = _find_website_page(env, website, xmlid='', url=route_url) - if page: - if bool(route_payload.get('published', True)): - _write_existing_fields(page, {'is_published': True, 'website_published': True}) - return page - if module_name: - if not _module_is_installed(env, module_name): - raise RuntimeError(f"Website bootstrap route {route_url!r} requires module {module_name!r}, but it is not installed.") - print(f"Website bootstrap route {route_url} is delegated to installed module {module_name}.") - return None - ir_http = env['ir.http'].sudo() - match = getattr(ir_http, '_match', None) - if callable(match): - try: - match(route_url) - return None - except Exception as error: - raise RuntimeError(f"Website bootstrap route {route_url!r} is not routable.") from error - print(f'Website bootstrap route verification skipped for {route_url}: ir.http._match unavailable.') - return None - -def _apply_website_bootstrap(env, parsed_payload): - if not parsed_payload: - return - website_payload = parsed_payload.get('website_bootstrap') - if not isinstance(website_payload, dict) or not website_payload: - return - if 'website' not in env.registry: - raise RuntimeError('Website bootstrap supplied, but the website module is not installed.') - - website_model = env['website'].sudo() - website = website_model.search([], order='id', limit=1) - if not website: - create_values = _field_values(website_model, {'name': str(website_payload.get('name') or 'Website').strip() or 'Website'}) - website = website_model.create(create_values or {'name': str(website_payload.get('name') or 'Website').strip() or 'Website'}) - - website_values = {} - website_name = str(website_payload.get('name') or '').strip() - if website_name: - website_values['name'] = website_name - canonical_url = str(website_payload.get('canonical_url') or '').strip() - if canonical_url: - env['ir.config_parameter'].sudo().set_param('web.base.url', canonical_url) - env['ir.config_parameter'].sudo().set_param('web.base.url.freeze', 'True') - website_values['domain'] = canonical_url - default_lang = str(website_payload.get('default_lang') or '').strip() - if default_lang and 'default_lang_id' in website._fields: - lang = env['res.lang'].sudo().search([('code', '=', default_lang)], limit=1) - if lang: - website_values['default_lang_id'] = lang.id - logo_path = _resolve_bootstrap_logo_path(website_payload.get('logo_path')) - if logo_path is not None and 'logo' in website._fields: - website_values['logo'] = base64.b64encode(logo_path.read_bytes()).decode('ascii') - _write_existing_fields(website, website_values) - - homepage_url = str(website_payload.get('homepage_url') or '').strip() - primary_page_xmlid = str(website_payload.get('primary_page_xmlid') or '').strip() - homepage_page = _find_website_page(env, website, xmlid=primary_page_xmlid, url=homepage_url) - if homepage_page: - page_values = {'is_published': True, 'website_published': True} - if 'website_id' in homepage_page._fields: - page_values['website_id'] = website.id - _write_existing_fields(homepage_page, page_values) - _write_existing_fields(website, {'homepage_id': homepage_page.id}) - elif primary_page_xmlid: - raise RuntimeError(f"Website bootstrap primary page XML ID not found: {primary_page_xmlid}") - - routes_source = website_payload.get('routes_source') if isinstance(website_payload.get('routes_source'), dict) else {} - fallback_module = str(routes_source.get('module') or '').strip() - if homepage_url and not homepage_page: - _verify_route(env, website, {'url': homepage_url, 'module': fallback_module, 'published': True}, fallback_module=fallback_module) - for route_payload in website_payload.get('routes') or []: - if isinstance(route_payload, dict): - route_page = _verify_route(env, website, route_payload, fallback_module=fallback_module) - if route_page and bool(route_payload.get('homepage')): - _write_existing_fields(website, {'homepage_id': route_page.id}) - - print('website_bootstrap_applied=true') - registry = Registry(payload['db']) with registry.cursor() as cr: env = api.Environment(cr, SUPERUSER_ID, {}) - instance_override_payload = _load_instance_override_payload() + instance_override_payload = load_instance_override_payload() typed_override_payload_present = instance_override_payload is not None if 'launchplane.settings' in env.registry: env['launchplane.settings'].sudo().apply_from_env() cr.commit() - elif _payload_has_launchplane_settings(instance_override_payload): + elif payload_has_launchplane_settings(instance_override_payload): raise RuntimeError( 'Launchplane supplied ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64, ' 'but launchplane.settings is not installed.' ) else: print('Launchplane settings addon not installed; skipping settings apply.') - _apply_website_bootstrap(env, instance_override_payload) + apply_website_bootstrap(env, instance_override_payload) cr.commit() """).replace("__PAYLOAD__", json.dumps(payload)) diff --git a/docker/scripts/run_odoo_startup.py b/docker/scripts/run_odoo_startup.py index c3b408f..546bb4d 100644 --- a/docker/scripts/run_odoo_startup.py +++ b/docker/scripts/run_odoo_startup.py @@ -401,173 +401,22 @@ def _apply_admin_password_if_configured(settings: StartupSettings) -> None: def _apply_environment_overrides_if_available(settings: StartupSettings) -> None: script = """ -import base64 -import binascii -import json -import os -from pathlib import Path +from odoo_website_bootstrap import ( + apply_website_bootstrap, + load_instance_override_payload, + payload_has_launchplane_settings, +) -def _load_instance_override_payload(): - encoded_payload = os.environ.get('ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64', '').strip() - if not encoded_payload: - return None - try: - decoded_payload = base64.b64decode(encoded_payload, validate=True) - parsed_payload = json.loads(decoded_payload.decode('utf-8')) - except (ValueError, UnicodeDecodeError, binascii.Error, json.JSONDecodeError) as error: - raise RuntimeError('Invalid Odoo instance override payload.') from error - if not isinstance(parsed_payload, dict): - raise RuntimeError('Odoo instance override payload must decode to an object.') - return parsed_payload - -def _payload_has_launchplane_settings(parsed_payload): - if not parsed_payload: - return False - return bool(parsed_payload.get('config_parameters') or parsed_payload.get('addon_settings')) - -def _field_values(record, values): - return {key: value for key, value in values.items() if key in record._fields} - -def _write_existing_fields(record, values): - filtered_values = _field_values(record, values) - if filtered_values: - record.sudo().write(filtered_values) - -def _module_is_installed(module_name): - normalized_module_name = str(module_name or '').strip() - if not normalized_module_name: - return False - module = env['ir.module.module'].sudo().search( - [('name', '=', normalized_module_name), ('state', '=', 'installed')], - limit=1, - ) - return bool(module) - -def _resolve_bootstrap_logo_path(raw_logo_path): - logo_path = str(raw_logo_path or '').strip() - if not logo_path: - return None - candidate_paths = [] - candidate = Path(logo_path) - if candidate.is_absolute(): - candidate_paths.append(candidate) - else: - candidate_paths.append(Path('/opt/project') / logo_path) - candidate_paths.append(Path('/opt/project/addons') / logo_path) - for candidate_path in candidate_paths: - if candidate_path.is_file(): - return candidate_path - formatted_candidates = ', '.join(str(candidate_path) for candidate_path in candidate_paths) - raise RuntimeError(f'Website bootstrap logo file not found: {formatted_candidates}') - -def _find_website_page(website, *, xmlid, url): - page = None - if xmlid: - candidate = env.ref(xmlid, raise_if_not_found=False) - if candidate and candidate._name == 'website.page': - page = candidate.sudo() - if page: - return page - if not url: - return None - page_domain = [('url', '=', url)] - if 'website_id' in env['website.page']._fields: - page_domain = ['&', ('url', '=', url), '|', ('website_id', '=', False), ('website_id', '=', website.id)] - return env['website.page'].sudo().search(page_domain, order='website_id desc,id', limit=1) - -def _verify_route(website, route_payload, *, fallback_module): - route_url = str(route_payload.get('url') or '').strip() - if not route_url: - return None - module_name = str(route_payload.get('module') or fallback_module or '').strip() - page = _find_website_page(website, xmlid='', url=route_url) - if page: - if bool(route_payload.get('published', True)): - _write_existing_fields(page, {'is_published': True, 'website_published': True}) - return page - if module_name: - if not _module_is_installed(module_name): - raise RuntimeError(f"Website bootstrap route {route_url!r} requires module {module_name!r}, but it is not installed.") - print(f"Website bootstrap route {route_url} is delegated to installed module {module_name}.") - return None - match = getattr(env['ir.http'].sudo(), '_match', None) - if callable(match): - try: - match(route_url) - return None - except Exception as error: - raise RuntimeError(f"Website bootstrap route {route_url!r} is not routable.") from error - print(f'Website bootstrap route verification skipped for {route_url}: ir.http._match unavailable.') - return None - -def _apply_website_bootstrap(parsed_payload): - if not parsed_payload: - return - website_payload = parsed_payload.get('website_bootstrap') - if not isinstance(website_payload, dict) or not website_payload: - return - if 'website' not in env.registry: - raise RuntimeError('Website bootstrap supplied, but the website module is not installed.') - - website_model = env['website'].sudo() - website = website_model.search([], order='id', limit=1) - if not website: - create_values = _field_values(website_model, {'name': str(website_payload.get('name') or 'Website').strip() or 'Website'}) - website = website_model.create(create_values or {'name': str(website_payload.get('name') or 'Website').strip() or 'Website'}) - - website_values = {} - website_name = str(website_payload.get('name') or '').strip() - if website_name: - website_values['name'] = website_name - canonical_url = str(website_payload.get('canonical_url') or '').strip() - if canonical_url: - env['ir.config_parameter'].sudo().set_param('web.base.url', canonical_url) - env['ir.config_parameter'].sudo().set_param('web.base.url.freeze', 'True') - website_values['domain'] = canonical_url - default_lang = str(website_payload.get('default_lang') or '').strip() - if default_lang and 'default_lang_id' in website._fields: - lang = env['res.lang'].sudo().search([('code', '=', default_lang)], limit=1) - if lang: - website_values['default_lang_id'] = lang.id - logo_path = _resolve_bootstrap_logo_path(website_payload.get('logo_path')) - if logo_path is not None and 'logo' in website._fields: - website_values['logo'] = base64.b64encode(logo_path.read_bytes()).decode('ascii') - _write_existing_fields(website, website_values) - - homepage_url = str(website_payload.get('homepage_url') or '').strip() - primary_page_xmlid = str(website_payload.get('primary_page_xmlid') or '').strip() - homepage_page = _find_website_page(website, xmlid=primary_page_xmlid, url=homepage_url) - if homepage_page: - page_values = {'is_published': True, 'website_published': True} - if 'website_id' in homepage_page._fields: - page_values['website_id'] = website.id - _write_existing_fields(homepage_page, page_values) - _write_existing_fields(website, {'homepage_id': homepage_page.id}) - elif primary_page_xmlid: - raise RuntimeError(f"Website bootstrap primary page XML ID not found: {primary_page_xmlid}") - - routes_source = website_payload.get('routes_source') if isinstance(website_payload.get('routes_source'), dict) else {} - fallback_module = str(routes_source.get('module') or '').strip() - if homepage_url and not homepage_page: - _verify_route(website, {'url': homepage_url, 'module': fallback_module, 'published': True}, fallback_module=fallback_module) - for route_payload in website_payload.get('routes') or []: - if isinstance(route_payload, dict): - route_page = _verify_route(website, route_payload, fallback_module=fallback_module) - if route_page and bool(route_payload.get('homepage')): - _write_existing_fields(website, {'homepage_id': route_page.id}) - - print('website_bootstrap_applied=true') - -instance_override_payload = _load_instance_override_payload() +instance_override_payload = load_instance_override_payload() typed_override_payload_present = instance_override_payload is not None if 'launchplane.settings' in env.registry: env['launchplane.settings'].sudo().apply_from_env() -elif _payload_has_launchplane_settings(instance_override_payload): +elif payload_has_launchplane_settings(instance_override_payload): raise RuntimeError( 'Launchplane supplied ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64, ' 'but launchplane.settings is not installed.' ) -_apply_website_bootstrap(instance_override_payload) +apply_website_bootstrap(env, instance_override_payload) env.cr.commit() print('launchplane_settings_applied=true') """ diff --git a/tests/test_docker_script_override_guards.py b/tests/test_docker_script_override_guards.py index 8918629..d904f04 100644 --- a/tests/test_docker_script_override_guards.py +++ b/tests/test_docker_script_override_guards.py @@ -9,12 +9,12 @@ def test_data_workflow_fails_when_typed_override_payload_has_no_consumer(self) - script = (REPO_ROOT / "docker/scripts/run_odoo_data_workflows.py").read_text(encoding="utf-8") self.assertIn("typed_override_payload_present", script) - self.assertIn("_payload_has_launchplane_settings", script) + self.assertIn("payload_has_launchplane_settings", script) self.assertIn("ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64", script) self.assertIn("launchplane.settings", script) self.assertIn("but launchplane.settings is not installed", script) - self.assertIn("_apply_website_bootstrap", script) - self.assertIn("website_bootstrap_applied=true", script) + self.assertIn("apply_website_bootstrap", script) + self.assertIn("from odoo_website_bootstrap import", script) self.assertNotIn("environment.overrides", script) self.assertNotIn("authentik.sso.config", script) @@ -22,15 +22,23 @@ def test_startup_fails_when_typed_override_payload_has_no_consumer(self) -> None script = (REPO_ROOT / "docker/scripts/run_odoo_startup.py").read_text(encoding="utf-8") self.assertIn("typed_override_payload_present", script) - self.assertIn("_payload_has_launchplane_settings", script) + self.assertIn("payload_has_launchplane_settings", script) self.assertIn("ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64", script) self.assertIn("launchplane.settings", script) self.assertIn("but launchplane.settings is not installed", script) - self.assertIn("_apply_website_bootstrap", script) - self.assertIn("website_bootstrap_applied=true", script) + self.assertIn("apply_website_bootstrap", script) + self.assertIn("from odoo_website_bootstrap import", script) self.assertNotIn("environment.overrides", script) self.assertNotIn("authentik.sso.config", script) + def test_website_bootstrap_helper_is_part_of_docker_payload(self) -> None: + dockerfile = (REPO_ROOT / "docker/Dockerfile").read_text(encoding="utf-8") + helper = (REPO_ROOT / "docker/scripts/odoo_website_bootstrap.py").read_text(encoding="utf-8") + + self.assertIn("COPY /docker/scripts /payload/volumes/scripts", dockerfile) + self.assertIn("def apply_website_bootstrap", helper) + self.assertIn("website_bootstrap_applied=true", helper) + if __name__ == "__main__": unittest.main()