Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 73 additions & 10 deletions docker/scripts/odoo_website_bootstrap.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import json
import os
from pathlib import Path
from typing import Any
from typing import Any, cast
from urllib.parse import urlparse

ODOO_INSTANCE_OVERRIDES_PAYLOAD_ENV_KEY = "ODOO_INSTANCE_OVERRIDES_PAYLOAD_B64"
Expand Down Expand Up @@ -133,6 +133,12 @@ def _assert_field_record_id(record: Any, field_name: str, expected_record: Any,
raise RuntimeError(f"Website bootstrap failed to persist {label}: expected record {expected_id!r}, got {actual_id!r}.")


def _field_record_matches(record: Any, field_name: str, expected_record: Any) -> bool:
if not record or field_name not in record._fields:
return False
return _record_id(_field_value(record, field_name)) == _record_id(expected_record)


def _config_parameter_value(env: Any, key: str) -> str:
parameter_model = env["ir.config_parameter"].sudo()
get_param = getattr(parameter_model, "get_param", None)
Expand Down Expand Up @@ -163,32 +169,48 @@ def _marker_bool(value: bool) -> str:

def _print_bootstrap_readback(
*,
env: Any,
website: Any,
canonical_url: str,
homepage_url: str,
homepage_page: Any | None,
primary_page_xmlid: str,
primary_page_xmlid_found: bool,
logo_expected: bool,
page_website_bound_count: int,
view_website_bound_count: int,
) -> None:
canonical_host = _canonical_host(canonical_url)
website_domain = str(_field_value(website, "domain") or "") if "domain" in website._fields else ""
actual_homepage_url = str(_field_value(website, "homepage_url") or "") if "homepage_url" in website._fields else ""
actual_homepage = _field_value(website, "homepage_id") if "homepage_id" in website._fields else None
actual_homepage_id = _record_id(actual_homepage)
homepage_page_id = _record_id(homepage_page)
homepage_view = _field_value(homepage_page, "view_id") if homepage_page and "view_id" in homepage_page._fields else None
logo_present = bool(_field_value(website, "logo")) if "logo" in website._fields else False
web_base_url_matches = False
if canonical_url:
web_base_url_matches = _config_parameter_value(env, "web.base.url") == canonical_url

print(f"website_bootstrap_website_id={getattr(website, 'id', '')}")
print(f"website_bootstrap_domain_set={_marker_bool(bool(website_domain))}")
print(f"website_bootstrap_domain_matches_canonical={_marker_bool(bool(canonical_host) and website_domain == canonical_host)}")
print(f"website_bootstrap_web_base_url_matches={_marker_bool(not canonical_url or web_base_url_matches)}")
print(f"website_bootstrap_homepage_url_set={_marker_bool(bool(actual_homepage_url))}")
print(f"website_bootstrap_homepage_url_matches={_marker_bool(bool(homepage_url) and actual_homepage_url == homepage_url)}")
print(f"website_bootstrap_homepage_page_found={_marker_bool(bool(homepage_page_id))}")
print(f"website_bootstrap_primary_page_xmlid_found={_marker_bool(bool(primary_page_xmlid) and primary_page_xmlid_found)}")
print(
f"website_bootstrap_homepage_matches_page={_marker_bool(bool(homepage_page_id) and actual_homepage_id == homepage_page_id)}"
)
print(
f"website_bootstrap_homepage_page_website_matches={_marker_bool(_field_record_matches(homepage_page, 'website_id', website))}"
)
print(
f"website_bootstrap_homepage_view_website_matches={_marker_bool(_field_record_matches(homepage_view, 'website_id', website))}"
)
print(f"website_bootstrap_page_website_bound_count={page_website_bound_count}")
print(f"website_bootstrap_view_website_bound_count={view_website_bound_count}")
print(f"website_bootstrap_logo_present={_marker_bool(not logo_expected or logo_present)}")


Expand All @@ -200,7 +222,7 @@ def _select_website(
default_website: Any | None = None,
) -> Any:
if primary_page and "website_id" in primary_page._fields:
page_website = _field_value(primary_page, "website_id")
page_website = cast(Any, _field_value(primary_page, "website_id"))
if page_website:
return page_website.sudo()
canonical_host = _canonical_host(canonical_url)
Expand Down Expand Up @@ -228,6 +250,29 @@ def _clear_duplicate_canonical_domains(website_model: Any, *, website: Any, cano
duplicates.sudo().write({"domain": ""})


def _bind_page_to_website(page: Any, website: Any, *, published: bool) -> tuple[bool, bool]:
if not page:
return False, False
page_values: dict[str, object] = {}
if published:
page_values.update({"is_published": True, "website_published": True})
page_bound = False
if "website_id" in page._fields:
page_values["website_id"] = website.id
page_bound = True
_write_existing_fields(page, page_values)
if page_bound:
_assert_field_record_id(page, "website_id", website, label="page website")

view = cast(Any, _field_value(page, "view_id")) if "view_id" in page._fields else None
view_bound = False
if view and "website_id" in view._fields:
_write_existing_fields(view, {"website_id": website.id})
_assert_field_record_id(view, "website_id", website, label="page view website")
view_bound = True
return page_bound, view_bound


def _homepage_values(website: Any, *, homepage_url: str, homepage_page: Any | None) -> dict[str, object]:
values: dict[str, object] = {}
if homepage_page and "homepage_id" in website._fields:
Expand Down Expand Up @@ -283,10 +328,18 @@ def _find_website_page_by_xmlid(env: Any, *, xmlid: str) -> Any | None:
def _find_website_page_by_url(env: Any, website: Any, *, url: str) -> Any | None:
if not url:
return None
page_model = env["website.page"].sudo()
page_domain: list[Any] = [("url", "=", url)]
if "website_id" in env["website.page"]._fields:
if "website_id" in page_model._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)
scoped_page = page_model.search(page_domain, order="website_id desc,id", limit=1)
if scoped_page:
return scoped_page
# A previous partial bootstrap can leave the requested URL bound to a
# stale website. Reclaim the exact page so the selected website owns the
# public route instead of delegating it back to the stale binding.
return page_model.search([("url", "=", url)], order="id", limit=1)
return page_model.search(page_domain, order="id", limit=1)


def _find_website_page(env: Any, website: Any, *, xmlid: str, url: str) -> tuple[Any | None, bool]:
Expand All @@ -305,8 +358,6 @@ def _verify_route(env: Any, website: Any, route_payload: dict[str, object], *, f
module_name = str(route_payload.get("module") or fallback_module or "").strip()
page = _find_website_page_by_url(env, website, 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):
Expand Down Expand Up @@ -365,6 +416,8 @@ def apply_website_bootstrap(env: Any, parsed_payload: dict[str, object] | None)
_set_config_parameter(env, "web.base.url.freeze", "True")
_clear_duplicate_canonical_domains(website_model, website=website, canonical_url=canonical_url)
website_values["domain"] = _canonical_host(canonical_url)
if "sequence" in website._fields:
website_values["sequence"] = 0
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)
Expand All @@ -386,11 +439,12 @@ def apply_website_bootstrap(env: Any, parsed_payload: dict[str, object] | None)
_assert_binary_field_value(website, "logo", logo_value, label="website logo")

homepage_page = primary_page or _find_website_page_by_url(env, website, url=homepage_url)
page_website_bound_count = 0
view_website_bound_count = 0
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)
page_bound, view_bound = _bind_page_to_website(homepage_page, website, published=True)
page_website_bound_count += int(page_bound)
view_website_bound_count += int(view_bound)
_write_existing_fields(website, _homepage_values(website, homepage_url=homepage_url, homepage_page=homepage_page))
final_homepage_url = homepage_url
final_homepage_page = homepage_page
Expand All @@ -402,11 +456,17 @@ def apply_website_bootstrap(env: Any, parsed_payload: dict[str, object] | None)
route_page = _verify_route(
env, website, {"url": homepage_url, "module": fallback_module, "published": True}, fallback_module=fallback_module
)
page_bound, view_bound = _bind_page_to_website(route_page, website, published=True)
page_website_bound_count += int(page_bound)
view_website_bound_count += int(view_bound)
_write_existing_fields(website, _homepage_values(website, homepage_url=homepage_url, homepage_page=route_page))
final_homepage_page = route_page
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)
page_bound, view_bound = _bind_page_to_website(route_page, website, published=bool(route_payload.get("published", True)))
page_website_bound_count += int(page_bound)
view_website_bound_count += int(view_bound)
if bool(route_payload.get("homepage")):
route_url = str(route_payload.get("url") or "").strip()
_write_existing_fields(website, _homepage_values(website, homepage_url=route_url, homepage_page=route_page))
Expand All @@ -419,12 +479,15 @@ def apply_website_bootstrap(env: Any, parsed_payload: dict[str, object] | None)
_assert_field_record_id(website, "homepage_id", final_homepage_page, label="homepage page")

_print_bootstrap_readback(
env=env,
website=website,
canonical_url=canonical_url,
homepage_url=final_homepage_url,
homepage_page=final_homepage_page,
primary_page_xmlid=primary_page_xmlid,
primary_page_xmlid_found=primary_page_xmlid_found,
logo_expected=logo_expected,
page_website_bound_count=page_website_bound_count,
view_website_bound_count=view_website_bound_count,
)
print("website_bootstrap_applied=true")
5 changes: 4 additions & 1 deletion docs/tooling/workspace-cli.md
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,10 @@ Notes
Launchplane-owned runtime records. Data workflows and startup apply bootstrap
state idempotently after modules are installed, verify required public website
identity fields before reporting success, and avoid hard-coded tenant
defaults.
defaults. Page-backed bootstrap also binds discovered `website.page` records,
their website-specific views when available, and route readback markers to the
selected website so post-deploy proof can distinguish payload rendering from
public website identity persistence.
- 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
Expand Down
77 changes: 69 additions & 8 deletions tests/test_odoo_website_bootstrap.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,10 +76,14 @@ def __init__(
def sudo(self) -> FakeModel:
return self

def search(self, *unused_args: object, **unused_kwargs: object) -> FakeRecord:
self.searches.append((unused_args, unused_kwargs))
def search(self, *args: object, **kwargs: object) -> FakeRecord:
self.searches.append((args, kwargs))
if self.records is not None:
return self.records[0] if self.records else FakeRecord(truthy=False)
if not self.records:
return FakeRecord(truthy=False)
if len(self.records) > 1:
return self.records.pop(0)
return self.records[0]
return self.record

def create(self, values: dict[str, object]) -> FakeRecord:
Expand Down Expand Up @@ -186,10 +190,13 @@ def test_config_parameter_web_base_url_supplies_canonical_when_bootstrap_payload

def test_page_backed_homepage_requires_primary_page_and_persists_it(self) -> None:
env = FakeEnv()
env.website._fields.add("sequence")
env.website.sequence = 10
view = FakeRecord(record_id=142, fields=("website_id",), values={"model_name": "ir.ui.view"})
page = FakeRecord(
record_id=42,
fields=("is_published", "website_published", "website_id"),
values={"model_name": "website.page"},
fields=("is_published", "website_published", "website_id", "view_id"),
values={"model_name": "website.page", "view_id": view},
)
env.refs["cm_website.website_page_cell_mechanic"] = page
payload = {
Expand All @@ -207,11 +214,58 @@ def test_page_backed_homepage_requires_primary_page_and_persists_it(self) -> Non
},
}

website_bootstrap.apply_website_bootstrap(env, payload)
output = io.StringIO()
with redirect_stdout(output):
website_bootstrap.apply_website_bootstrap(env, payload)

self.assertEqual(env.website.homepage_id, page.id)
self.assertEqual(env.website.homepage_url, "/cell-mechanic")
self.assertEqual(env.website.sequence, 0)
self.assertIn({"name": "Cell Mechanic", "domain": "cm-website-testing.example.com", "sequence": 0}, env.website.writes)
self.assertIn({"is_published": True, "website_published": True, "website_id": 1}, page.writes)
self.assertIn({"website_id": 1}, view.writes)
self.assertIn("website_bootstrap_homepage_page_website_matches=true", output.getvalue())
self.assertIn("website_bootstrap_homepage_view_website_matches=true", output.getvalue())
self.assertIn("website_bootstrap_page_website_bound_count=1", output.getvalue())
self.assertIn("website_bootstrap_view_website_bound_count=1", output.getvalue())

def test_url_lookup_reclaims_page_bound_to_stale_website(self) -> None:
env = FakeEnv()
target_website = FakeRecord(
record_id=1,
fields=("name", "domain", "homepage_id", "homepage_url", "logo", "sequence"),
values={"name": "My Website", "domain": "", "sequence": 10},
)
stale_website = FakeRecord(
record_id=2,
fields=("name", "domain", "homepage_id", "homepage_url", "logo", "sequence"),
values={"name": "Old Target", "domain": "old.example.com", "sequence": 10},
)
stale_page = FakeRecord(
record_id=45,
fields=("is_published", "website_published", "website_id"),
values={"model_name": "website.page", "website_id": stale_website},
)
env.website = target_website
env.default_website = target_website
env.pages = FakeModel(records=[FakeRecord(truthy=False), stale_page], fields=("website_id",))
payload = {
"website_bootstrap": {
"name": "Cell Mechanic",
"canonical_url": "https://cm-website-testing.example.com",
"homepage_url": "/cell-mechanic",
}
}

website_bootstrap.apply_website_bootstrap(env, payload)

self.assertEqual(target_website.homepage_id, stale_page.id)
self.assertEqual(stale_page.website_id, target_website.id)
self.assertIn({"is_published": True, "website_published": True, "website_id": target_website.id}, stale_page.writes)
self.assertIn(
(([("url", "=", "/cell-mechanic")],), {"order": "id", "limit": 1}),
env.pages.searches,
)

def test_primary_page_website_wins_over_existing_canonical_domain_match(self) -> None:
env = FakeEnv()
Expand Down Expand Up @@ -316,10 +370,11 @@ def test_bad_primary_page_xmlid_fails_even_when_url_fallback_page_exists(self) -

def test_route_homepage_readback_reports_final_route_homepage(self) -> None:
env = FakeEnv()
route_view = FakeRecord(record_id=144, fields=("website_id",), values={"model_name": "ir.ui.view"})
route_page = FakeRecord(
record_id=44,
fields=("is_published", "website_published", "website_id"),
values={"model_name": "website.page"},
fields=("is_published", "website_published", "website_id", "view_id"),
values={"model_name": "website.page", "view_id": route_view},
)
env.pages = FakeModel(record=route_page, fields=("website_id",))
payload = {
Expand All @@ -343,8 +398,14 @@ def test_route_homepage_readback_reports_final_route_homepage(self) -> None:

self.assertEqual(env.website.homepage_id, route_page.id)
self.assertEqual(env.website.homepage_url, "/shop")
self.assertIn({"is_published": True, "website_published": True, "website_id": 1}, route_page.writes)
self.assertIn({"website_id": 1}, route_view.writes)
self.assertIn("website_bootstrap_homepage_url_matches=true", output.getvalue())
self.assertIn("website_bootstrap_homepage_matches_page=true", output.getvalue())
self.assertIn("website_bootstrap_homepage_page_website_matches=true", output.getvalue())
self.assertIn("website_bootstrap_homepage_view_website_matches=true", output.getvalue())
self.assertIn("website_bootstrap_page_website_bound_count=1", output.getvalue())
self.assertIn("website_bootstrap_view_website_bound_count=1", output.getvalue())

def test_logo_readback_mismatch_fails_before_success_marker(self) -> None:
env = FakeEnv()
Expand Down