From f4c7e01bee60c708de865a39ce0bd540cc5adb11 Mon Sep 17 00:00:00 2001 From: Lukas Gold Date: Thu, 3 Sep 2026 16:06:40 +0200 Subject: [PATCH 1/2] fix(core): prefer a registered class over a generated one - load_entity looked up classes by name in osw.model.entity only - a packaged class registered for the category IRI was invisible - the generated class then took over the oold type registry slot - warn instead of silently replacing a foreign registration - closes #138 --- src/osw/core.py | 68 ++++-- tests/test_load_entity_registered_class.py | 230 +++++++++++++++++++++ 2 files changed, 282 insertions(+), 16 deletions(-) create mode 100644 tests/test_load_entity_registered_class.py diff --git a/src/osw/core.py b/src/osw/core.py index b85a4b7..3420291 100644 --- a/src/osw/core.py +++ b/src/osw/core.py @@ -31,6 +31,7 @@ set_resolver, ) from oold.generator import Generator +from oold.model.v1 import _types as oold_type_registry from oold.utils.codegen import OOLDJsonSchemaParser from opensemantic.v1 import OswBaseModel from pydantic import PydanticDeprecatedSince20 @@ -1234,6 +1235,10 @@ def load_entity( entity = None schemas = [] schemas_fetched = True + # maps a category (page title, e.g. "Category:OSW...") to the class + # to use for it, either an already registered class (e.g. a packaged + # model class) or a freshly compiled one + category_to_cls: Dict[str, Type[model.Entity]] = {} jsondata = page.get_slot_content("jsondata") if param.remove_empty: remove_empty(jsondata) @@ -1255,21 +1260,50 @@ def load_entity( # If a schema_to_use is provided, we do not need to check if the # model exists if not param.model_to_use: - if not hasattr(model, cls_name): - if param.autofetch_schema: - self.fetch_schema( - OSW.FetchSchemaParam( - schema_title=category, - mode="append", - offline_pages=param.offline_pages, + # Prefer a class already registered for this category IRI + # (e.g. a packaged model class) over compiling a new one. + # Compiling one anyway would take over the oold type + # registry entry for this category and hide the packaged + # class' typed fields/helpers. + registered_cls = oold_type_registry.get(category) + if registered_cls is not None: + category_to_cls[category] = registered_cls + else: + if not hasattr(model, cls_name): + if param.autofetch_schema: + self.fetch_schema( + OSW.FetchSchemaParam( + schema_title=category, + mode="append", + offline_pages=param.offline_pages, + ) ) + if not hasattr(model, cls_name): + schemas_fetched = False + print( + f"Error: Model {cls_name} not found. Schema " + f"{category} needs to be fetched first." ) - if not hasattr(model, cls_name): - schemas_fetched = False - print( - f"Error: Model {cls_name} not found. Schema {category} " - f"needs to be fetched first." - ) + else: + generated_cls = getattr(model, cls_name) + # The class we are about to use may have just + # claimed (or may already hold) the registry slot + # for this category. If a different class is + # registered for it, someone's registration was + # silently overwritten - do not raise, but make + # sure this does not pass silently. + conflicting_cls = oold_type_registry.get(category) + if ( + conflicting_cls is not None + and conflicting_cls is not generated_cls + ): + _logger.warning( + f"Class '{generated_cls}' generated for " + f"category '{category}' claims the oold " + f"type registry slot already held by a " + f"different class '{conflicting_cls}'." + ) + category_to_cls[category] = generated_cls if not schemas_fetched: continue @@ -1281,13 +1315,15 @@ def load_entity( _logger.error("Error: no schema defined") elif len(schemas) == 1: - cls: Type[model.Entity] = getattr(model, schemas[0]["title"]) + # category_to_cls is fully populated for every category that + # reached this point (see the loop above) + cls: Type[model.Entity] = category_to_cls[jsondata["type"][0]] entity: model.Entity = cls(**jsondata) else: bases = [] - for schema in schemas: - bases.append(getattr(model, schema["title"])) + for category in jsondata["type"]: + bases.append(category_to_cls[category]) cls = create_model("Test", __base__=tuple(bases)) entity: model.Entity = cls(**jsondata) except Exception as e: diff --git a/tests/test_load_entity_registered_class.py b/tests/test_load_entity_registered_class.py new file mode 100644 index 0000000..6cea20c --- /dev/null +++ b/tests/test_load_entity_registered_class.py @@ -0,0 +1,230 @@ +"""Unit tests for OSW.load_entity() preferring an already registered class. + +Regression guard for #138: load_entity() used to decide whether to compile a +class for a category by checking ``hasattr(model, cls_name)``, keyed by class +name and only looking inside ``osw.model.entity``. This missed classes from +packaged modules (e.g. ``opensemantic.base.v1._model.Database``) that are +already registered for that category IRI in oold's type registry +(``oold.model.v1._types``), causing load_entity() to silently compile and use +a different, incomplete class instead. + +These tests run fully offline: WtSite.get_page() is fed pages through +``offline_pages`` (see ``osw.wtsite.WtSite.GetPageParam``), so no network or +wiki credentials are required. +""" + +import json +import threading +import uuid as uuid_module +from typing import Any, Dict, Union + +from oold.model.v1 import _types as oold_type_registry +from opensemantic.base.v1 import Database + +import osw.model.entity as model +from osw.core import OSW +from osw.utils.wiki import remove_empty +from osw.wtsite import WtPage, WtSite + + +class OfflineWtPage(WtPage): + """A WtPage that pretends to exist without touching a wiki.""" + + def __init__(self, wtSite: Any = None, title: str = None): + self.wtSite = wtSite + self.title = title + self.exists = True + self._original_content = "" + self.changed: bool = False + self._dict = [] + self._slots: Dict[str, Union[str, dict]] = {"main": ""} + self._slots_changed: Dict[str, bool] = {"main": False} + self._content_model: Dict[str, str] = {"main": "wikitext"} + + +class _FakeConnection: + """Just enough of a requests session for WtSite._clear_cookies().""" + + cookies = [] + + +class _FakeMwSite: + connection = _FakeConnection() + + +def make_offline_wtsite() -> WtSite: + """A WtSite that never touches the network (bypasses __init__).""" + ws = WtSite.__new__(WtSite) + ws._page_cache = {} + ws._cache_enabled = False + ws._session_lock = threading.RLock() + ws._site = _FakeMwSite() + return ws + + +def make_page_for_entity(entity) -> OfflineWtPage: + """Build an offline page whose jsondata slot holds the serialized entity.""" + page = OfflineWtPage(title=f"Item:{OSW.get_osw_id(entity.uuid)}") + jsondata = json.loads(entity.json(exclude_none=True)) + remove_empty(jsondata) + page.set_slot_content("jsondata", jsondata) + return page + + +def make_schema_page(category: str, cls_name: str) -> OfflineWtPage: + """Build an offline page holding the jsonschema slot for a category.""" + page = OfflineWtPage(title=category) + page.set_slot_content("jsonschema", {"title": cls_name}) + return page + + +def make_isolated_cls(name: str, base=model.Item): + """Build a model.Item subclass registered under its own private category IRI. + + Overriding schema_extra's title/uuid makes get_cls_iri() derive a fresh + "Category:OSW" IRI for this class alone, so defining it cannot + clobber the registration of any real category (e.g. "Category:Item"). + """ + namespace = { + "Config": type( + "Config", + (base.Config,), + { + "schema_extra": { + **base.Config.schema_extra, + "title": name, + "uuid": str(uuid_module.uuid4()), + } + }, + ), + "__qualname__": name, + } + return type(base)(name, (base,), namespace) + + +def test_load_entity_prefers_registered_class_over_generated_one(): + """A class already registered for the category IRI is used as-is, and no + replacement class is compiled into osw.model.entity for it.""" + assert not hasattr(model, "Database") + + db = Database(name="TestDb", label=[model.Label(text="Test Db")]) + category = db.type[0] + entity_page = make_page_for_entity(db) + schema_page = make_schema_page(category, "Database") + + osw_obj = OSW(site=make_offline_wtsite()) + + result = osw_obj.load_entity( + OSW.LoadEntityParam( + titles=[entity_page.title], + autofetch_schema=True, + offline_pages={ + entity_page.title: entity_page, + category: schema_page, + }, + ) + ) + + entity = result.entities[0] + assert type(entity) is Database + # the packaged class was used directly, nothing was compiled + assert not hasattr(model, "Database") + + +def test_load_entity_falls_back_to_generated_class_when_nothing_registered(): + """A category with nothing registered in oold's type registry still gets + the class already present in osw.model.entity, exactly like before.""" + category = "Category:OSWFakeCategoryNotRegistered00000000000000" + cls_name = "FakeGeneratedClass" + assert oold_type_registry.get(category) is None + + fake_cls = make_isolated_cls(cls_name) + setattr(model, cls_name, fake_cls) + try: + entity_page = OfflineWtPage(title="Item:OSWFakeEntity0000000000000000000000000") + jsondata = { + "type": [category], + "uuid": "00000000-0000-0000-0000-000000000000", + "name": "x", + "label": [{"text": "x"}], + } + remove_empty(jsondata) + entity_page.set_slot_content("jsondata", jsondata) + schema_page = make_schema_page(category, cls_name) + + osw_obj = OSW(site=make_offline_wtsite()) + + result = osw_obj.load_entity( + OSW.LoadEntityParam( + titles=[entity_page.title], + autofetch_schema=True, + offline_pages={ + entity_page.title: entity_page, + category: schema_page, + }, + ) + ) + + entity = result.entities[0] + assert type(entity) is fake_cls + finally: + delattr(model, cls_name) + + +def test_load_entity_warns_on_registry_conflict(monkeypatch, caplog): + """If the class about to be used for a category differs from whatever is + now registered for that IRI, load_entity() logs a warning instead of + silently letting the mismatch pass.""" + category = "Category:OSWConflictTest000000000000000000000000" + cls_name = "ConflictGeneratedClass" + assert oold_type_registry.get(category) is None + + generated_cls = make_isolated_cls(cls_name) + other_cls = make_isolated_cls("OtherRegisteredClass") + + def fake_fetch_schema(self, fetchSchemaParam=None): + # Simulate fetch_schema() compiling a class and importing it into + # osw.model.entity, while a *different* class ends up holding the + # oold registry slot for the same category. + setattr(model, cls_name, generated_cls) + oold_type_registry[category] = other_cls + + monkeypatch.setattr(OSW, "fetch_schema", fake_fetch_schema) + + try: + entity_page = OfflineWtPage( + title="Item:OSWConflictEntity00000000000000000000000000" + ) + jsondata = { + "type": [category], + "uuid": "11111111-1111-1111-1111-111111111111", + "name": "x", + "label": [{"text": "x"}], + } + remove_empty(jsondata) + entity_page.set_slot_content("jsondata", jsondata) + schema_page = make_schema_page(category, cls_name) + + osw_obj = OSW(site=make_offline_wtsite()) + + result = osw_obj.load_entity( + OSW.LoadEntityParam( + titles=[entity_page.title], + autofetch_schema=True, + offline_pages={ + entity_page.title: entity_page, + category: schema_page, + }, + ) + ) + + entity = result.entities[0] + assert type(entity) is generated_cls + assert any( + "claims the oold type registry slot" in record.message + for record in caplog.records + ) + finally: + if hasattr(model, cls_name): + delattr(model, cls_name) + oold_type_registry.pop(category, None) From 07e7e998bdb6265b6ed3734ad0c8869a892355dc Mon Sep 17 00:00:00 2001 From: Lukas Gold Date: Fri, 4 Sep 2026 09:59:04 +0200 Subject: [PATCH 2/2] fix(core): ignore registered subclasses of the canonical model class - controllers and result wrappers inherit the category IRI they extend - oold's registry keeps whichever of them was defined last - UploadFileResult thus replaced WikiFile and broke file up/download - prefer osw.model.entity's class when the registered one specializes it --- src/osw/core.py | 20 ++++++++ tests/test_load_entity_registered_class.py | 53 ++++++++++++++++++++++ 2 files changed, 73 insertions(+) diff --git a/src/osw/core.py b/src/osw/core.py index 3420291..0176a6b 100644 --- a/src/osw/core.py +++ b/src/osw/core.py @@ -1266,6 +1266,26 @@ def load_entity( # registry entry for this category and hide the packaged # class' typed fields/helpers. registered_cls = oold_type_registry.get(category) + # Controllers and result wrappers (e.g. + # WikiFileController, UploadFileResult) inherit their + # category IRI from the model class they extend, and + # the registry keeps whichever class was defined last. + # Such a specialization asks for fields a plain page's + # jsondata does not carry, so prefer the canonical + # class from osw.model.entity over a subclass of it. + canonical_cls = getattr(model, cls_name, None) + if ( + registered_cls is not None + and canonical_cls is not None + and registered_cls is not canonical_cls + and issubclass(registered_cls, canonical_cls) + ): + _logger.debug( + f"Ignoring '{registered_cls}' registered for " + f"category '{category}': it specializes " + f"'{canonical_cls}', which is used instead." + ) + registered_cls = None if registered_cls is not None: category_to_cls[category] = registered_cls else: diff --git a/tests/test_load_entity_registered_class.py b/tests/test_load_entity_registered_class.py index 6cea20c..36eebe9 100644 --- a/tests/test_load_entity_registered_class.py +++ b/tests/test_load_entity_registered_class.py @@ -171,6 +171,59 @@ def test_load_entity_falls_back_to_generated_class_when_nothing_registered(): delattr(model, cls_name) +def test_load_entity_ignores_a_registered_subclass_of_the_model_class(): + """osw's own controllers and result wrappers (WikiFileController, + UploadFileResult, ...) inherit the category IRI of the model class they + extend, and oold's registry keeps whichever class was defined last. Such a + specialization needs fields a plain page does not carry, so load_entity() + must fall back to the canonical class in osw.model.entity. + """ + category = "Category:OSWSubclassTest000000000000000000000000" + cls_name = "SubclassTestBase" + + base_cls = make_isolated_cls(cls_name) + # a controller-like specialization that additionally requires a field the + # page's jsondata does not provide, mirroring UploadFileResult.source + specialized_cls = type(base_cls)( + "SubclassTestController", + (base_cls,), + {"__annotations__": {"source": str}, "__qualname__": "SubclassTestController"}, + ) + setattr(model, cls_name, base_cls) + oold_type_registry[category] = specialized_cls + try: + entity_page = OfflineWtPage( + title="Item:OSWSubclassEntity0000000000000000000000000" + ) + jsondata = { + "type": [category], + "uuid": "22222222-2222-2222-2222-222222222222", + "name": "x", + "label": [{"text": "x"}], + } + remove_empty(jsondata) + entity_page.set_slot_content("jsondata", jsondata) + schema_page = make_schema_page(category, cls_name) + + osw_obj = OSW(site=make_offline_wtsite()) + + result = osw_obj.load_entity( + OSW.LoadEntityParam( + titles=[entity_page.title], + autofetch_schema=True, + offline_pages={ + entity_page.title: entity_page, + category: schema_page, + }, + ) + ) + + assert type(result.entities[0]) is base_cls + finally: + delattr(model, cls_name) + oold_type_registry.pop(category, None) + + def test_load_entity_warns_on_registry_conflict(monkeypatch, caplog): """If the class about to be used for a category differs from whatever is now registered for that IRI, load_entity() logs a warning instead of