arborist/tests/test_wikipedia_old.py

"""WikipediaSqlDump on the 'old' (revision history) table."""

from __future__ import annotations

import bz2
from pathlib import Path

import pytest

from arborist.sources import WikipediaOldDump, WikipediaSqlDump


# Minimal fabricated 'old' table dump matching the 2003-05-16 schema.
# Columns: old_id, old_namespace, old_title, old_text, old_comment,
#          old_user, old_user_text, old_timestamp, old_minor_edit, old_flags,
#          inverse_timestamp
_SAMPLE_SQL = """\
-- MySQL dump
DROP TABLE IF EXISTS old;
CREATE TABLE old (
  old_id int(8) unsigned NOT NULL auto_increment
);
INSERT INTO old VALUES (1, 0, 'Anarchism', 'first revision text [[liberty]] [[freedom]]', 'init', 0, 'alice', '20020101000000', 0, '', '79979898'),(2, 0, 'Anarchism', 'second revision text with [[autonomy]]', 'edit', 0, 'bob', '20020201000000', 0, '', '79979897'),(3, 1, 'Talk:Anarchism', 'talk page (skipped: namespace 1)', 'tk', 0, 'carol', '20020301000000', 0, '', '79979896');
"""


def test_old_dump_yields_revisions(tmp_path):
    p = tmp_path / "fake_old.sql.bz2"
    with bz2.open(p, "wt", encoding="utf-8") as f:
        f.write(_SAMPLE_SQL)

    src = WikipediaOldDump(path=p)
    docs = list(src.iter_documents())
    # Talk:Anarchism (namespace 1) is filtered; main namespace yields 2 revisions.
    assert len(docs) == 2

    # Revision 1
    assert docs[0].title == "Anarchism"
    assert docs[0].source_type == "wikipedia_old"
    assert "first revision" in docs[0].content
    assert docs[0].extra["old_timestamp"] == "20020101000000"
    assert docs[0].extra["old_id"] == "1"
    assert any(e.dst_uri.endswith("liberty") for e in docs[0].edges)

    # Revision 2 — same title, different content, different timestamp.
    assert docs[1].title == "Anarchism"
    assert "second revision" in docs[1].content
    assert docs[1].extra["old_timestamp"] == "20020201000000"
    assert docs[1].extra["old_id"] == "2"


def test_cur_and_old_share_uri_namespace(tmp_path):
    """Both tables produce the same URI for the same article title."""
    cur_sql = (
        "INSERT INTO cur VALUES "
        "(7, 0, 'Anarchism', 'cur revision text', '', 0, 'sys', '20030516000000', "
        "'', 0, 0, 0, 0, 0, '', '');\n"
    )
    old_sql = (
        "INSERT INTO old VALUES "
        "(11, 0, 'Anarchism', 'old revision text', '', 0, 'sys', "
        "'20020101000000', 0, '', '');\n"
    )
    cur_path = tmp_path / "c.sql.bz2"
    old_path = tmp_path / "o.sql.bz2"
    with bz2.open(cur_path, "wt", encoding="utf-8") as f:
        f.write(cur_sql)
    with bz2.open(old_path, "wt", encoding="utf-8") as f:
        f.write(old_sql)

    cur_doc = next(WikipediaSqlDump(cur_path, table="cur").iter_documents())
    old_doc = next(WikipediaSqlDump(old_path, table="old").iter_documents())
    assert cur_doc.uri == old_doc.uri
    assert cur_doc.source_type == "wikipedia_cur"
    assert old_doc.source_type == "wikipedia_old"
    # But content differs -> different document_root.
    assert cur_doc.content != old_doc.content


def test_real_dump_smoke():
    """If the real concatenated old dump exists locally, parse the first few rows."""
    real = Path("data/20030516_old_tablesql.bz2")
    if not real.exists():
        pytest.skip("real old dump not fetched; run 'make fetch-old'")
    src = WikipediaOldDump(path=real)
    docs: list = []
    for d in src.iter_documents():
        docs.append(d)
        if len(docs) >= 3:
            break
    assert len(docs) == 3
    for d in docs:
        assert d.title
        assert d.content
        assert d.extra.get("old_id")
        assert d.extra.get("old_timestamp")