+
diff --git a/templates/error.html b/templates/error.html
index 0120bec..e1f1e0a 100644
--- a/templates/error.html
+++ b/templates/error.html
@@ -8,7 +8,7 @@
Something went wrong
-
{{ message | e | replace('\n', '
\n') | safe }}
+
{{ message }}
Back to home
diff --git a/test_api.py b/test_api.py
index 14968ca..df1ccf5 100644
--- a/test_api.py
+++ b/test_api.py
@@ -3,6 +3,7 @@ from unittest.mock import Mock, patch
from simplejson.scanner import JSONDecodeError
+import web_view
from add_links import api
@@ -19,12 +20,25 @@ class ApiGetTests(unittest.TestCase):
session = Mock()
session.get.return_value = response
- with patch("add_links.api._get_active_session", return_value=session):
+ with patch("add_links.api._get_active_session", return_value=(session, "anon")):
with self.assertRaises(api.MediawikiError) as ctx:
api.api_get({"action": "query"})
self.assertEqual(str(ctx.exception), response.text)
+ def test_active_session_uses_logged_in_request_session(self) -> None:
+ oauth_session = Mock()
+
+ with web_view.app.test_request_context("/"):
+ with patch(
+ "add_links.mediawiki_oauth.get_oauth_session",
+ return_value=oauth_session,
+ ):
+ session, auth_mode = api._get_active_session()
+
+ self.assertIs(session, oauth_session)
+ self.assertEqual(auth_mode, "oauth")
+
if __name__ == "__main__":
unittest.main()
diff --git a/test_match.py b/test_match.py
index c50470e..a9bbb7a 100644
--- a/test_match.py
+++ b/test_match.py
@@ -70,6 +70,26 @@ class FindLinkInContentTests(unittest.TestCase):
self.assertEqual(replacement, "electoral fraud|ballot stuffing")
self.assertEqual(replaced_text, "ballot stuffing")
+ def test_italic_text_inside_existing_link_does_not_split_link(self) -> None:
+ content = (
+ 'For instance, "[[capitalism]]" is the name for the '
+ "[[Capitalist mode of production (Marxist theory)|capitalist "
+ "''mode of production'']] in which the means of production"
+ )
+
+ new_content, replacement, replaced_text = find_link_in_content(
+ "capitalist mode of production", content, "Means of production"
+ )
+
+ self.assertEqual(
+ new_content,
+ 'For instance, "[[capitalism]]" is the name for the '
+ "[[Capitalist mode of production (Marxist theory)|capitalist "
+ "''mode of production'']] in which the [[means of production]]",
+ )
+ self.assertEqual(replacement, "means of production")
+ self.assertEqual(replaced_text, "means of production")
+
if __name__ == "__main__":
unittest.main()
diff --git a/test_mediawiki_api.py b/test_mediawiki_api.py
new file mode 100644
index 0000000..b9fdd34
--- /dev/null
+++ b/test_mediawiki_api.py
@@ -0,0 +1,46 @@
+import unittest
+from unittest.mock import patch
+
+from add_links import mediawiki_api
+
+
+class EditPageTests(unittest.TestCase):
+ def test_edit_page_raises_api_error_with_mediawiki_message(self) -> None:
+ with patch(
+ "add_links.mediawiki_api.call",
+ return_value={"error": {"code": "badtoken", "info": "Invalid CSRF token."}},
+ ):
+ with self.assertRaises(mediawiki_api.APIError) as ctx:
+ mediawiki_api.edit_page(
+ pageid=1,
+ section=0,
+ text="text",
+ summary="summary",
+ baserevid="123",
+ token="bad",
+ )
+
+ self.assertEqual(str(ctx.exception), "Invalid CSRF token.")
+
+ def test_edit_page_raises_api_error_for_unexpected_response(self) -> None:
+ response = {"servedby": "mw-api-ext.eqiad.main"}
+
+ with patch("add_links.mediawiki_api.call", return_value=response):
+ with self.assertRaises(mediawiki_api.APIError) as ctx:
+ mediawiki_api.edit_page(
+ pageid=1,
+ section=0,
+ text="text",
+ summary="summary",
+ baserevid="123",
+ token="token",
+ )
+
+ self.assertEqual(
+ str(ctx.exception),
+ "Unexpected MediaWiki API response: {'servedby': 'mw-api-ext.eqiad.main'}",
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/test_web_view.py b/test_web_view.py
new file mode 100644
index 0000000..da460d2
--- /dev/null
+++ b/test_web_view.py
@@ -0,0 +1,225 @@
+import unittest
+from unittest.mock import patch
+
+import web_view
+
+
+def hit(title: str) -> web_view.Hit:
+ return {
+ "ns": 0,
+ "title": title,
+ "pageid": 1,
+ "size": 1,
+ "wordcount": 1,
+ "snippet": "",
+ "timestamp": "",
+ }
+
+
+class ArticlePageSkipTests(unittest.TestCase):
+ def setUp(self) -> None:
+ web_view.app.config["TESTING"] = True
+ self.client = web_view.app.test_client()
+
+ def test_session_skips_are_left_out_of_candidates(self) -> None:
+ hits = [
+ hit("Mass racial violence in the United States"),
+ hit("National Register of Historic Places listings in Brookhaven (town), New York"),
+ hit("Useful candidate"),
+ ]
+
+ with self.client.session_transaction() as session:
+ session["skipped"] = {
+ "African-American neighborhood": [
+ "Mass racial violence in the United States",
+ "National Register of Historic Places listings in Brookhaven (town), New York",
+ ]
+ }
+
+ with (
+ patch("web_view.mediawiki_oauth.get_username", return_value="Test user"),
+ patch("web_view.mediawiki_oauth.get_oauth_session", return_value=object()),
+ patch("web_view.api.get_wiki_info", return_value=None),
+ patch("web_view.search_count", return_value=3),
+ patch("web_view.search_count_with_link", return_value=0),
+ patch("web_view.search_no_link", return_value=(3, hits)),
+ ):
+ response = self.client.get("/link/African-American_neighborhood")
+
+ body = response.get_data(as_text=True)
+ self.assertEqual(response.status_code, 200)
+ self.assertNotIn(
+ 'data-title="Mass racial violence in the United States"', body
+ )
+ self.assertNotIn(
+ 'data-title="National Register of Historic Places listings in Brookhaven (town), New York"',
+ body,
+ )
+ self.assertIn("Useful candidate", body)
+
+ def test_redirect_target_is_left_out_of_candidates(self) -> None:
+ hits = [
+ hit("Documentary film"),
+ hit("Useful candidate"),
+ ]
+
+ with (
+ patch("web_view.mediawiki_oauth.get_username", return_value="Test user"),
+ patch("web_view.mediawiki_oauth.get_oauth_session", return_value=object()),
+ patch("web_view.api.get_wiki_info", return_value="Documentary film"),
+ patch("web_view.search_count", return_value=2),
+ patch("web_view.search_count_with_link", return_value=0),
+ patch("web_view.search_no_link", return_value=(2, hits)),
+ ):
+ response = self.client.get("/link/DVD_documentary")
+
+ body = response.get_data(as_text=True)
+ self.assertEqual(response.status_code, 200)
+ self.assertNotIn('data-title="Documentary film"', body)
+ self.assertIn("Useful candidate", body)
+
+ def test_redirect_target_case_variant_is_left_out_of_candidates(self) -> None:
+ hits = [hit("documentary film")]
+
+ with (
+ patch("web_view.mediawiki_oauth.get_username", return_value="Test user"),
+ patch("web_view.mediawiki_oauth.get_oauth_session", return_value=object()),
+ patch("web_view.api.get_wiki_info", return_value="Documentary film"),
+ patch("web_view.search_count", return_value=1),
+ patch("web_view.search_count_with_link", return_value=0),
+ patch("web_view.search_no_link", return_value=(1, hits)),
+ ):
+ response = self.client.get("/link/DVD_documentary")
+
+ body = response.get_data(as_text=True)
+ self.assertEqual(response.status_code, 200)
+ self.assertIn("No more candidates found for this article.", body)
+
+
+class ValidHitTests(unittest.TestCase):
+ def setUp(self) -> None:
+ web_view.app.config["TESTING"] = True
+ self.client = web_view.app.test_client()
+
+ def test_redirect_target_is_not_valid_hit(self) -> None:
+ with (
+ patch("web_view.mediawiki_oauth.get_username", return_value="Test user"),
+ patch("web_view.mediawiki_oauth.get_oauth_session", return_value=object()),
+ patch("web_view.get_diff") as get_diff,
+ ):
+ response = self.client.get(
+ "/api/1/valid_hit",
+ query_string={
+ "link_to": "DVD documentary",
+ "link_from": "Documentary film",
+ "redirect_to": "Documentary film",
+ },
+ )
+
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.get_json(), {"valid": False})
+ get_diff.assert_not_called()
+
+
+class LoginRequiredTests(unittest.TestCase):
+ def setUp(self) -> None:
+ web_view.app.config["TESTING"] = True
+ self.client = web_view.app.test_client()
+
+ def test_article_page_redirects_to_login_before_api_calls(self) -> None:
+ with (
+ patch("web_view.mediawiki_oauth.get_username", return_value=None),
+ patch("web_view.mediawiki_oauth.get_oauth_session", return_value=None),
+ patch("web_view.api.get_wiki_info") as get_wiki_info,
+ ):
+ response = self.client.get("/link/DVD_documentary")
+
+ self.assertEqual(response.status_code, 302)
+ self.assertIn("/oauth/start", response.location)
+ get_wiki_info.assert_not_called()
+
+ def test_valid_hit_requires_login_before_diff_call(self) -> None:
+ with (
+ patch("web_view.mediawiki_oauth.get_username", return_value=None),
+ patch("web_view.mediawiki_oauth.get_oauth_session", return_value=None),
+ patch("web_view.get_diff") as get_diff,
+ ):
+ response = self.client.get(
+ "/api/1/valid_hit",
+ query_string={"link_to": "Target", "link_from": "Candidate"},
+ )
+
+ self.assertEqual(response.status_code, 401)
+ self.assertEqual(response.get_json(), {"error": "login required"})
+ get_diff.assert_not_called()
+
+ def test_hits_api_requires_login_before_search(self) -> None:
+ with (
+ patch("web_view.mediawiki_oauth.get_username", return_value=None),
+ patch("web_view.mediawiki_oauth.get_oauth_session", return_value=None),
+ patch("web_view.core.do_search") as do_search,
+ ):
+ response = self.client.get(
+ "/api/1/hits", query_string={"title": "Target"}
+ )
+
+ self.assertEqual(response.status_code, 401)
+ self.assertEqual(response.get_json(), {"error": "login required"})
+ do_search.assert_not_called()
+
+
+class ProtectedCandidateTests(unittest.TestCase):
+ def test_edit_protected_hits_are_filtered_out(self) -> None:
+ hits = [hit("Editable"), hit("Protected")]
+ hits[0]["pageid"] = 10
+ hits[1]["pageid"] = 20
+
+ with patch(
+ "web_view.get_page_info_by_id",
+ return_value={
+ 10: {"pageid": 10, "title": "Editable", "protection": []},
+ 20: {
+ "pageid": 20,
+ "title": "Protected",
+ "protection": [
+ {"type": "edit", "level": "sysop", "expiry": "infinity"}
+ ],
+ },
+ },
+ ):
+ filtered = web_view.filter_edit_protected_hits(hits)
+
+ self.assertEqual([item["title"] for item in filtered], ["Editable"])
+
+ def test_search_no_link_filters_edit_protected_results(self) -> None:
+ hits = [hit("Editable"), hit("Protected")]
+ hits[0]["pageid"] = 10
+ hits[1]["pageid"] = 20
+
+ with (
+ patch(
+ "web_view.run_search",
+ return_value={"searchinfo": {"totalhits": 2}, "search": hits},
+ ),
+ patch(
+ "web_view.get_page_info_by_id",
+ return_value={
+ 10: {"pageid": 10, "title": "Editable", "protection": []},
+ 20: {
+ "pageid": 20,
+ "title": "Protected",
+ "protection": [
+ {"type": "edit", "level": "sysop", "expiry": "infinity"}
+ ],
+ },
+ },
+ ),
+ ):
+ totalhits, filtered = web_view.search_no_link("Target")
+
+ self.assertEqual(totalhits, 2)
+ self.assertEqual([item["title"] for item in filtered], ["Editable"])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/web_view.py b/web_view.py
index 6f7c9c5..00866cd 100755
--- a/web_view.py
+++ b/web_view.py
@@ -1,7 +1,6 @@
#!/usr/bin/python3
import html
-import itertools
import json
import re
import sys
@@ -37,6 +36,14 @@ class Hit(typing.TypedDict):
timestamp: str
+class PageInfo(typing.TypedDict, total=False):
+ """Wikipedia page metadata."""
+
+ pageid: int
+ title: str
+ protection: list[dict[str, str]]
+
+
def render_error(message: str) -> str:
"""Render shared error page."""
return flask.render_template("error.html", message=message)
@@ -97,6 +104,28 @@ def article_url(title: str) -> str:
return flask.url_for("article_page", url_title=title.replace(" ", "_"))
+def canonical_title_key(title: str) -> str:
+ """Normalize a Wikipedia article title for same-page comparisons."""
+ title = title.replace("_", " ").strip()
+ return title[:1].casefold() + title[1:]
+
+
+def same_article_title(a: str | None, b: str | None) -> bool:
+ """Compare titles using Wikipedia's first-letter case-insensitive rule."""
+ if not a or not b:
+ return False
+ return canonical_title_key(a) == canonical_title_key(b)
+
+
+def is_self_link_candidate(
+ hit_title: str, from_title: str, redirect_to: str | None = None
+) -> bool:
+ """Return true when a candidate is the source article or its redirect target."""
+ return same_article_title(hit_title, from_title) or same_article_title(
+ hit_title, redirect_to
+ )
+
+
def get_hit_count(q: str) -> int:
"""Search Wikipedia and return hit count."""
return typing.cast(int, run_search(q, limit=0)["searchinfo"]["totalhits"])
@@ -115,13 +144,60 @@ def search_count_with_link(q: str, redirect_to: str | None = None) -> int:
return count
+def chunks(items: list[int], size: int) -> typing.Iterator[list[int]]:
+ """Yield fixed-size chunks from a list."""
+ for offset in range(0, len(items), size):
+ yield items[offset : offset + size]
+
+
+def get_page_info_by_id(pageids: list[int]) -> dict[int, PageInfo]:
+ """Fetch metadata for pages by page ID."""
+ if not pageids:
+ return {}
+
+ pages: dict[int, PageInfo] = {}
+ for pageid_chunk in chunks(pageids, 50):
+ ret = api.api_get(
+ {
+ "prop": "info",
+ "inprop": "protection",
+ "pageids": "|".join(str(pageid) for pageid in pageid_chunk),
+ }
+ )
+ for page in ret["query"]["pages"]:
+ pageid = typing.cast(int, page["pageid"])
+ pages[pageid] = typing.cast(PageInfo, page)
+ return pages
+
+
+def is_edit_protected(page: PageInfo | None) -> bool:
+ """Check if a page has edit protection."""
+ if not page:
+ return False
+ return any(
+ protection.get("type") == "edit" or protection.get("action") == "edit"
+ for protection in page.get("protection", [])
+ )
+
+
+def filter_edit_protected_hits(hits: list[Hit]) -> list[Hit]:
+ """Remove candidates that cannot be edited because of page protection."""
+ page_info = get_page_info_by_id([hit["pageid"] for hit in hits])
+ return [
+ hit for hit in hits if not is_edit_protected(page_info.get(hit["pageid"]))
+ ]
+
+
def search_no_link(q: str, redirect_to: str | None = None) -> tuple[int, list[Hit]]:
"""Search for mentions of article title with no link included."""
exclude = f' -linksto:"{q}"'
if redirect_to:
exclude += f' -linksto:"{redirect_to}"'
query = run_search(article_title_to_search_query(q) + exclude, "max")
- return (query["searchinfo"]["totalhits"], query["search"])
+ return (
+ query["searchinfo"]["totalhits"],
+ filter_edit_protected_hits(query["search"]),
+ )
@app.before_request
@@ -131,6 +207,22 @@ def global_user() -> None:
flask.g.oauth_session = mediawiki_oauth.get_oauth_session()
+def is_logged_in() -> bool:
+ """Return true when the current request has a Wikipedia OAuth session."""
+ return flask.g.oauth_session is not None
+
+
+def login_redirect() -> Response:
+ """Redirect the current browser request through Wikipedia OAuth."""
+ next_url = flask.request.full_path.rstrip("?")
+ return flask.redirect(flask.url_for("start_oauth", next=next_url))
+
+
+def login_required_json() -> tuple[werkzeug.wrappers.response.Response, int]:
+ """Return a JSON login-required response without making MediaWiki API calls."""
+ return flask.jsonify(error="login required"), 401
+
+
@app.route("/")
def index() -> str | Response:
"""Index page."""
@@ -317,6 +409,9 @@ def match_type(q: str, snippet: str) -> str | None:
def _record_skip(from_title: str, hit_title: str) -> None:
"""Record that a candidate was skipped or saved for this article."""
+ hit_title = hit_title.strip()
+ if not hit_title:
+ return
skipped: dict[str, list[str]] = flask.session.get("skipped", {})
article_skipped = skipped.get(from_title, [])
if hit_title not in article_skipped:
@@ -327,6 +422,10 @@ def _record_skip(from_title: str, hit_title: str) -> None:
def handle_post(url_title: str) -> Response:
"""Handle POST request."""
+ if not is_logged_in():
+ next_url = flask.url_for("article_page", url_title=url_title)
+ return flask.redirect(flask.url_for("start_oauth", next=next_url))
+
from_title = url_title.replace("_", " ").strip()
hit_title = flask.request.form["hit"]
try:
@@ -334,6 +433,8 @@ def handle_post(url_title: str) -> Response:
except mediawiki_oauth.LoginNeeded:
next_url = flask.url_for("article_page", url_title=url_title, title=hit_title)
return flask.redirect(flask.url_for("start_oauth", next=next_url))
+ except NoMatch:
+ return flask.make_response(render_error("No valid unlinked mention found."), 400)
except (mediawiki_api.APIError, api.MediawikiError) as e:
return flask.make_response(
render_mediawiki_error(e, prefix="Save failed"), 502
@@ -355,6 +456,9 @@ def article_page(url_title: str) -> str | Response:
if flask.request.method == "POST":
return handle_post(url_title)
+ if not is_logged_in():
+ return login_redirect()
+
from_title = url_title.replace("_", " ").strip()
try:
@@ -371,10 +475,8 @@ def article_page(url_title: str) -> str | Response:
render_mediawiki_error(e), 502
)
- # Filter out candidates already processed this session
- session_skipped: set[str] = set(
- flask.session.get("skipped", {}).get(from_title, [])
- )
+ article_skipped = flask.session.get("skipped", {}).get(from_title, [])
+ session_skipped: set[str] = set(article_skipped)
# If a specific candidate was requested, move it to the front
title_param = flask.request.args.get("title")
@@ -382,14 +484,12 @@ def article_page(url_title: str) -> str | Response:
hits = [h for h in hits if h["title"] == title_param] + \
[h for h in hits if h["title"] != title_param]
- # Record and apply explicit skip-past
- after = flask.request.args.get("after")
- if after:
- _record_skip(from_title, after)
- session_skipped.add(after)
-
- hits = [h for h in hits if h["title"] not in session_skipped
- and h["title"] != from_title and h["title"] != case_flip_first(from_title)]
+ hits = [
+ h
+ for h in hits
+ if h["title"] not in session_skipped
+ and not is_self_link_candidate(h["title"], from_title, redirect_to)
+ ]
if not hits:
return flask.render_template("all_done.html")
@@ -404,19 +504,23 @@ def article_page(url_title: str) -> str | Response:
with_link=with_link,
hits=hits,
url_title=url_title,
+ skipped_titles=article_skipped,
saves_this_session=saves_this_session,
)
def do_save(title: str, hit_title: str) -> str:
"""Update page on Wikipedia."""
- token = mediawiki_oauth.get_token()
-
try:
redirect_to = api.get_wiki_info(title)
except (api.MissingPage, api.MultipleRedirects, api.MediawikiError):
redirect_to = None
+ if is_self_link_candidate(hit_title, title, redirect_to):
+ raise NoMatch
+
+ token = mediawiki_oauth.get_token()
+
found = get_match(title, hit_title, redirect_to)
summary = (
@@ -444,6 +548,9 @@ def save_done() -> str:
@app.route("/api/1/hits")
def api_hits() -> werkzeug.wrappers.response.Response:
"""Return candidates for the given article title."""
+ if not is_logged_in():
+ return login_required_json()
+
title = flask.request.args.get("title")
assert title
ret = core.do_search(title)
@@ -456,10 +563,17 @@ def api_hits() -> werkzeug.wrappers.response.Response:
@app.route("/api/1/valid_hit")
def api_valid_hit() -> werkzeug.wrappers.response.Response:
"""Check if a candidate article has a valid unlinked mention."""
+ if not is_logged_in():
+ return login_required_json()
+
link_to = flask.request.args["link_to"]
link_from = flask.request.args["link_from"]
redirect_to = flask.request.args.get("redirect_to") or None
+ if is_self_link_candidate(link_from, link_to, redirect_to):
+ _record_skip(link_to, link_from)
+ return flask.jsonify(valid=False)
+
try:
found = get_diff(link_to, link_from, redirect_to)
except NoMatch: