govoplan-connectors/tests/test_mediawiki_transport.py

from __future__ import annotations

import json
from urllib.parse import parse_qs, urlsplit
import unittest
from unittest.mock import patch

from govoplan_core.security.http_fetch import HttpFetchResponse
from govoplan_connectors.backend.mediawiki_transport import HttpMediaWikiTransport


def response(payload: dict[str, object]) -> HttpFetchResponse:
    return HttpFetchResponse(
        status=200,
        headers={"Content-Type": "application/json"},
        body=json.dumps(payload).encode(),
    )


class MediaWikiHttpTransportTests(unittest.TestCase):
    def setUp(self) -> None:
        self.transport = HttpMediaWikiTransport()

    @patch("govoplan_connectors.backend.mediawiki_transport.fetch_http")
    def test_discovery_uses_the_action_api_and_sanitized_auth_header(self, fetch) -> None:
        fetch.return_value = response(
            {"query": {"general": {"generator": "MediaWiki 1.43"}}}
        )

        self.transport.discover(
            endpoint_url="https://wiki.example.test",
            credential={"access_token": "secret-token"},
        )

        call = fetch.call_args
        self.assertEqual("https://wiki.example.test/api.php", urlsplit(call.args[0])._replace(query="").geturl())
        query = parse_qs(urlsplit(call.args[0]).query)
        self.assertEqual(["query"], query["action"])
        self.assertEqual(["siteinfo|userinfo"], query["meta"])
        self.assertEqual("Bearer secret-token", call.kwargs["headers"]["Authorization"])
        self.assertNotIn("secret-token", call.args[0])
        self.assertEqual(10_000_000, call.kwargs["max_bytes"])

    @patch("govoplan_connectors.backend.mediawiki_transport.fetch_http")
    def test_full_backfill_resolves_page_details_and_cursor(self, fetch) -> None:
        fetch.side_effect = (
            response(
                {
                    "curtimestamp": "2026-08-22T10:00:00Z",
                    "continue": {"apcontinue": "Next_Page"},
                    "query": {
                        "allpages": [
                            {"pageid": 41, "ns": 0, "title": "First"},
                            {"pageid": 42, "ns": 0, "title": "Second"},
                        ]
                    },
                }
            ),
            response(
                {
                    "query": {
                        "pages": [
                            {"pageid": 41, "ns": 0, "title": "First"},
                            {"pageid": 42, "ns": 0, "title": "Second"},
                        ]
                    }
                }
            ),
        )

        batch = self.transport.changes(
            endpoint_url="https://wiki.example.test/api.php",
            credential=None,
            cursor=None,
            limit=2,
            force_full=True,
        )

        self.assertEqual(2, len(batch.changes))
        self.assertEqual("Next_Page", batch.next_cursor)
        self.assertFalse(batch.complete)
        details = parse_qs(urlsplit(fetch.call_args_list[1].args[0]).query)
        self.assertEqual(["41|42"], details["pageids"])
        self.assertIn("revisions", details["prop"][0])

    @patch("govoplan_connectors.backend.mediawiki_transport.fetch_http")
    def test_delta_preserves_deletion_log_tombstone_and_cursor(self, fetch) -> None:
        fetch.return_value = response(
            {
                "curtimestamp": "2026-08-22T11:00:00Z",
                "query": {
                    "recentchanges": [
                        {
                            "type": "log",
                            "logtype": "delete",
                            "logid": 700,
                            "pageid": 0,
                            "ns": 0,
                            "title": "Deleted page",
                            "timestamp": "2026-08-22T10:59:00Z",
                        }
                    ]
                },
            }
        )

        batch = self.transport.changes(
            endpoint_url="https://wiki.example.test",
            credential=None,
            cursor="rccontinue-token",
            limit=50,
            force_full=False,
        )

        self.assertEqual("delete", batch.changes[0]["change_kind"])
        self.assertEqual("logid:700", batch.changes[0]["change_cursor"])
        self.assertTrue(batch.complete)

    @patch("govoplan_connectors.backend.mediawiki_transport.fetch_http")
    def test_publish_uses_csrf_body_expected_revision_and_canonical_url(self, fetch) -> None:
        fetch.return_value = response(
            {
                "edit": {
                    "result": "Success",
                    "pageid": 99,
                    "oldrevid": 900,
                    "newrevid": 901,
                    "title": "Published Guide",
                }
            }
        )

        result = self.transport.publish(
            endpoint_url="https://wiki.example.test",
            credential={"csrf_token": "csrf-secret", "access_token": "token"},
            title="Published Guide",
            body="Reviewed body",
            summary="Approved",
            expected_revision="900",
            minor=True,
        )

        call = fetch.call_args
        self.assertEqual("POST", call.kwargs["method"])
        body = parse_qs(call.kwargs["body"].decode())
        self.assertEqual(["edit"], body["action"])
        self.assertEqual(["900"], body["baserevid"])
        self.assertEqual(["1"], body["minor"])
        self.assertEqual(["csrf-secret"], body["token"])
        self.assertEqual("99", result.page_id)
        self.assertEqual(
            "https://wiki.example.test/wiki/Published_Guide",
            result.canonical_url,
        )


if __name__ == "__main__":
    unittest.main()