Let a wiki's owner decide whether API, widget and public-link runs edit it

A run from an agent's API key or widget acts as the agent's owner, so it
could rewrite any wiki the owner can edit. A new per-wiki setting,
wiki_outside_edits (off by default), decides whether such runs, and runs
from the agent's public link, get the wiki's edit actions. While it is off
they are offered only wiki_view, and the tool refuses writes itself after
reading the live setting. The owner changes it through the owner-only
/api/sources/<id>/wiki/settings route; tokens can read it but not change it.
This commit is contained in:
arc53-machine committed 2026-09-29 16:07:57 +01:00
1 parent 33a73bc442
commit 3b44b6851d
13 files changed
+560 -6

No files matched your search

+52 -4
View File
@@ -3,6 +3,7 @@ from typing import Any, Dict, List, Optional
from docsgpt.agents.tools.base import Tool
from docsgpt.agents.tools.path_utils import validate_tool_path
from docsgpt.storage.db.repositories.sources import SourcesRepository
from docsgpt.storage.db.repositories.wiki_pages import (
WikiPageConflict,
WikiPagesRepository,
@@ -21,6 +22,11 @@ MAX_WIKI_PAGE_BYTES = 1_000_000
_WRITE_ACTIONS = frozenset({"create", "str_replace", "insert", "delete", "rename"})
OUTSIDE_EDITS_DENIED = (
"Error: This wiki's owner doesn't let API, widget or public-link users edit it, "
"so it can't be changed from here. You can still read it."
)
class WikiTool(Tool):
"""Wiki
@@ -39,6 +45,9 @@ class WikiTool(Tool):
self.config = config
self.source_id: Optional[str] = config.get("source_id")
self.source_owner_id: Optional[str] = config.get("source_owner_id")
# An API-key, widget or public-link run: it writes only while the
# wiki's owner allows such edits.
self.outside_caller: bool = bool(config.get("outside_caller"))
decoded_token = config.get("decoded_token") or {}
self.updated_by: Optional[str] = (
(decoded_token.get("sub") if decoded_token else None)
@@ -228,8 +237,30 @@ class WikiTool(Tool):
return message
if access is None or not access.can("edit"):
return message
if self.outside_caller and not self._outside_edits_allowed():
return OUTSIDE_EDITS_DENIED
return None
def _outside_edits_allowed(self) -> bool:
"""Read the wiki's live ``wiki_outside_edits`` setting.
Read on every write rather than trusted from the run's setup, so the
owner turning it off stops a conversation that is already going.
Fails closed on a missing row or a failed lookup.
Returns:
bool: Whether an API, widget or public-link run may edit the wiki.
"""
try:
with db_readonly() as conn:
row = SourcesRepository(conn).get_by_id(str(self.source_id))
except Exception:
logger.exception(
"Wiki outside-edits check failed for source %s", self.source_id
)
return False
return outside_edits_allowed(row)
def get_config_requirements(self) -> Dict[str, Any]:
return {}
@@ -500,11 +531,17 @@ class WikiTool(Tool):
return f"Renamed: {validated_old} -> {validated_new}"
def build_wiki_tool_entry() -> Dict[str, Any]:
"""Build the synthetic tools_dict entry for the WikiTool."""
def build_wiki_tool_entry(writes_allowed: bool = True) -> Dict[str, Any]:
"""Build the synthetic tools_dict entry for the WikiTool.
Args:
writes_allowed: False offers the model only ``wiki_view``.
"""
entry = {"name": "wiki"}
entry["actions"] = [
{**action, "active": True} for action in _wiki_actions_metadata()
{**action, "active": True}
for action in _wiki_actions_metadata()
if writes_allowed or action["name"] == "wiki_view"
]
return entry
@@ -513,11 +550,17 @@ def _wiki_actions_metadata() -> List[Dict[str, Any]]:
return WikiTool().get_actions_metadata()
def outside_edits_allowed(source_row: Optional[Dict[str, Any]]) -> bool:
"""Whether a wiki's owner lets API, widget and public-link runs edit it."""
return bool(source_row and source_row.get("wiki_outside_edits"))
def build_wiki_tool_config(
source_id: str,
source_owner_id: str,
decoded_token: Optional[Dict] = None,
user: Optional[str] = None,
outside_caller: bool = False,
) -> Dict[str, Any]:
"""Build the config dict passed to the injected WikiTool."""
return {
@@ -525,6 +568,7 @@ def build_wiki_tool_config(
"source_owner_id": source_owner_id,
"decoded_token": decoded_token,
"user": user,
"outside_caller": bool(outside_caller),
}
@@ -534,15 +578,19 @@ def add_wiki_tool(tools_dict: Dict, config: Dict) -> None:
Mirrors ``add_internal_search_tool``: the entry carries ``id=WIKI_TOOL_ID``
so the executor can resolve the synthetic (DB-rowless) tool, and a ``config``
the executor copies into the loaded tool. Mutates ``tools_dict`` in place.
``writes_allowed=False`` (an API, widget or public-link run on a wiki
whose owner hasn't allowed their edits) offers only ``wiki_view``; the
tool still refuses writes itself from ``outside_caller``.
"""
if not config or not config.get("source_id") or not config.get("source_owner_id"):
return
entry = build_wiki_tool_entry()
entry = build_wiki_tool_entry(writes_allowed=config.get("writes_allowed", True) is not False)
entry["id"] = WIKI_TOOL_ID
entry["config"] = build_wiki_tool_config(
source_id=config["source_id"],
source_owner_id=config["source_owner_id"],
decoded_token=config.get("decoded_token"),
user=config.get("user"),
outside_caller=bool(config.get("outside_caller")),
)
tools_dict[WIKI_TOOL_ID] = entry
@@ -0,0 +1,33 @@
"""0043 wiki outside edits — the wiki owner's say on API, widget and public-link edits.
An agent run from its API key or widget acts as the agent's owner, and a
public-link user is a stranger to them, so neither should rewrite a wiki the
agent can edit unless the wiki's owner allows it. ``wiki_outside_edits``
records that choice on the source; it is off by default, so such runs can
still read the wiki but not change it.
Idempotent both ways.
Revision ID: 0043_wiki_outside_edits
Revises: 0042_schedule_created_via_api
"""
from typing import Sequence, Union
from alembic import op
revision: str = "0043_wiki_outside_edits"
down_revision: Union[str, None] = "0042_schedule_created_via_api"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.execute(
"ALTER TABLE sources ADD COLUMN IF NOT EXISTS wiki_outside_edits BOOLEAN NOT NULL DEFAULT false;"
)
def downgrade() -> None:
op.execute("ALTER TABLE sources DROP COLUMN IF EXISTS wiki_outside_edits;")
@@ -27,6 +27,7 @@ from docsgpt.core.model_utils import (
get_provider_from_model_id,
validate_model_id,
)
from docsgpt.agents.tools.wiki import outside_edits_allowed
from docsgpt.core.settings import settings
from docsgpt.guardrails.config import AgentConfig
from sqlalchemy import text as sql_text
@@ -1259,10 +1260,22 @@ class StreamProcessor:
writable wiki source; the first match wins and the scan stops there so
this runs at most one owner+source lookup per chat on the hot path.
Returns None when no writable wiki source is present.
An API-key, widget or public-link run (``outside_caller``) gets the
edit actions only when the wiki's owner turned on
``wiki_outside_edits``; otherwise ``writes_allowed`` is False and the
tool offers only ``wiki_view``.
"""
caller = self.decoded_token.get("sub") if self.decoded_token else None
if not caller:
return None
# Processors built without __init__ (tests, resume helpers) lack these.
run_config = getattr(self, "agent_config", None) or {}
outside_caller = bool(
run_config.get("external_api_caller")
or run_config.get("public_link_caller")
or getattr(self, "external_caller", False)
)
wiki_config: Optional[Dict[str, Any]] = None
try:
@@ -1286,6 +1299,8 @@ class StreamProcessor:
"source_owner_id": owner,
"decoded_token": self.decoded_token,
"user": caller,
"outside_caller": outside_caller,
"writes_allowed": not outside_caller or outside_edits_allowed(source_doc),
}
break
except Exception:
+3
View File
@@ -224,6 +224,7 @@ RULES: dict[tuple[str, str], Rule] = {
("/api/get_chunks", "GET"): _rule("sources:read", (QUERY, "id")),
("/api/sources/<string:source_id>/wiki/pages", "GET"): _rule("sources:read", (VIEW, "source_id")),
("/api/sources/<string:source_id>/wiki/page", "GET"): _rule("sources:read", (VIEW, "source_id")),
("/api/sources/<string:source_id>/wiki/settings", "GET"): _rule("sources:read", (VIEW, "source_id")),
("/api/sources/<string:source_id>/graph", "GET"): _rule("sources:read", (VIEW, "source_id")),
("/api/sources/<string:source_id>/graph/node/<string:node_id>", "GET"): _rule(
"sources:read", (VIEW, "source_id")
@@ -357,6 +358,8 @@ DENIED: dict[str, tuple[str, ...]] = {
"/api/teams/<string:team_id>/grants": ("POST", "DELETE"),
"/api/teams/<string:team_id>/transfer_owner": ("*",),
"/api/resource_settings": ("PUT",),
# Who may edit a wiki from outside the app is the owner's call in a session.
"/api/sources/<string:source_id>/wiki/settings": ("PUT",),
"/swagger.json": ("*",),
}
DENIED_PREFIXES = (
+81
View File
@@ -1056,6 +1056,87 @@ class WikiPage(Resource):
)
def _wiki_settings_body(doc: dict, ra) -> dict:
"""The wiki settings response: the stored switch plus the caller's access."""
return {
"success": True,
"allow_outside_edits": bool(doc.get("wiki_outside_edits")),
**ra.payload(),
}
@sources_ns.route("/sources/<string:source_id>/wiki/settings")
class WikiSettings(Resource):
@api.doc(
description="A wiki's settings. Anyone who can see the wiki may read "
"them; returns allow_outside_edits plus the caller's access."
)
def get(self, source_id):
decoded_token = request.decoded_token
if not decoded_token:
return make_response(jsonify({"success": False}), 401)
user = decoded_token.get("sub")
try:
with db_readonly() as conn:
try:
doc, ra = load_source(conn, source_id, user, "use")
except AccessDenied as err:
return denied_response(err)
except Exception as err:
current_app.logger.error(
f"Error reading wiki settings for {source_id}: {err}", exc_info=True
)
return make_response(jsonify({"success": False}), 400)
return make_response(jsonify(_wiki_settings_body(doc, ra)), 200)
@api.doc(
description="Change a wiki's settings (owner only, manage_settings). "
"Body: {\"allow_outside_edits\": bool}: whether runs from the agent's "
"API key, widget or public link may edit the wiki."
)
def put(self, source_id):
decoded_token = request.decoded_token
if not decoded_token:
return make_response(jsonify({"success": False}), 401)
user = decoded_token.get("sub")
data = request.get_json(silent=True) or {}
allowed = data.get("allow_outside_edits")
if not isinstance(allowed, bool):
return make_response(
jsonify(
{"success": False, "message": "allow_outside_edits must be true or false"}
),
400,
)
try:
with db_session() as conn:
try:
doc, ra = load_source(conn, source_id, user, "manage_settings")
except AccessDenied as err:
return denied_response(err)
if SourceConfig.parse(doc.get("config")).kind != "wiki":
return make_response(
jsonify({"success": False, "message": "Source is not a wiki"}), 400
)
SourcesRepository(conn).set_wiki_outside_edits(
str(doc["id"]), ra.owner_id, allowed
)
record_event(
conn,
"source.wiki_settings_updated",
actor=user,
source_id=str(doc["id"]),
allow_outside_edits=allowed,
)
doc["wiki_outside_edits"] = allowed
except Exception as err:
current_app.logger.error(
f"Error updating wiki settings for {source_id}: {err}", exc_info=True
)
return make_response(jsonify({"success": False}), 400)
return make_response(jsonify(_wiki_settings_body(doc, ra)), 200)
def _source_is_blank(doc):
"""True when a source has no ingested files to convert into pages."""
structure = doc.get("directory_structure") or {}
+2
View File
@@ -381,6 +381,8 @@ sources_table = Table(
),
# Whose account a shared resource runs with: the owner's, or each member's.
Column("credential_mode", Text, nullable=False, server_default="owner"),
# A wiki's owner lets API, widget and public-link runs edit it (off: read only).
Column("wiki_outside_edits", Boolean, nullable=False, server_default="false"),
)
agents_table = Table(
@@ -399,6 +399,30 @@ class SourcesRepository:
)
self._conn.execute(stmt)
def set_wiki_outside_edits(self, source_id: str, user_id: str, allowed: bool) -> bool:
"""Record whether API, widget and public-link runs may edit this wiki.
Kept out of :meth:`update`'s columns so no route that forwards a
request body can change it; only the owner-checked wiki settings
route calls this.
Args:
source_id: The source's UUID.
user_id: The owner's id; the row is scoped to it.
allowed: The new value.
Returns:
bool: Whether a row was updated.
"""
t = sources_table
result = self._conn.execute(
t.update()
.where(t.c.id == source_id)
.where(t.c.user_id == user_id)
.values(wiki_outside_edits=bool(allowed), updated_at=func.now())
)
return result.rowcount > 0
def get_by_legacy_id(
self, legacy_mongo_id: str, user_id: Optional[str] = None,
) -> Optional[dict]:
+16 -2
View File
@@ -11,7 +11,7 @@ from unittest.mock import MagicMock, patch
import pytest
def _executor_kwargs(monkeypatch, config_extra=None, **run_kwargs):
def _executor_kwargs(monkeypatch, config_extra=None, agent_kwargs=None, **run_kwargs):
from docsgpt.agents import headless_runner as hr
agent = MagicMock(name="agent")
@@ -27,7 +27,12 @@ def _executor_kwargs(monkeypatch, config_extra=None, **run_kwargs):
monkeypatch.setattr(hr, "get_prompt", lambda _pid: "system prompt")
monkeypatch.setattr(hr.RetrieverCreator, "create_retriever", classmethod(lambda cls, *a, **kw: retriever))
monkeypatch.setattr(hr, "ToolExecutor", _executor)
monkeypatch.setattr(hr.AgentCreator, "create_agent", classmethod(lambda cls, *a, **kw: agent))
def _create_agent(cls, *_args, **kwargs):
if agent_kwargs is not None:
agent_kwargs.update(kwargs)
return agent
monkeypatch.setattr(hr.AgentCreator, "create_agent", classmethod(_create_agent))
monkeypatch.setattr(hr.QuotaService, "check", lambda *a, **kw: None)
config = {"user_id": "u1", "id": "agent-1", "default_model_id": "m", **(config_extra or {})}
with patch("docsgpt.core.model_utils.validate_model_id", return_value=True), \
@@ -51,3 +56,12 @@ class TestHeadlessCallerRules:
kwargs = _executor_kwargs(monkeypatch, config, **{flag: True})
assert kwargs[flag] is True
assert kwargs["api_write_allowlist"] == ["tool-1:send"]
@pytest.mark.parametrize("flag", ["external_caller", "public_link_caller"])
def test_outside_caller_run_gets_no_wiki_editor(self, monkeypatch, flag):
# A scheduled or webhook run has no wiki tool at all, so an outside
# caller's schedule can't edit a wiki whatever the wiki allows.
agent_kwargs = {}
_executor_kwargs(monkeypatch, agent_kwargs=agent_kwargs, **{flag: True})
assert agent_kwargs
assert "wiki_config" not in agent_kwargs
+151
View File
@@ -640,3 +640,154 @@ class TestBuildAgentGating:
assert cfg is not None
# v1 binds the first writable wiki source; the extra is skipped.
assert cfg["source_id"] == "wiki-1"
# =====================================================================
# API, widget and public-link callers (the wiki's outside-edits setting)
# =====================================================================
def _outside_tool(monkeypatch, allowed):
from docsgpt.agents.tools.wiki import WikiTool
class _Sources:
def __init__(self, conn):
pass
def get_by_id(self, sid):
return {"id": sid, "wiki_outside_edits": allowed}
monkeypatch.setattr("docsgpt.agents.tools.wiki.SourcesRepository", _Sources)
return WikiTool(
{
"source_id": "src-1",
"source_owner_id": "owner-sub",
"decoded_token": {"sub": "owner-sub"},
"user": "owner-sub",
"outside_caller": True,
}
)
_WRITES = (
("create", {"path": "/b.md", "content": "x"}),
("str_replace", {"path": "/a.md", "old_str": "one", "new_str": "two"}),
("insert", {"path": "/a.md", "insert_line": 1, "insert_text": "x"}),
("delete", {"path": "/a.md"}),
("rename", {"old_path": "/a.md", "new_path": "/c.md"}),
)
@pytest.mark.unit
class TestOutsideCallerWrites:
def test_refused_while_the_setting_is_off(self, patched_wiki, monkeypatch, reembed_mock):
_FakeWikiRepo().upsert("src-1", "/a.md", "one")
tool = _outside_tool(monkeypatch, False)
for action, kwargs in _WRITES:
result = tool.execute_action(action, **kwargs)
assert "API, widget or public-link" in result, action
reembed_mock.assert_not_called()
assert _FakeWikiRepo().get_by_path("src-1", "/a.md")["content"] == "one"
# Reading stays open to them.
assert "one" in tool.execute_action("view", path="/a.md")
def test_allowed_once_the_owner_turns_it_on(self, patched_wiki, monkeypatch):
tool = _outside_tool(monkeypatch, True)
assert tool.execute_action("create", path="/b.md", content="x") == "Page created: /b.md"
def test_a_missing_row_refuses(self, patched_wiki, monkeypatch):
tool = _outside_tool(monkeypatch, True)
class _Gone:
def __init__(self, conn):
pass
def get_by_id(self, sid):
return None
monkeypatch.setattr("docsgpt.agents.tools.wiki.SourcesRepository", _Gone)
assert "API, widget or public-link" in tool.execute_action("create", path="/b.md", content="x")
def test_owner_and_team_runs_skip_the_setting(self, wiki_tool, monkeypatch):
class _Boom:
def __init__(self, conn):
raise AssertionError("an in-app run must not read the setting")
monkeypatch.setattr("docsgpt.agents.tools.wiki.SourcesRepository", _Boom)
assert wiki_tool.execute_action("create", path="/b.md", content="x") == "Page created: /b.md"
@pytest.mark.unit
class TestReadOnlyEntry:
def _entry(self, **extra):
from docsgpt.agents.tools.wiki import WIKI_TOOL_ID, add_wiki_tool
tools_dict = {}
add_wiki_tool(tools_dict, {"source_id": "s1", "source_owner_id": "owner", "user": "owner", **extra})
return tools_dict[WIKI_TOOL_ID]
def test_outside_caller_without_the_setting_is_offered_only_view(self):
entry = self._entry(outside_caller=True, writes_allowed=False)
assert [a["name"] for a in entry["actions"]] == ["wiki_view"]
assert entry["config"]["outside_caller"] is True
def test_writes_offered_when_allowed(self):
entry = self._entry(outside_caller=True, writes_allowed=True)
names = {a["name"] for a in entry["actions"]}
assert {"wiki_view", "wiki_create", "wiki_str_replace", "wiki_delete"} <= names
assert entry["config"]["outside_caller"] is True
def test_in_app_config_keeps_every_action(self):
entry = self._entry()
assert len(entry["actions"]) == 6
assert entry["config"]["outside_caller"] is False
@pytest.mark.unit
class TestBuildConfigOutsideCallers:
def _cfg(self, monkeypatch, agent_config, allowed=False):
from docsgpt.api.answer.services.stream_processor import StreamProcessor
class _SrcRepo:
def __init__(self, conn):
pass
def get_any(self, sid, owner):
return {"id": sid, "config": {"kind": "wiki"}, "wiki_outside_edits": allowed}
monkeypatch.setattr("docsgpt.api.answer.services.stream_processor.SourcesRepository", _SrcRepo)
monkeypatch.setattr("docsgpt.api.answer.services.stream_processor.db_readonly", _noop_conn)
monkeypatch.setattr(
"docsgpt.api.answer.services.stream_processor._wiki_write_owner", lambda conn, sid, uid: "owner-x"
)
proc = StreamProcessor.__new__(StreamProcessor)
proc.all_sources = [{"id": "wiki-src"}]
proc.decoded_token = {"sub": "owner-x"}
if agent_config is not None:
proc.agent_config = agent_config
return proc._build_wiki_config()
@pytest.mark.parametrize("flag", ["external_api_caller", "public_link_caller"])
def test_outside_caller_gets_read_only_while_off(self, monkeypatch, flag):
cfg = self._cfg(monkeypatch, {flag: True})
assert cfg["outside_caller"] is True
assert cfg["writes_allowed"] is False
@pytest.mark.parametrize("flag", ["external_api_caller", "public_link_caller"])
def test_outside_caller_may_write_when_on(self, monkeypatch, flag):
cfg = self._cfg(monkeypatch, {flag: True}, allowed=True)
assert cfg["outside_caller"] is True
assert cfg["writes_allowed"] is True
def test_v1_key_holder_gets_read_only(self, monkeypatch):
from docsgpt.api.answer.services.stream_processor import StreamProcessor
monkeypatch.setattr(StreamProcessor, "external_caller", True, raising=False)
cfg = self._cfg(monkeypatch, {})
assert cfg["writes_allowed"] is False
@pytest.mark.parametrize("agent_config", [None, {}, {"external_api_caller": False}])
def test_owner_and_team_unaffected(self, monkeypatch, agent_config):
cfg = self._cfg(monkeypatch, agent_config)
assert cfg["outside_caller"] is False
assert cfg["writes_allowed"] is True
@@ -116,3 +116,33 @@ class TestWikiConfigAccess:
assert cfg["source_owner_id"] == OWNER and cfg["user"] == "ed"
assert self._cfg(use_conn, "vi", sid) is None
assert self._cfg(use_conn, "eve", sid) is None
class TestWikiOutsideEdits:
"""API, widget and public-link runs edit a wiki only when its owner allows."""
def _tools(self, conn, caller, sid, agent_config):
from docsgpt.agents.tools.wiki import WIKI_TOOL_ID, add_wiki_tool
from docsgpt.api.answer.services.stream_processor import StreamProcessor
proc = StreamProcessor.__new__(StreamProcessor)
proc.all_sources = [{"id": sid}]
proc.decoded_token = {"sub": caller}
proc.agent_config = agent_config
cfg = proc._build_wiki_config()
tools = {}
add_wiki_tool(tools, cfg)
return {a["name"] for a in tools[WIKI_TOOL_ID]["actions"]}
@pytest.mark.parametrize("flag", ["external_api_caller", "public_link_caller"])
def test_outside_caller_reads_until_the_owner_allows_edits(self, use_conn, flag):
sid = str(SourcesRepository(use_conn).create("W", user_id=OWNER, config={"kind": "wiki"})["id"])
assert self._tools(use_conn, OWNER, sid, {flag: True}) == {"wiki_view"}
SourcesRepository(use_conn).set_wiki_outside_edits(sid, OWNER, True)
assert "wiki_create" in self._tools(use_conn, OWNER, sid, {flag: True})
def test_owner_and_team_editor_keep_every_action(self, use_conn):
sid = str(SourcesRepository(use_conn).create("W", user_id=OWNER, config={"kind": "wiki"})["id"])
_share(use_conn, "source", sid, "ed", "editor")
assert "wiki_create" in self._tools(use_conn, OWNER, sid, {})
assert "wiki_create" in self._tools(use_conn, "ed", sid, {})
+2
View File
@@ -98,6 +98,8 @@ class TestClassification:
("/api/devices/pairings", "POST"),
("/api/connectors/auth", "GET"),
("/api/mcp_server/callback", "GET"),
("/api/resource_settings", "PUT"),
("/api/sources/<string:source_id>/wiki/settings", "PUT"),
],
)
def test_sensitive_routes_are_never_token_reachable(self, rule, method):
@@ -606,3 +606,102 @@ class TestWikiPageEdit:
assert response.status_code == 403
mock_reembed.assert_not_called()
def _settings_call(app, pg_conn, sid, user, method="GET", body=None):
from docsgpt.api.user.sources.routes import WikiSettings
with _patch_db(pg_conn), app.test_request_context(
f"/api/sources/{sid}/wiki/settings", method=method, json=body
):
from flask import request
request.decoded_token = {"sub": user} if user else None
resource = WikiSettings()
return resource.get(sid) if method == "GET" else resource.put(sid)
class TestWikiSettings:
def _wiki(self, pg_conn, owner, kind="wiki"):
from docsgpt.storage.db.repositories.sources import SourcesRepository
src = SourcesRepository(pg_conn).create(
"wiki", user_id=owner, type=kind, config={"kind": kind}
)
return str(src["id"])
def _stored(self, pg_conn, sid):
from docsgpt.storage.db.repositories.sources import SourcesRepository
return SourcesRepository(pg_conn).get_by_id(sid)["wiki_outside_edits"]
def test_returns_401_unauthenticated(self, app, pg_conn):
assert _settings_call(app, pg_conn, str(uuid.uuid4()), None).status_code == 401
assert _settings_call(app, pg_conn, str(uuid.uuid4()), None, "PUT", {}).status_code == 401
def test_defaults_to_off(self, app, pg_conn):
sid = self._wiki(pg_conn, "alice-ws")
response = _settings_call(app, pg_conn, sid, "alice-ws")
assert response.status_code == 200
assert response.json["allow_outside_edits"] is False
assert "manage_settings" in response.json["allowed_actions"]
def test_owner_turns_it_on_and_off(self, app, pg_conn):
sid = self._wiki(pg_conn, "alice-ws-on")
response = _settings_call(
app, pg_conn, sid, "alice-ws-on", "PUT", {"allow_outside_edits": True}
)
assert response.status_code == 200
assert response.json["allow_outside_edits"] is True
assert self._stored(pg_conn, sid) is True
assert _settings_call(app, pg_conn, sid, "alice-ws-on").json["allow_outside_edits"] is True
_settings_call(app, pg_conn, sid, "alice-ws-on", "PUT", {"allow_outside_edits": False})
assert self._stored(pg_conn, sid) is False
def test_change_is_audited(self, app, pg_conn):
from sqlalchemy import text
sid = self._wiki(pg_conn, "alice-ws-audit")
_settings_call(app, pg_conn, sid, "alice-ws-audit", "PUT", {"allow_outside_edits": True})
row = pg_conn.execute(
text("SELECT metadata FROM auth_events WHERE event = 'source.wiki_settings_updated' AND actor_id = :a"),
{"a": "alice-ws-audit"},
).fetchone()
assert row is not None
assert row[0]["source_id"] == sid and row[0]["allow_outside_edits"] is True
@pytest.mark.parametrize("level", ["editor", "viewer"])
def test_team_member_reads_but_cannot_change(self, app, pg_conn, level):
owner, member = f"alice-ws-{level}", f"bob-ws-{level}"
sid = self._wiki(pg_conn, owner)
_grant_team_access(pg_conn, owner, member, sid, level)
read = _settings_call(app, pg_conn, sid, member)
assert read.status_code == 200
assert read.json["allow_outside_edits"] is False
assert "manage_settings" not in read.json["allowed_actions"]
response = _settings_call(
app, pg_conn, sid, member, "PUT", {"allow_outside_edits": True}
)
assert response.status_code == 403
assert self._stored(pg_conn, sid) is False
def test_stranger_gets_404(self, app, pg_conn):
sid = self._wiki(pg_conn, "alice-ws-404")
assert _settings_call(app, pg_conn, sid, "eve-ws").status_code == 404
response = _settings_call(app, pg_conn, sid, "eve-ws", "PUT", {"allow_outside_edits": True})
assert response.status_code == 404
assert self._stored(pg_conn, sid) is False
@pytest.mark.parametrize("body", [{}, {"allow_outside_edits": "yes"}, {"allow_outside_edits": 1}])
def test_bad_body_is_400(self, app, pg_conn, body):
sid = self._wiki(pg_conn, "alice-ws-400")
response = _settings_call(app, pg_conn, sid, "alice-ws-400", "PUT", body)
assert response.status_code == 400
assert self._stored(pg_conn, sid) is False
def test_not_a_wiki_is_400(self, app, pg_conn):
sid = self._wiki(pg_conn, "alice-ws-classic", kind="classic")
response = _settings_call(
app, pg_conn, sid, "alice-ws-classic", "PUT", {"allow_outside_edits": True}
)
assert response.status_code == 400
assert self._stored(pg_conn, sid) is False
+52
View File
@@ -0,0 +1,52 @@
"""Migration round-trip test for 0043_wiki_outside_edits."""
from __future__ import annotations
import os
import subprocess
import sys
from pathlib import Path
import pytest
from sqlalchemy import text
pytestmark = pytest.mark.integration
_0042 = "0042_schedule_created_via_api"
def _run_alembic(url: str, *args: str) -> None:
ini = Path(__file__).resolve().parents[3] / "docsgpt" / "alembic.ini"
subprocess.check_call(
[sys.executable, "-m", "alembic", "-c", str(ini), *args],
timeout=120,
env={**os.environ, "POSTGRES_URI": url},
)
def _column_exists(conn) -> bool:
return conn.execute(
text(
"SELECT 1 FROM information_schema.columns WHERE table_schema = 'public' "
"AND table_name = 'sources' AND column_name = 'wiki_outside_edits'"
)
).fetchone() is not None
class TestMigration0043RoundTrip:
def test_head_defaults_to_off(self, pg_engine):
with pg_engine.begin() as conn:
value = conn.execute(
text("INSERT INTO sources (user_id, name) VALUES ('u', 'w') RETURNING wiki_outside_edits")
).scalar()
assert value is False
def test_downgrade_drops_then_upgrade_restores(self, pg_engine):
url = pg_engine.url.render_as_string(hide_password=False)
_run_alembic(url, "downgrade", _0042)
with pg_engine.connect() as conn:
assert not _column_exists(conn)
_run_alembic(url, "upgrade", "head")
with pg_engine.connect() as conn:
assert _column_exists(conn)