Benjamin Mako Hill
af62c4cea4
Wikitravel-style guides get written twice. Someone who does not know
wikitext contributes the information:
Wonderful Hotel, +15555550123
and someone who does comes along later and polishes it:
{{listing|name=Wonderful Hotel|phone=+15555550123}}
Both carry the same information. The second adds two dozen characters of
markup around it. --content-sizes measures that split, emitting two
nullable int64 columns from a single mwparserfromhell tree walk per
revision.
content_chars counts what a contributor who did not know wikitext would
have typed: text nodes, template parameter values, tag contents,
wikilink labels (or targets when unlabeled), external link URLs and
their labels, and heading titles. markup_chars is everything else --
braces, pipes, parameter names, tag names and attributes, link targets,
quote marks, and comments, which never reach a reader.
The two columns sum to the revision length, so their ratio reads
directly as how much of a revision is formatting. That invariant is what
the end-to-end test checks on a real dump, along with nulls for deleted
revisions, which come through the same machinery as the other
parser-based columns.
Emphasis needs no special handling: mwparserfromhell represents ''' and
'' as b and i tags, so the quote marks fall out as markup and the words
they wrap as content, which is the wanted answer.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
886 lines
34 KiB
Python
886 lines
34 KiB
Python
import os
|
|
import subprocess
|
|
import sys
|
|
import tracemalloc
|
|
from io import StringIO
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pyarrow as pa
|
|
import pyarrow.json as pj
|
|
import pytest
|
|
from pandas import DataFrame
|
|
from pandas.testing import assert_frame_equal, assert_series_equal
|
|
|
|
from wikiq import build_table, build_schema, RegexPair, WikitextParser
|
|
from wikiq_test_utils import (
|
|
BASELINE_DIR,
|
|
IKWIKI,
|
|
REGEXTEST,
|
|
SAILORMOON,
|
|
TEST_DIR,
|
|
TEST_OUTPUT_DIR,
|
|
TWINPEAKS,
|
|
WIKIQ,
|
|
WikiqTester,
|
|
requires_mediawiki_utilities,
|
|
requires_pywikidiff2,
|
|
)
|
|
|
|
|
|
def setup():
|
|
tracemalloc.start()
|
|
|
|
if not os.path.exists(TEST_OUTPUT_DIR):
|
|
os.mkdir(TEST_OUTPUT_DIR)
|
|
|
|
|
|
setup()
|
|
|
|
|
|
def read_jsonl_with_schema(filepath: str, **schema_kwargs) -> pd.DataFrame:
|
|
"""Read JSONL file using PyArrow with explicit schema from wikiq."""
|
|
table, _, _ = build_table(**schema_kwargs)
|
|
schema = build_schema(table, **schema_kwargs)
|
|
pa_table = pj.read_json(
|
|
filepath,
|
|
parse_options=pj.ParseOptions(explicit_schema=schema),
|
|
)
|
|
return pa_table.to_pandas()
|
|
|
|
|
|
# with / without pwr DONE
|
|
# with / without url encode DONE
|
|
# with / without collapse user DONE
|
|
# with output to stdout DONE
|
|
# note that the persistence radius is 7 by default
|
|
# reading various file formats including
|
|
# 7z, gz, bz2, xml DONE
|
|
# wikia and wikipedia data DONE
|
|
# malformed xmls DONE
|
|
|
|
def test_WP_noargs():
|
|
tester = WikiqTester(IKWIKI, "noargs")
|
|
|
|
try:
|
|
tester.call_wikiq()
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
def test_WP_namespaces():
|
|
tester = WikiqTester(IKWIKI, "namespaces")
|
|
|
|
try:
|
|
tester.call_wikiq("-n 0", "-n 1")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
# as a test let's make sure that we get equal data frames
|
|
test = pd.read_table(tester.output)
|
|
num_wrong_ns = sum(~test.namespace.isin({0, 1}))
|
|
assert num_wrong_ns == 0
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
def test_WP_revert_radius():
|
|
tester = WikiqTester(IKWIKI, "revert_radius")
|
|
|
|
try:
|
|
tester.call_wikiq("-n 0", "-n 1", "-rr 1")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
# as a test let's make sure that we get equal data frames
|
|
test = pd.read_table(tester.output)
|
|
num_wrong_ns = sum(~test.namespace.isin({0, 1}))
|
|
assert num_wrong_ns == 0
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
def test_WP_no_revert_radius():
|
|
tester = WikiqTester(IKWIKI, "no_revert_radius")
|
|
|
|
try:
|
|
tester.call_wikiq("-rr 0")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
# as a test let's make sure that we get equal data frames
|
|
test = pd.read_table(tester.output)
|
|
num_reverted = sum(i is None for i in test.revert)
|
|
assert num_reverted == 0
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
def test_WP_collapse_user():
|
|
tester = WikiqTester(IKWIKI, "collapse_user")
|
|
|
|
try:
|
|
tester.call_wikiq("--collapse-user")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
def test_noargs():
|
|
tester = WikiqTester(SAILORMOON, "noargs", in_compression="7z")
|
|
|
|
try:
|
|
tester.call_wikiq()
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
|
|
def test_jsonl_noargs():
|
|
"""Test JSONL output format with baseline comparison."""
|
|
tester = WikiqTester(SAILORMOON, "noargs", in_compression="7z", out_format="jsonl", baseline_format="jsonl")
|
|
|
|
try:
|
|
tester.call_wikiq()
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = read_jsonl_with_schema(tester.output)
|
|
baseline = read_jsonl_with_schema(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
|
|
def test_jsonl_tsv_equivalence():
|
|
"""Test that JSONL and TSV outputs contain equivalent data."""
|
|
tester_tsv = WikiqTester(SAILORMOON, "equiv_tsv", in_compression="7z", out_format="tsv")
|
|
tester_jsonl = WikiqTester(SAILORMOON, "equiv_jsonl", in_compression="7z", out_format="jsonl")
|
|
|
|
try:
|
|
tester_tsv.call_wikiq()
|
|
tester_jsonl.call_wikiq()
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
tsv_df = pd.read_table(tester_tsv.output)
|
|
jsonl_df = read_jsonl_with_schema(tester_jsonl.output)
|
|
|
|
# Row counts must match
|
|
assert len(tsv_df) == len(jsonl_df), f"Row count mismatch: TSV={len(tsv_df)}, JSONL={len(jsonl_df)}"
|
|
|
|
# Column sets must match
|
|
assert set(tsv_df.columns) == set(jsonl_df.columns), \
|
|
f"Column mismatch: TSV={set(tsv_df.columns)}, JSONL={set(jsonl_df.columns)}"
|
|
|
|
# Sort both by revid for comparison
|
|
tsv_df = tsv_df.sort_values("revid").reset_index(drop=True)
|
|
jsonl_df = jsonl_df.sort_values("revid").reset_index(drop=True)
|
|
|
|
# Normalize null values: TSV uses nan, schema-based JSONL uses None
|
|
jsonl_df = jsonl_df.replace({None: np.nan})
|
|
|
|
# Compare columns - schema-based reading handles types correctly
|
|
for col in tsv_df.columns:
|
|
if col == "date_time":
|
|
# TSV reads as string, JSONL with schema reads as datetime
|
|
tsv_dates = pd.to_datetime(tsv_df[col]).dt.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
jsonl_dates = jsonl_df[col].dt.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
assert_series_equal(tsv_dates, jsonl_dates, check_names=False)
|
|
else:
|
|
# Allow dtype differences (TSV infers int64, schema uses int32)
|
|
assert_series_equal(tsv_df[col], jsonl_df[col], check_names=False, check_dtype=False)
|
|
|
|
|
|
def test_collapse_user():
|
|
tester = WikiqTester(SAILORMOON, "collapse-user", in_compression="7z")
|
|
|
|
try:
|
|
tester.call_wikiq("--collapse-user", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
@requires_pywikidiff2
|
|
def test_pwr_wikidiff2():
|
|
tester = WikiqTester(SAILORMOON, "persistence_wikidiff2", in_compression="7z")
|
|
|
|
try:
|
|
tester.call_wikiq("--persistence wikidiff2", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
def test_pwr_segment():
|
|
tester = WikiqTester(SAILORMOON, "persistence_segment", in_compression="7z")
|
|
|
|
try:
|
|
tester.call_wikiq("--persistence segment", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
@requires_mediawiki_utilities
|
|
def test_pwr_legacy():
|
|
tester = WikiqTester(SAILORMOON, "persistence_legacy", in_compression="7z")
|
|
|
|
try:
|
|
tester.call_wikiq("--persistence legacy", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
def test_pwr():
|
|
tester = WikiqTester(SAILORMOON, "persistence", in_compression="7z")
|
|
|
|
try:
|
|
tester.call_wikiq("--persistence", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
|
|
test = test.reindex(columns=sorted(test.columns))
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
@requires_pywikidiff2
|
|
def test_diff():
|
|
tester = WikiqTester(SAILORMOON, "diff", in_compression="7z", out_format='jsonl')
|
|
|
|
try:
|
|
tester.call_wikiq("--diff", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
assert "diff" in test.columns, "diff column should exist"
|
|
assert "diff_timeout" in test.columns, "diff_timeout column should exist"
|
|
assert len(test) > 0, "Should have output rows"
|
|
|
|
@requires_pywikidiff2
|
|
def test_diff_plus_pwr():
|
|
tester = WikiqTester(SAILORMOON, "diff_pwr", in_compression="7z", out_format='jsonl')
|
|
|
|
try:
|
|
tester.call_wikiq("--diff --persistence wikidiff2", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
assert "diff" in test.columns, "diff column should exist"
|
|
assert "token_revs" in test.columns, "token_revs column should exist"
|
|
assert len(test) > 0, "Should have output rows"
|
|
|
|
@requires_pywikidiff2
|
|
def test_text():
|
|
tester = WikiqTester(SAILORMOON, "text", in_compression="7z", out_format='jsonl')
|
|
|
|
try:
|
|
tester.call_wikiq("--diff", "--text","--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
assert "text" in test.columns, "text column should exist"
|
|
assert "diff" in test.columns, "diff column should exist"
|
|
assert len(test) > 0, "Should have output rows"
|
|
|
|
def test_malformed_noargs():
|
|
tester = WikiqTester(wiki=TWINPEAKS, case_name="noargs", in_compression="7z")
|
|
want_exception = (
|
|
"xml.etree.ElementTree.ParseError: no element found: line 1369, column 0"
|
|
)
|
|
|
|
try:
|
|
tester.call_wikiq()
|
|
except subprocess.CalledProcessError as exc:
|
|
errlines = exc.stderr.decode("utf8").splitlines()
|
|
assert errlines[-1] == want_exception
|
|
else:
|
|
pytest.fail("No exception raised, want: {}".format(want_exception))
|
|
|
|
def test_stdout_noargs():
|
|
tester = WikiqTester(wiki=SAILORMOON, case_name="noargs", in_compression="7z")
|
|
|
|
try:
|
|
outs = tester.call_wikiq("--stdout", "--fandom-2020", out=False).decode(
|
|
"utf8"
|
|
)
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(StringIO(outs))
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
def test_bad_regex():
|
|
tester = WikiqTester(wiki=REGEXTEST, case_name="bad_regex")
|
|
|
|
# sample arguments for checking that bad arguments get terminated / test_regex_arguments
|
|
bad_arguments_list = [
|
|
# label is missing
|
|
"-RP '\\b\\d+\\b'",
|
|
# number of reg and number of labels do not match
|
|
"-RP 'NPO V' -RP THE -RPl testlabel",
|
|
# cp but rp label
|
|
"-CP '(Tamil|Li)' -RPl testlabel",
|
|
# regex is missing
|
|
"-CPl testlabel",
|
|
"-RP '\\b\\w{3}\\b' -RPl threeletters -CP '\\b\\w{3}\\b'",
|
|
]
|
|
|
|
for arguments in bad_arguments_list:
|
|
try:
|
|
tester.call_wikiq("--stdout", arguments, out=False)
|
|
except subprocess.CalledProcessError as exc:
|
|
# we want to check that the bad arguments were caught and sys.exit is stopping the code
|
|
print(exc.stderr.decode("utf-8"))
|
|
else:
|
|
pytest.fail("No exception raised, want Exception")
|
|
|
|
def test_good_regex():
|
|
# sample arguments for checking the outcomes of good arguments / test_basic_regex
|
|
good_arguments_list = [
|
|
"-RP '\\b\\d{3}\\b' -RPl threedigits",
|
|
"-RP 'TestCase' -RP 'page' -RPl testcases -RPl page_word",
|
|
"-CP 'Chevalier' -CPl chev_com -RP 'welcome to Wikipedia' -RPl wiki_welcome -CP 'Warning' -CPl warning",
|
|
"-CP 'WP:EVADE' -CPl wp_evade",
|
|
]
|
|
|
|
for i, arguments in enumerate(good_arguments_list):
|
|
tester = WikiqTester(wiki=REGEXTEST, case_name="basic", suffix=str(i))
|
|
|
|
try:
|
|
tester.call_wikiq(arguments)
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
print(i)
|
|
|
|
def test_capturegroup_regex():
|
|
cap_arguments_list = [
|
|
"-RP 'Li Chevalier' -RPl li_cheval -CP '(?P<letter>\\b[a-zA-Z]{3}\\b)|(?P<number>\\b\\d+\\b)|(?P<cat>\\bcat\\b)' -CPl three",
|
|
"-CP '(?P<a>\\bTestCaseA\\b)|(?P<b>\\bTestCaseB\\b)|(?P<c>\\bTestCaseC\\b)|(?P<d>\\bTestCaseD\\b)' -CPl testcase -RP '(?P<npov>npov|NPOV)|(?P<neutral>neutral point of view)' -RPl npov",
|
|
]
|
|
|
|
for i, arguments in enumerate(cap_arguments_list):
|
|
tester = WikiqTester(
|
|
wiki=REGEXTEST, case_name="capturegroup", suffix=str(i)
|
|
)
|
|
|
|
try:
|
|
tester.call_wikiq(arguments)
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
|
|
baseline = pd.read_table(tester.baseline_file)
|
|
assert_frame_equal(test, baseline, check_like=True)
|
|
|
|
def test_regex_none_content():
|
|
# deleted or suppressed revisions yield None for text and comments, and
|
|
# matchmake must tolerate that in both the capture-group and plain paths
|
|
pair = RegexPair(r"(?P<letter>\b[a-zA-Z]{3}\b)|(?P<number>\b\d+\b)", "cap")
|
|
assert pair.matchmake(None) == {"cap_letter": None, "cap_number": None}
|
|
|
|
pair = RegexPair(r"\b\d{3}\b", "digits")
|
|
assert pair.matchmake(None) == {"digits": None}
|
|
|
|
def test_regex_deleted_revisions():
|
|
# the ikwiki dump contains revisions with deleted text and deleted
|
|
# comments; regex matching must handle them rather than crashing
|
|
tester = WikiqTester(wiki=IKWIKI, case_name="regex_deleted")
|
|
|
|
try:
|
|
tester.call_wikiq(
|
|
"-RP '(?P<npov>npov|NPOV)' -RPl npov",
|
|
"-CP '(?P<talk>[Tt]alk)' -CPl talk",
|
|
)
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
|
|
deleted = test[test["deleted"]]
|
|
assert len(deleted) > 0
|
|
assert deleted["npov_npov"].isna().all()
|
|
assert deleted["talk_talk"].isna().all()
|
|
|
|
def test_redirect_detection():
|
|
from wikiq.tables import RedirectDetector
|
|
|
|
detector = RedirectDetector()
|
|
|
|
# a plain redirect directive
|
|
assert detector.detect("#REDIRECT [[Target]]") == (True, "Target")
|
|
|
|
# localized aliases are only recognized when configured
|
|
assert detector.detect("#OMDIRIGERING [[Mål]]") == (False, None)
|
|
swedish = RedirectDetector(["OMDIRIGERING"])
|
|
assert swedish.detect("#OMDIRIGERING [[Mål]]") == (True, "Mål")
|
|
assert swedish.detect("#REDIRECT [[Target]]") == (True, "Target")
|
|
|
|
# fragment and label are stripped from the target
|
|
assert detector.detect("#REDIRECT [[Target#Section|label]]") == (True, "Target")
|
|
|
|
# leading whitespace and lowercase are accepted
|
|
assert detector.detect(" \n#redirect [[Target]]") == (True, "Target")
|
|
|
|
# each revision is classified by its own text
|
|
history = ["#REDIRECT [[A]]", "An article now.", "#REDIRECT [[B]]"]
|
|
assert [detector.detect(t) for t in history] == [
|
|
(True, "A"), (False, None), (True, "B"),
|
|
]
|
|
|
|
# a directive that is not at the start of the text is not a redirect
|
|
assert detector.detect("Some text. #REDIRECT [[Target]]") == (False, None)
|
|
|
|
def test_redirect_columns_e2e():
|
|
# revision-level redirect columns on a real dump: the ikwiki dump
|
|
# contains both redirect revisions and revisions with deleted text
|
|
tester = WikiqTester(IKWIKI, "redirect_columns")
|
|
|
|
try:
|
|
tester.call_wikiq()
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
|
|
assert "revision_is_redirect" in test.columns
|
|
assert "revision_redirect_target" in test.columns
|
|
assert "redirect_target" not in test.columns
|
|
|
|
# revisions with deleted text have null values in both columns
|
|
deleted = test[test["deleted"]]
|
|
assert len(deleted) > 0
|
|
assert deleted["revision_is_redirect"].isna().all()
|
|
assert deleted["revision_redirect_target"].isna().all()
|
|
|
|
# the dump contains real redirect revisions, and every detected
|
|
# redirect has a target
|
|
redirects = test[test["revision_is_redirect"] == True]
|
|
assert len(redirects) > 0
|
|
assert redirects["revision_redirect_target"].notna().all()
|
|
|
|
def test_content_sizes():
|
|
# hand-computed fixtures: (text, content_chars, markup_chars)
|
|
cases = [
|
|
# plain text is all content
|
|
("Hello world.", 12, 0),
|
|
# the same information written by someone who knows wikitext: the
|
|
# values survive as content, the scaffolding around them is markup
|
|
("Wonderful Hotel, +15555550123", 29, 0),
|
|
("{{listing|name=Wonderful Hotel|phone=+15555550123}}", 27, 24),
|
|
# emphasis is markup applied to content, not content itself
|
|
("''whatever''", 8, 4),
|
|
("Some '''bold''' prose.", 16, 6),
|
|
# tag contents are content; the tag name and attributes are not
|
|
('Before <span class="x">inner text</span> after', 23, 23),
|
|
# a wikilink renders its label, so the target is markup
|
|
("[[Main Page|the main page]]", 13, 14),
|
|
# an unlabeled wikilink renders its target, which is then content
|
|
("[[Help]]", 4, 4),
|
|
# both halves of an external link are information someone typed
|
|
("[http://x.com label]", 17, 3),
|
|
# comments never reach a reader, so they are entirely markup
|
|
("Text <!-- hidden --> more", 10, 15),
|
|
# nesting: values at any depth are content
|
|
("{{outer|a={{inner|b=deep}} shallow}}", 12, 24),
|
|
]
|
|
|
|
parser = WikitextParser()
|
|
for text, content, markup in cases:
|
|
result = parser.content_sizes(text)
|
|
assert result == (content, markup), f"{text!r}: got {result}, want {(content, markup)}"
|
|
# every character lands in exactly one column
|
|
assert result[0] + result[1] == len(text), f"{text!r}: columns do not sum to length"
|
|
|
|
assert parser.content_sizes(None) is None
|
|
|
|
def test_content_sizes_e2e():
|
|
# the ikwiki dump contains revisions with deleted text, which must get
|
|
# null content sizes; for every other revision the two columns must
|
|
# account for the whole revision
|
|
tester = WikiqTester(IKWIKI, "content_sizes")
|
|
|
|
try:
|
|
tester.call_wikiq("--content-sizes", "--text")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_table(tester.output)
|
|
|
|
assert "content_chars" in test.columns
|
|
assert "markup_chars" in test.columns
|
|
|
|
deleted = test[test["deleted"]]
|
|
assert len(deleted) > 0
|
|
assert deleted["content_chars"].isna().all()
|
|
assert deleted["markup_chars"].isna().all()
|
|
|
|
present = test[~test["deleted"] & test["text"].notna() & test["content_chars"].notna()]
|
|
assert len(present) > 0
|
|
total = present["content_chars"] + present["markup_chars"]
|
|
assert (total == present["text"].str.len()).all()
|
|
# a real wiki has both kinds of revision: some almost pure text, some
|
|
# carrying a lot of markup
|
|
assert (present["markup_chars"] == 0).any()
|
|
assert (present["markup_chars"] > 0).any()
|
|
|
|
def test_external_links_only():
|
|
"""Test that --external-links extracts external links correctly."""
|
|
import mwparserfromhell
|
|
|
|
tester = WikiqTester(SAILORMOON, "external_links_only", in_compression="7z", out_format="jsonl")
|
|
|
|
try:
|
|
# Also include --text so we can verify extraction against actual wikitext
|
|
tester.call_wikiq("--external-links", "--text", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
|
|
# Verify external_links column exists
|
|
assert "external_links" in test.columns, "external_links column should exist"
|
|
|
|
# Verify citations column does NOT exist
|
|
assert "citations" not in test.columns, "citations column should NOT exist when only --external-links is used"
|
|
|
|
# Verify column has list/array type (pandas reads pyarrow lists as numpy arrays)
|
|
assert test["external_links"].apply(lambda x: x is None or hasattr(x, '__len__')).all(), \
|
|
"external_links should be a list/array type or None"
|
|
|
|
# Verify that extracted URLs look like valid URIs (have a scheme or are protocol-relative)
|
|
all_urls = []
|
|
for links in test["external_links"]:
|
|
if links is not None and len(links) > 0:
|
|
all_urls.extend(links)
|
|
|
|
for url in all_urls:
|
|
# External links can be http, https, mailto, ftp, etc. or protocol-relative (//)
|
|
has_scheme = ":" in url and url.index(":") < 10 # scheme:... with short scheme
|
|
is_protocol_relative = url.startswith("//")
|
|
assert has_scheme or is_protocol_relative, \
|
|
f"External link should be a valid URI, got: {url}"
|
|
|
|
# Verify extraction matches mwparserfromhell for a sample of rows with text
|
|
rows_with_links = test[test["external_links"].apply(lambda x: x is not None and len(x) > 0)]
|
|
if len(rows_with_links) > 0:
|
|
# Test up to 5 rows
|
|
sample = rows_with_links.head(5)
|
|
for idx, row in sample.iterrows():
|
|
text = row["text"]
|
|
if text:
|
|
wikicode = mwparserfromhell.parse(text)
|
|
expected_links = [str(link.url) for link in wikicode.filter_external_links()]
|
|
actual_links = list(row["external_links"])
|
|
assert actual_links == expected_links, \
|
|
f"Row {idx}: external_links mismatch. Expected {expected_links}, got {actual_links}"
|
|
|
|
print(f"External links only test passed! {len(test)} rows, {len(all_urls)} total URLs extracted")
|
|
|
|
|
|
def test_citations_only():
|
|
"""Test that --citations extracts citations correctly."""
|
|
import mwparserfromhell
|
|
from wikiq.wikitext_parser import WikitextParser
|
|
|
|
tester = WikiqTester(SAILORMOON, "citations_only", in_compression="7z", out_format="jsonl")
|
|
|
|
try:
|
|
# Also include --text so we can verify extraction against actual wikitext
|
|
tester.call_wikiq("--citations", "--text", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
|
|
# Verify citations column exists
|
|
assert "citations" in test.columns, "citations column should exist"
|
|
|
|
# Verify external_links column does NOT exist
|
|
assert "external_links" not in test.columns, "external_links column should NOT exist when only --citations is used"
|
|
|
|
# Verify column has list/array type (pandas reads pyarrow lists as numpy arrays)
|
|
assert test["citations"].apply(lambda x: x is None or hasattr(x, '__len__')).all(), \
|
|
"citations should be a list/array type or None"
|
|
|
|
# Verify that extracted citations have correct prefixes (ref: or template:)
|
|
all_citations = []
|
|
for citations in test["citations"]:
|
|
if citations is not None and len(citations) > 0:
|
|
all_citations.extend(citations)
|
|
|
|
for citation in all_citations:
|
|
assert citation.startswith("ref:") or citation.startswith("template:"), \
|
|
f"Citation should start with 'ref:' or 'template:', got: {citation}"
|
|
|
|
# Verify extraction matches WikitextParser for a sample of rows with text
|
|
rows_with_citations = test[test["citations"].apply(lambda x: x is not None and len(x) > 0)]
|
|
if len(rows_with_citations) > 0:
|
|
parser = WikitextParser()
|
|
# Test up to 5 rows
|
|
sample = rows_with_citations.head(5)
|
|
for idx, row in sample.iterrows():
|
|
text = row["text"]
|
|
if text:
|
|
expected_citations = parser.extract_citations(text)
|
|
actual_citations = list(row["citations"])
|
|
assert actual_citations == expected_citations, \
|
|
f"Row {idx}: citations mismatch. Expected {expected_citations}, got {actual_citations}"
|
|
|
|
print(f"Citations only test passed! {len(test)} rows, {len(all_citations)} total citations extracted")
|
|
|
|
|
|
def test_external_links_and_citations():
|
|
"""Test that both --external-links and --citations work together (shared parser)."""
|
|
import mwparserfromhell
|
|
from wikiq.wikitext_parser import WikitextParser
|
|
|
|
tester = WikiqTester(SAILORMOON, "external_links_and_citations", in_compression="7z", out_format="jsonl")
|
|
|
|
try:
|
|
# Also include --text so we can verify extraction against actual wikitext
|
|
tester.call_wikiq("--external-links", "--citations", "--text", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
|
|
# Verify both columns exist
|
|
assert "external_links" in test.columns, "external_links column should exist"
|
|
assert "citations" in test.columns, "citations column should exist"
|
|
|
|
# Verify both columns have list/array types (pandas reads pyarrow lists as numpy arrays)
|
|
assert test["external_links"].apply(lambda x: x is None or hasattr(x, '__len__')).all(), \
|
|
"external_links should be a list/array type or None"
|
|
assert test["citations"].apply(lambda x: x is None or hasattr(x, '__len__')).all(), \
|
|
"citations should be a list/array type or None"
|
|
|
|
# Verify URLs look like valid URIs (have a scheme or are protocol-relative)
|
|
all_urls = []
|
|
for links in test["external_links"]:
|
|
if links is not None and len(links) > 0:
|
|
all_urls.extend(links)
|
|
|
|
for url in all_urls:
|
|
# External links can be http, https, mailto, ftp, etc. or protocol-relative (//)
|
|
has_scheme = ":" in url and url.index(":") < 10 # scheme:... with short scheme
|
|
is_protocol_relative = url.startswith("//")
|
|
assert has_scheme or is_protocol_relative, \
|
|
f"External link should be a valid URI, got: {url}"
|
|
|
|
# Verify citations have correct prefixes
|
|
all_citations = []
|
|
for citations in test["citations"]:
|
|
if citations is not None and len(citations) > 0:
|
|
all_citations.extend(citations)
|
|
|
|
for citation in all_citations:
|
|
assert citation.startswith("ref:") or citation.startswith("template:"), \
|
|
f"Citation should start with 'ref:' or 'template:', got: {citation}"
|
|
|
|
# Verify extraction matches WikitextParser for a sample of rows with text
|
|
# This tests that the shared parser optimization works correctly
|
|
parser = WikitextParser()
|
|
rows_with_content = test[
|
|
(test["external_links"].apply(lambda x: x is not None and len(x) > 0)) |
|
|
(test["citations"].apply(lambda x: x is not None and len(x) > 0))
|
|
]
|
|
if len(rows_with_content) > 0:
|
|
# Test up to 5 rows
|
|
sample = rows_with_content.head(5)
|
|
for idx, row in sample.iterrows():
|
|
text = row["text"]
|
|
if text:
|
|
# Verify external links
|
|
wikicode = mwparserfromhell.parse(text)
|
|
expected_links = [str(link.url) for link in wikicode.filter_external_links()]
|
|
actual_links = list(row["external_links"]) if row["external_links"] is not None else []
|
|
assert actual_links == expected_links, \
|
|
f"Row {idx}: external_links mismatch. Expected {expected_links}, got {actual_links}"
|
|
|
|
# Verify citations
|
|
expected_citations = parser.extract_citations(text)
|
|
actual_citations = list(row["citations"]) if row["citations"] is not None else []
|
|
assert actual_citations == expected_citations, \
|
|
f"Row {idx}: citations mismatch. Expected {expected_citations}, got {actual_citations}"
|
|
|
|
print(f"External links and citations test passed! {len(test)} rows, {len(all_urls)} URLs, {len(all_citations)} citations")
|
|
|
|
|
|
def test_no_wikitext_columns():
|
|
"""Test that neither external_links nor citations columns exist without flags."""
|
|
tester = WikiqTester(SAILORMOON, "no_wikitext_columns", in_compression="7z", out_format="jsonl")
|
|
|
|
try:
|
|
tester.call_wikiq("--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
|
|
# Verify neither column exists
|
|
assert "external_links" not in test.columns, "external_links column should NOT exist without --external-links flag"
|
|
assert "citations" not in test.columns, "citations column should NOT exist without --citations flag"
|
|
|
|
print(f"No wikitext columns test passed! {len(test)} rows processed")
|
|
|
|
|
|
def test_wikilinks():
|
|
"""Test that --wikilinks extracts internal wikilinks correctly."""
|
|
import mwparserfromhell
|
|
|
|
tester = WikiqTester(SAILORMOON, "wikilinks", in_compression="7z", out_format="jsonl")
|
|
|
|
try:
|
|
tester.call_wikiq("--wikilinks", "--text", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
|
|
# Verify wikilinks column exists
|
|
assert "wikilinks" in test.columns, "wikilinks column should exist"
|
|
|
|
# Verify column has list/array type
|
|
assert test["wikilinks"].apply(lambda x: x is None or hasattr(x, '__len__')).all()
|
|
|
|
# Verify extraction matches mwparserfromhell for sample rows
|
|
rows_with_links = test[test["wikilinks"].apply(lambda x: x is not None and len(x) > 0)]
|
|
if len(rows_with_links) > 0:
|
|
sample = rows_with_links.head(5)
|
|
for idx, row in sample.iterrows():
|
|
text = row["text"]
|
|
if text:
|
|
wikicode = mwparserfromhell.parse(text)
|
|
expected = []
|
|
for link in wikicode.filter_wikilinks():
|
|
title = str(link.title).strip()
|
|
display_text = str(link.text).strip() if link.text else None
|
|
expected.append({"title": title, "text": display_text})
|
|
|
|
actual = list(row["wikilinks"])
|
|
# Convert to comparable format (pandas may read as dicts or named tuples)
|
|
actual_dicts = [{"title": item["title"], "text": item["text"]} for item in actual]
|
|
assert actual_dicts == expected, f"Row {idx}: wikilinks mismatch"
|
|
|
|
print(f"Wikilinks test passed! {len(test)} rows processed")
|
|
|
|
|
|
def test_templates():
|
|
"""Test that --templates extracts templates correctly."""
|
|
import mwparserfromhell
|
|
|
|
tester = WikiqTester(SAILORMOON, "templates", in_compression="7z", out_format="jsonl")
|
|
|
|
try:
|
|
tester.call_wikiq("--templates", "--text", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
|
|
# Verify templates column exists
|
|
assert "templates" in test.columns, "templates column should exist"
|
|
|
|
# Verify column has list/array type
|
|
assert test["templates"].apply(lambda x: x is None or hasattr(x, '__len__')).all()
|
|
|
|
# Verify extraction matches mwparserfromhell for sample rows
|
|
rows_with_templates = test[test["templates"].apply(lambda x: x is not None and len(x) > 0)]
|
|
if len(rows_with_templates) > 0:
|
|
sample = rows_with_templates.head(5)
|
|
for idx, row in sample.iterrows():
|
|
text = row["text"]
|
|
if text:
|
|
wikicode = mwparserfromhell.parse(text)
|
|
expected = []
|
|
for template in wikicode.filter_templates():
|
|
name = str(template.name).strip()
|
|
params = {}
|
|
for param in template.params:
|
|
param_name = str(param.name).strip()
|
|
param_value = str(param.value).strip()
|
|
params[param_name] = param_value
|
|
expected.append({"name": name, "params": params})
|
|
|
|
actual = list(row["templates"])
|
|
# Convert to comparable format
|
|
actual_list = []
|
|
for item in actual:
|
|
actual_list.append({
|
|
"name": item["name"],
|
|
"params": dict(item["params"]) if item["params"] else {}
|
|
})
|
|
assert actual_list == expected, f"Row {idx}: templates mismatch"
|
|
|
|
print(f"Templates test passed! {len(test)} rows processed")
|
|
|
|
|
|
def test_headings():
|
|
"""Test that --headings extracts section headings correctly."""
|
|
import mwparserfromhell
|
|
|
|
tester = WikiqTester(SAILORMOON, "headings", in_compression="7z", out_format="jsonl")
|
|
|
|
try:
|
|
tester.call_wikiq("--headings", "--text", "--fandom-2020")
|
|
except subprocess.CalledProcessError as exc:
|
|
pytest.fail(exc.stderr.decode("utf8"))
|
|
|
|
test = pd.read_json(tester.output, lines=True)
|
|
|
|
# Verify headings column exists
|
|
assert "headings" in test.columns, "headings column should exist"
|
|
|
|
# Verify column has list/array type
|
|
assert test["headings"].apply(lambda x: x is None or hasattr(x, '__len__')).all()
|
|
|
|
# Verify extraction matches mwparserfromhell for sample rows
|
|
rows_with_headings = test[test["headings"].apply(lambda x: x is not None and len(x) > 0)]
|
|
if len(rows_with_headings) > 0:
|
|
sample = rows_with_headings.head(5)
|
|
for idx, row in sample.iterrows():
|
|
text = row["text"]
|
|
if text:
|
|
wikicode = mwparserfromhell.parse(text)
|
|
expected = []
|
|
for heading in wikicode.filter_headings():
|
|
level = heading.level
|
|
heading_text = str(heading.title).strip()
|
|
expected.append({"level": level, "text": heading_text})
|
|
|
|
actual = list(row["headings"])
|
|
# Convert to comparable format
|
|
actual_list = [{"level": item["level"], "text": item["text"]} for item in actual]
|
|
assert actual_list == expected, f"Row {idx}: headings mismatch"
|
|
|
|
print(f"Headings test passed! {len(test)} rows processed")
|
|
|
|
|
|
|