jobtrackingapp/tools/summarizer/tests/test_document_extractor.py

"""Real, synthetic documents; no model artifacts or inference."""

import io
import socket
import sys
import zipfile
import subprocess
import json
from pathlib import Path

import pytest
from docx import Document
from pypdf import PdfWriter
from pypdf.generic import DecodedStreamObject, DictionaryObject, NameObject, TextStringObject

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import parser_child


def docx_bytes():
    document = Document()
    document.add_heading("Erfaring", level=1)
    document.add_paragraph("Utvikler | Eksempel AS | 2021–2023")
    document.add_paragraph("Bygget sikre API-er", style="List Bullet")
    table = document.add_table(rows=1, cols=2)
    table.cell(0, 0).text = "Språk"
    table.cell(0, 1).text = "Norsk, English"
    output = io.BytesIO()
    document.save(output)
    return output.getvalue()


def pdf_bytes(*, pages=2, external=False):
    writer = PdfWriter()
    for number in range(pages):
        page = writer.add_blank_page(width=612, height=792)
        font = DictionaryObject({NameObject("/Type"): NameObject("/Font"), NameObject("/Subtype"): NameObject("/Type1"), NameObject("/BaseFont"): NameObject("/Helvetica")})
        page[NameObject("/Resources")] = DictionaryObject({NameObject("/Font"): DictionaryObject({NameObject("/F1"): writer._add_object(font)})})
        stream = DecodedStreamObject()
        stream.set_data(f"BT /F1 12 Tf 40 740 Td (Experience page {number + 1}) Tj 0 -20 Td (Engineer 2020-2022) Tj ET".encode())
        page[NameObject("/Contents")] = writer._add_object(stream)
    if external:
        writer._root_object[NameObject("/OpenAction")] = DictionaryObject({NameObject("/S"): NameObject("/URI"), NameObject("/URI"): TextStringObject("http://127.0.0.1/private")})
    output = io.BytesIO()
    writer.write(output)
    return output.getvalue()


@pytest.fixture(autouse=True)
def no_network(monkeypatch):
    def denied(*args, **kwargs):
        raise AssertionError("Document extraction attempted network access")
    monkeypatch.setattr(socket.socket, "connect", denied)
    monkeypatch.setattr(socket, "getaddrinfo", denied)


def test_docx_preserves_ordered_sections_lists_tables_and_unknown_pages():
    result = parser_child.extract_document(".docx", docx_bytes())
    assert result["method"] == "docling-docx-v1"
    assert result["ocr_used"] is False
    assert result["page_count"] is None
    assert [block["kind"] for block in result["blocks"]] == ["section_header", "paragraph", "list_item", "table"]
    assert result["blocks"][0] == {"id": "b0001", "text": "Erfaring", "kind": "section_header", "page": None}
    assert "2021–2023" in result["blocks"][1]["text"]
    assert "Norsk, English" in result["blocks"][3]["text"]
    assert "docx_page_numbers_unavailable" in result["warnings"]


def test_native_pdf_keeps_page_evidence_and_declares_layout_limits():
    result = parser_child.extract_document(".pdf", pdf_bytes())
    assert result["method"] == "docling-native-pdf-v1"
    assert result["page_count"] == 2
    assert [block["page"] for block in result["blocks"]] == [1, 1, 2, 2]
    assert [block["id"] for block in result["blocks"]] == ["b0001", "b0002", "b0003", "b0004"]
    assert "Experience page 1" in result["blocks"][0]["text"]
    assert "pdf_native_order_no_layout_or_tables" in result["warnings"]
    assert "ocr_disabled" in result["warnings"]


@pytest.mark.parametrize("extension,data,code", [
    (".pdf", b"not pdf", "pdf_signature"),
    (".pdf", b"%PDF-broken", "pdf_invalid"),
    (".pdf", pdf_bytes(pages=41), "pdf_page_limit"),
    (".pdf", pdf_bytes(external=True), "external_content"),
    (".txt", b"plain text", "unsupported_type"),
], ids=["signature", "malformed", "pages", "external", "type"])
def test_rejects_invalid_or_active_documents_before_conversion(extension, data, code):
    with pytest.raises(parser_child.ParserError) as failure:
        parser_child.extract_document(extension, data)
    assert failure.value.code == code


def test_docx_rejects_external_relationships():
    payload = io.BytesIO()
    with zipfile.ZipFile(io.BytesIO(docx_bytes())) as source, zipfile.ZipFile(payload, "w", zipfile.ZIP_DEFLATED) as output:
        for entry in source.infolist():
            body = source.read(entry)
            if entry.filename == "word/_rels/document.xml.rels":
                body = body.replace(b"</Relationships>", b'<Relationship Id="external" Type="image" TargetMode="External" Target="file:///etc/passwd"/></Relationships>')
            output.writestr(entry.filename, body)
    with pytest.raises(parser_child.ParserError) as failure:
        parser_child.extract_document(".docx", payload.getvalue())
    assert failure.value.code == "external_content"


def test_docx_rejects_archive_traversal():
    payload = io.BytesIO()
    with zipfile.ZipFile(io.BytesIO(docx_bytes())) as source, zipfile.ZipFile(payload, "w", zipfile.ZIP_DEFLATED) as output:
        for entry in source.infolist():
            output.writestr(entry.filename, source.read(entry))
        output.writestr("../secret", "inert")
    with pytest.raises(parser_child.ParserError) as failure:
        parser_child.extract_document(".docx", payload.getvalue())
    assert failure.value.code == "docx_unsafe_archive"


def test_extraction_refuses_truncated_output_instead_of_silently_losing_text(monkeypatch):
    monkeypatch.setattr(parser_child, "MAX_EXTRACT_CHARS", 10)
    with pytest.raises(parser_child.ParserError) as failure:
        parser_child.extract_document(".pdf", pdf_bytes())
    assert failure.value.code == "extracted_text_limit"


def test_document_child_uses_real_converter_and_cleans_no_source_files(tmp_path):
    source, output = tmp_path / "input.bin", tmp_path / "result.json"
    source.write_bytes(docx_bytes())
    result = subprocess.run([sys.executable, str(Path(parser_child.__file__)), "--input", str(source), "--output", str(output), "--extension", ".docx", "--mode", "document", "--memory-bytes", str(1024 ** 3)], capture_output=True, timeout=20)
    assert result.returncode == 0, (result.stderr, output.read_text() if output.exists() else "missing output")
    assert json.loads(output.read_text(encoding="utf-8"))["method"] == "docling-docx-v1"
    assert source.read_bytes().startswith(b"PK")


def test_document_child_forbids_socket_and_external_process_side_effects():
    for event in ["socket.connect", "socket.connect_ex", "socket.getaddrinfo", "subprocess.Popen", "os.system"]:
        with pytest.raises(parser_child.ParserError) as failure:
            parser_child._deny_import_side_effects(event, ())
        assert failure.value.code == "external_content"