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
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ classifiers = [
dependencies = [
# registered_credentials(), used to find the wiki of a file without a URL
"oold>=2.3.0",
"opensemantic",
"opensemantic>=0.2.5",
# the file controllers this module wraps
"opensemantic.core>=0.59.0",
"opensemantic.base>=0.42.7",
Expand Down
53 changes: 48 additions & 5 deletions src/osw/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -2044,6 +2044,51 @@ def __init__(self, **data):
overwrite=self.policy,
)

def _check_property_title_unchanged(
self, entity: model.Entity, entity_title: str
) -> None:
"""Raises if a property page with the uuid of 'entity' exists under another
title, since writing would create a second property page with that uuid.

Parameters
----------
entity
The property entity about to be stored
entity_title
The full target page title, e.g. 'Property:HasName'

Raises
------
ValueError
If the wiki holds the uuid on a differing 'Property:' page
"""
uuid = getattr(entity, "uuid", None)
if not uuid:
return

def normalize(full_title: str) -> str:
# MediaWiki: '_' equals ' ' and the first character is case-insensitive
namespace, _sep, name = full_title.replace("_", " ").partition(":")
return f"{namespace}:{name[:1].upper()}{name[1:]}"

# the base Entity @context maps uuid to the SMW property HasUuid
existing = self.site.semantic_search(f"[[HasUuid::{uuid!s}]]")
existing = [t for t in existing if t.startswith("Property:")]
differing = [t for t in existing if normalize(t) != normalize(entity_title)]
if not differing:
return
existing_str = ", ".join(f"'{t}'" for t in differing)
raise ValueError(
f"Cannot store property with uuid {uuid}: the wiki already holds it "
f"as {existing_str}, but the target title is '{entity_title}'. "
"Writing would create a second property page with the same uuid, "
"which happens when 'name' (or 'meta.wiki_page.title') changed. "
"If a replacement/rename of the property is intended, move the page "
f"'{differing[0]}' to '{entity_title}' in the wiki first, then repeat "
"the update. Otherwise keep the original 'name', or pass the existing "
"title in 'meta.wiki_page.title'."
)

@staticmethod
def _apply_overwrite_policy(param: OSW._ApplyOverwriteParam) -> WtPage:
if param.inplace:
Expand Down Expand Up @@ -2498,6 +2543,8 @@ def store_entity_(
if overwrite_class_param is None:
raise TypeError("'overwrite_class_param' must not be None!")
entity_title = namespace_ + ":" + title_
if namespace_ == "Property" and param.offline is not True:
self._check_property_title_unchanged(entity_, entity_title)
page = self._apply_overwrite_policy(
OSW._ApplyOverwriteParam(
page=WtPage(
Expand Down Expand Up @@ -2772,18 +2819,14 @@ def delete_entity_(entity_, comment_: str = None):
comment_:
Command for the change log, by default None
"""
title_ = None
namespace_ = None
if hasattr(entity_, "meta"):
if entity_.meta and entity_.meta.wiki_page:
if entity_.meta.wiki_page.title:
title_ = entity_.meta.wiki_page.title
if entity_.meta.wiki_page.namespace:
namespace_ = entity_.meta.wiki_page.namespace
if namespace_ is None:
namespace_ = get_namespace(entity_)
if title_ is None:
title_ = OSW.get_osw_id(entity_.uuid)
title_ = get_title(entity_)
if namespace_ is None or title_ is None:
_logger.error("Unsupported entity type")
return
Expand Down
11 changes: 11 additions & 0 deletions src/osw/service/errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,3 +127,14 @@ class BinaryContent(OpError):

type = "BinaryContent"
exit_code = 3


class StoreFailed(OpError):
"""``store_entity`` could not store an entity.

Keeps the wire ``type`` a non-``OpError`` ``StoreEntityPartialError`` had,
while the message carries the reason of each failed page.
"""

type = "StoreEntityPartialError"
exit_code = 1
26 changes: 18 additions & 8 deletions src/osw/service/ops/entities.py
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,10 @@ def create_or_update_entity(
and writes a second entity instead of changing the intended one. Read the
current record with ``get_entity`` first and keep its uuid.

The page name of an entity of a ``Category:Property`` subclass is
``Property:<name>``. An update that changes the ``name`` of a property is
refused: move the page in the wiki first.

``overwrite`` controls update behavior: one of true | false | only empty |
replace remote | keep existing. The default, true, merges field by field
into the stored record: fields absent from ``jsondata`` keep their stored
Expand Down Expand Up @@ -163,15 +167,21 @@ def create_or_update_entity(
raise errors.ValidationError(
f"jsondata does not validate against {category}: {exc}"
)
store = ctx.osw.store_entity(
OSW.StoreEntityParam(
entities=[entity],
namespace=namespace,
overwrite=_parse_overwrite(overwrite),
edit_comment=comment,
bot_edit=True,
try:
store = ctx.osw.store_entity(
OSW.StoreEntityParam(
entities=[entity],
namespace=namespace,
overwrite=_parse_overwrite(overwrite),
edit_comment=comment,
bot_edit=True,
)
)
)
except OSW.StoreEntityPartialError as exc:
# the exception text lists titles only, the reasons are in .failed
raise errors.StoreFailed(
"; ".join(f"{title}: {reason}" for title, reason in exc.failed.items())
) from exc
titles = list(store.pages.keys())
# WtPage.exists is set once, when the page is loaded, and no write path
# refreshes it, so after the store it still reports the pre-write state.
Expand Down
43 changes: 43 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -1010,3 +1010,46 @@ def test_misspelled_command_option_keeps_clicks_suggestion(runner):
combined = _usage_error(result)
assert "--jsondata" in combined
assert "before the command" not in combined


def test_entity_put_reports_the_reason_of_a_failed_store(
runner, configured_env, monkeypatch
):
"""The CLI shows why store_entity refused, not just the failed titles."""
from opensemantic.core.v1 import AnnotationProperty

from osw.core import OSW
from osw.service.ops import entities

fake_osw = MagicMock()
fake_osw.fetch_schema.return_value = MagicMock(error_messages=[])
fake_osw.store_entity.side_effect = OSW.StoreEntityPartialError(
OSW.StoreEntityResult(
change_id="c1",
pages={},
failed={
"Property:NewName": ValueError(
"move the page 'Property:OldName' to 'Property:NewName'"
)
},
)
)
monkeypatch.setattr("osw.service.context.OswExpress", lambda **kwargs: fake_osw)
monkeypatch.setattr(
entities, "_resolve_category_class", lambda category: AnnotationProperty
)

result = runner.invoke(
app,
[
"entity",
"put",
"Category:Property",
"--jsondata",
'{"name": "NewName", "label": [{"text": "x"}]}',
],
)

assert result.exit_code == 1
assert "move the page" in result.stderr
assert "Property:OldName" in result.stderr
179 changes: 179 additions & 0 deletions tests/test_property_title.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,179 @@
"""Unit tests for the page title of Property entities.

A Property entity is stored as 'Property:<name>' (get_title from the opensemantic
package). Writing a property whose name changed would create a second property
page with the same uuid, so store_entity_() refuses it when the wiki holds the
uuid under a different 'Property:' title.

These run fully offline: WtPage.init, the overwrite policy, WtPage.edit and the
write verification are stubbed and the site is a MagicMock.
"""

from unittest.mock import MagicMock
from uuid import uuid4

import pytest
from opensemantic.core.v1 import AnnotationProperty, Label

import osw.model.entity as model
from osw.core import OSW
from osw.service import errors
from osw.service.config import Settings
from osw.service.context import Context, Policy
from osw.service.ops import entities
from osw.wtsite import WtPage


@pytest.fixture
def edits(monkeypatch):
"""The titles of the pages an edit was sent for."""
sent = []
monkeypatch.setattr(WtPage, "init", lambda self: setattr(self, "exists", False))
monkeypatch.setattr(
OSW, "_apply_overwrite_policy", staticmethod(lambda param: param.page)
)
monkeypatch.setattr(WtPage, "edit", lambda self, *a, **kw: sent.append(self.title))
monkeypatch.setattr(OSW, "_get_missing_page_titles", lambda self, titles: [])
return sent


def _osw(hits=()):
site = MagicMock()
site.semantic_search.return_value = list(hits)
return OSW.construct(site=site)


def _prop(name, **kwargs):
return AnnotationProperty(name=name, label=[Label(text=name)], **kwargs)


def _store(osw, entity):
return osw.store_entity(OSW.StoreEntityParam(entities=[entity], parallel=False))


def test_property_without_meta_is_written_to_its_name(edits):
osw = _osw()
_store(osw, _prop("HasName"))
assert edits == ["Property:HasName"]


def test_update_with_same_name_passes(edits):
prop = _prop("HasName")
osw = _osw(hits=["Property:HasName"])
_store(osw, prop)
assert edits == ["Property:HasName"]
osw.site.semantic_search.assert_called_once_with(f"[[HasUuid::{prop.uuid}]]")


def test_no_hit_passes(edits):
osw = _osw(hits=[])
_store(osw, _prop("HasName"))
assert edits == ["Property:HasName"]


def test_renamed_property_is_refused(edits):
prop = _prop("NewName")
osw = _osw(hits=["Property:OldName"])
with pytest.raises(OSW.StoreEntityPartialError) as exc_info:
_store(osw, prop)
(exc,) = exc_info.value.failed.values()
assert isinstance(exc, ValueError)
text = str(exc)
assert "Property:OldName" in text
assert "Property:NewName" in text
assert str(prop.uuid) in text
assert "move" in text
assert edits == []


def test_hit_in_another_namespace_is_ignored(edits):
osw = _osw(hits=["Item:OSW123"])
_store(osw, _prop("HasName"))
assert edits == ["Property:HasName"]


@pytest.mark.parametrize(
"name, hit",
[("hasName", "Property:HasName"), ("Has_Name", "Property:Has Name")],
)
def test_title_comparison_is_normalized_like_mediawiki(edits, name, hit):
osw = _osw(hits=[hit])
_store(osw, _prop(name))
assert len(edits) == 1


def test_explicit_meta_title_wins_over_name(edits):
prop = _prop("NewName")
prop.meta = _meta(title="Pinned")
osw = _osw(hits=["Property:Pinned"])
_store(osw, prop)
assert edits == ["Property:Pinned"]


def test_item_keeps_the_uuid_title_and_runs_no_query(edits):
item = model.Item(label=[model.Label(text="Solo")])
osw = _osw()
_store(osw, item)
assert edits == [f"Item:OSW{str(item.uuid).replace('-', '')}"]
osw.site.semantic_search.assert_not_called()


def _meta(title=None, namespace=None):
return model.Meta(wiki_page=model.WikiPage(title=title, namespace=namespace))


# -- delete_entity -----------------------------------------------------------
def _delete(entity):
site = MagicMock()
page = MagicMock(exists=True)
site.get_page.return_value.pages = [page]
OSW.construct(site=site).delete_entity(
OSW.DeleteEntityParam(entities=[entity], parallel=False)
)
(call,) = site.get_page.call_args_list
page.delete.assert_called_once()
return call.args[0].titles


def test_delete_property_without_meta_targets_its_name():
assert _delete(_prop("HasName")) == ["Property:HasName"]


def test_delete_property_with_meta_title_targets_the_meta_title():
prop = _prop("HasName")
prop.meta = _meta(title="Pinned")
assert _delete(prop) == ["Property:Pinned"]


# -- service op --------------------------------------------------------------
def test_create_or_update_entity_surfaces_the_rename_refusal(monkeypatch, edits):
osw = _osw(hits=["Property:OldName"])
monkeypatch.setattr(
OSW, "fetch_schema", lambda self, param: MagicMock(error_messages=[])
)
monkeypatch.setattr(
entities, "_resolve_category_class", lambda category: AnnotationProperty
)
monkeypatch.setattr(entities.config, "get_active_domain", lambda: "w.example.org")
ctx = Context(
Settings(domain="w.example.org", username="u", password="p"),
Policy(),
osw=osw,
)
with pytest.raises(errors.OpError) as exc_info:
entities.create_or_update_entity(
ctx,
category="Category:OSW0000000000000000000000000000AnnotationProperty",
jsondata={
"uuid": str(uuid4()),
"name": "NewName",
"label": [{"text": "x"}],
},
)
text = str(exc_info.value)
assert "Property:OldName" in text
assert "Property:NewName" in text
assert "move" in text
assert edits == []
# the MCP wire type stays the one a failed store had before
assert exc_info.value.payload()["type"] == "StoreEntityPartialError"
Loading
Loading