Skip to content

fix(core): skip a page whose category has no usable schema #209

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LukasGold wants to merge 1 commit into
base: fix/load-entity-registered-class
Choose a base branch
from
+254 −88
Open

fix(core): skip a page whose category has no usable schema #209

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
188 changes: 100 additions & 88 deletions src/osw/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1254,101 +1254,113 @@ def load_entity(
remove_empty(jsondata)
if jsondata:
for category in jsondata["type"]:
schema = (
self.site
.get_page(
WtSite.GetPageParam(
titles=[category], offline_pages=param.offline_pages
# one unusable category must not end the whole call
try:
schema = (
self.site
.get_page(
WtSite.GetPageParam(
titles=[category],
offline_pages=param.offline_pages,
)
)
.pages[0]
.get_slot_content("jsonschema")
)
.pages[0]
.get_slot_content("jsonschema")
)
schemas.append(schema)
# generate model if not already exists
cls_name: str = schema["title"]
# If a schema_to_use is provided, we do not need to check if
# the model exists
if not param.model_to_use:
# 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)
# 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
# the schema title can also name a module
# attribute that is not a class
and isinstance(canonical_cls, type)
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:
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
_logger.error(
f"Model {cls_name} not found. Schema "
f"{category} needs to be fetched first."
schemas.append(schema)
# generate model if not already exists
cls_name: str = schema["title"]
# If a schema_to_use is provided, we do not need to check if
# the model exists
if not param.model_to_use:
# 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)
# 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
# the schema title can also name a module
# attribute that is not a class
and isinstance(canonical_cls, type)
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:
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
# the slot holder specializes
# generated_cls, e.g. a controller
# that inherits the category IRI:
# the case the guard above expects,
# not a conflict. A fetch can leave
# a non-class under the schema title.
and not (
isinstance(generated_cls, type)
and issubclass(
conflicting_cls, generated_cls
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
_logger.error(
f"Model {cls_name} not found. Schema "
f"{category} needs to be fetched first."
)
):
_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}'."
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
)
category_to_cls[category] = generated_cls
if (
conflicting_cls is not None
and conflicting_cls is not generated_cls
# the slot holder specializes
# generated_cls, e.g. a controller
# that inherits the category IRI:
# the case the guard above expects,
# not a conflict. A fetch can leave
# a non-class under the schema title.
and not (
isinstance(generated_cls, type)
and issubclass(
conflicting_cls, 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
except Exception as e:
schemas_fetched = False
_logger.error(
f"Error reading the schema of category "
f"{category} for page {page.title}: {e}"
)
continue
if not schemas_fetched:
continue
# fetch_schema() reloads osw.model.entity, which replaces every
Expand Down
154 changes: 154 additions & 0 deletions tests/test_load_entity_bad_category.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,154 @@
"""Unit tests for load_entity() when a category page has no usable schema.

Regression guard for #202: https://github.com/OpenSemanticLab/osw-python/issues/202
load_entity() read each category's schema in a per-page loop without a guard
around it. A category page with no jsonschema slot, a schema without a
"title" key, or a title that is not a string all raised out of the loop and
ended the whole call, so no further title in a multi-title load was ever
processed.

These run fully offline: the fake site never touches the network.
"""

import json
from types import SimpleNamespace

import osw.model.entity as model
from osw.core import OSW
from osw.utils.wiki import remove_empty

_CATEGORY = "Category:OSWBadCategoryTest0000000000000000000"


class _FakePage:
"""A page that serves a fixed jsondata or jsonschema slot."""

def __init__(self, title, jsondata=None, schema=None):
self.title = title
self._jsondata = jsondata
self._schema = schema

def get_slot_content(self, slot):
if slot == "jsondata":
return self._jsondata
if slot == "jsonschema":
return self._schema
return None


class _FakeSite:
"""A site that resolves get_page() by title, so a page fetch and a
category (schema) fetch return different content."""

def __init__(self, pages_by_title, cache_enabled=False):
self._pages_by_title = pages_by_title
self.cache_enabled = cache_enabled

def get_cache_enabled(self):
return self.cache_enabled

def enable_cache(self):
self.cache_enabled = True

def disable_cache(self):
self.cache_enabled = False

def get_page(self, param):
pages = [self._pages_by_title[title] for title in param.titles]
return SimpleNamespace(pages=pages)


def _bad_category_site(title, schema):
"""A site with one page whose only category serves the given schema."""
jsondata = {"type": [_CATEGORY], "uuid": "00000000-0000-0000-0000-000000000000"}
entity_page = _FakePage(title, jsondata=jsondata)
category_page = _FakePage(_CATEGORY, schema=schema)
return _FakeSite({title: entity_page, _CATEGORY: category_page})


def test_missing_jsonschema_slot_does_not_raise():
"""get_slot_content("jsonschema") returning None must not raise."""
title = "Item:BadNoSlot"
site = _bad_category_site(title, schema=None)
osw_obj = OSW.construct(site=site)

result = osw_obj.load_entity(title)

assert result is None


def test_schema_without_title_key_does_not_raise():
"""A schema dict with no "title" key must not raise."""
title = "Item:BadNoTitle"
site = _bad_category_site(title, schema={})
osw_obj = OSW.construct(site=site)

result = osw_obj.load_entity(title)

assert result is None


def test_schema_title_not_a_string_does_not_raise():
"""A "title" that is not a string must not raise."""
title = "Item:BadTitleType"
site = _bad_category_site(title, schema={"title": 123})
osw_obj = OSW.construct(site=site)

result = osw_obj.load_entity(title)

assert result is None


def _valid_item_page(title):
"""A page whose category and jsondata build a real model.Item entity."""
item = model.Item(label=[model.Label(text="Test Item")])
category = item.type[0]
jsondata = json.loads(item.json(exclude_none=True))
remove_empty(jsondata)
entity_page = _FakePage(title, jsondata=jsondata)
category_page = _FakePage(category, schema={"title": "Item"})
return entity_page, category_page, item


def test_bad_category_on_first_page_does_not_block_the_second():
"""A multi-title load must not lose every title because one page's
category is unusable; the rest of the loop must still be processed."""
bad_title = "Item:BadFirst"
good_title = "Item:GoodSecond"

bad_jsondata = {
"type": [_CATEGORY],
"uuid": "11111111-1111-1111-1111-111111111111",
}
bad_entity_page = _FakePage(bad_title, jsondata=bad_jsondata)
bad_category_page = _FakePage(_CATEGORY, schema=None)

good_entity_page, good_category_page, item = _valid_item_page(good_title)

site = _FakeSite({
bad_title: bad_entity_page,
good_title: good_entity_page,
_CATEGORY: bad_category_page,
good_category_page.title: good_category_page,
})
osw_obj = OSW.construct(site=site)

result = osw_obj.load_entity([bad_title, good_title])

assert len(result) == 1
assert result[0].uuid == item.uuid


def test_bad_category_logs_the_page_title_and_the_category(caplog):
"""The logged error must name both the failing page and the category, so
the fault can be traced back to the wiki page that needs a fix."""
title = "Item:BadLogged"
site = _bad_category_site(title, schema=None)
osw_obj = OSW.construct(site=site)

osw_obj.load_entity(title)

assert any(
title in record.message and _CATEGORY in record.message
for record in caplog.records
)
Loading