Building/.template/development/tests/run-tests.py

from __future__ import annotations

import argparse
import itertools
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path


ROOT = Path(__file__).resolve().parents[3]
TEST_ROOT = ROOT / ".template" / "development" / "tests"
MANIFEST_PATH = TEST_ROOT / "manifest.json"
SNAPSHOT_ROOT = TEST_ROOT / "snapshots"


class TestFailure(RuntimeError):
    pass


def check_architecture() -> None:
    """Проверяет направление зависимостей без запуска Typst renderer."""
    template_root = ROOT / ".template" / "lib"
    import_pattern = re.compile(r'#import\s+"([^"]+)"')
    graph: dict[Path, set[Path]] = {}

    for source in template_root.rglob("*.typ"):
        graph[source.resolve()] = set()
        text = source.read_text(encoding="utf-8")
        source_area = source.relative_to(template_root).parts[0]
        code_without_line_comments = re.sub(r"//.*", "", text)
        if source_area == "domain":
            forbidden_runtime = re.search(
                r"(?:#(?:set|show)\b|\b(?:page|place|image|query|state)\s*\()",
                code_without_line_comments,
            )
            if forbidden_runtime:
                raise TestFailure(
                    f"Domain-модуль {source.relative_to(ROOT)} содержит presentation/runtime вызов "
                    f"{forbidden_runtime.group(0)!r}."
                )
        for raw_target in import_pattern.findall(text):
            if raw_target.startswith("@"):
                continue
            target = (source.parent / raw_target).resolve()
            try:
                relative = target.relative_to(template_root.resolve())
            except ValueError:
                continue
            if target.suffix == ".typ":
                graph[source.resolve()].add(target)

            target_area = relative.parts[0]
            forbidden = {
                "domain": {"application", "infrastructure", "presentation"},
                "infrastructure": {"application", "presentation"},
                "application": {"presentation"},
            }.get(source_area, set())
            if target_area in forbidden:
                raise TestFailure(
                    "Нарушена DDD-граница: "
                    f"{source.relative_to(ROOT)} -> {target.relative_to(ROOT)}"
                )

        if re.search(r'#import\s+"[^\"]*(?:main\.typ|chapters/|assets/|\.private/)', text):
            raise TestFailure(
                f"Library-модуль {source.relative_to(ROOT)} импортирует пользовательский слой."
            )

    visiting: set[Path] = set()
    visited: set[Path] = set()

    def visit(node: Path, stack: list[Path]) -> None:
        if node in visiting:
            cycle = stack[stack.index(node):] + [node]
            rendered = " -> ".join(str(item.relative_to(ROOT)) for item in cycle)
            raise TestFailure(f"Циклические Typst-импорты: {rendered}")
        if node in visited:
            return
        visiting.add(node)
        stack.append(node)
        for target in graph.get(node, set()):
            if target in graph:
                visit(target, stack)
        stack.pop()
        visiting.remove(node)
        visited.add(node)

    for node in graph:
        visit(node, [])


def check_workspace() -> None:
    """Проверяет минимальный корень и синхронизируемую конфигурацию VS Code."""
    allowed_directories = {
        ".git",
        ".private",
        ".template",
        ".vscode",
        "assets",
        "chapters",
        "docs",
    }
    actual_directories = {item.name for item in ROOT.iterdir() if item.is_dir()}
    unexpected = actual_directories - allowed_directories
    if unexpected:
        raise TestFailure(
            "В корне обнаружены лишние каталоги: " + ", ".join(sorted(unexpected))
        )

    required_files = {
        ".gitignore",
        "README.md",
        "main.typ",
    }
    missing_files = {name for name in required_files if not (ROOT / name).is_file()}
    if missing_files:
        raise TestFailure(
            "В корне отсутствуют обязательные файлы: " + ", ".join(sorted(missing_files))
        )

    extensions_path = ROOT / ".vscode" / "extensions.json"
    settings_path = ROOT / ".vscode" / "settings.json"
    extensions = json.loads(extensions_path.read_text(encoding="utf-8"))
    recommendations = set(extensions.get("recommendations", []))
    required_extensions = {
        "alefragnani.bookmarks",
        "streetsidesoftware.code-spell-checker",
        "streetsidesoftware.code-spell-checker-russian",
        "ms-ceintl.vscode-language-pack-ru",
        "myriad-dreamin.tinymist",
        "wayou.vscode-todo-highlight",
        "gruntfuggly.todo-tree",
        "mhutchie.git-graph",
        "local.typst-typewriter",
        "zotst.zotst",
    }
    missing_extensions = required_extensions - recommendations
    if missing_extensions:
        raise TestFailure(
            "В extensions.json отсутствуют рекомендации: "
            + ", ".join(sorted(missing_extensions))
        )

    settings = json.loads(settings_path.read_text(encoding="utf-8"))
    if settings.get("files.autoSave") != "afterDelay":
        raise TestFailure("Workspace должен включать files.autoSave=afterDelay.")
    if settings.get("files.exclude", {}).get(".template") is not True:
        raise TestFailure(".template должен быть скрыт в пользовательском Explorer VS Code.")
    if settings.get("files.exclude", {}).get(".vscode") is not True:
        raise TestFailure(".vscode должен быть скрыт в пользовательском Explorer VS Code.")
    if settings.get("files.exclude", {}).get("docs") is True:
        raise TestFailure("Публичная документация docs/ не должна быть скрыта.")
    if settings.get("typstTypewriter.navigator.mainFiles") != ["main.typ"]:
        raise TestFailure("Typewriter должен показывать единственный main.typ.")

    gitignore = (ROOT / ".gitignore").read_text(encoding="utf-8")
    if ".private/" not in gitignore:
        raise TestFailure("В .gitignore отсутствует защита каталога .private/.")
    if ".vscode/" in gitignore:
        raise TestFailure(".vscode должен синхронизироваться через Git.")

    private_example = ROOT / "docs" / "examples" / "private" / "settings.typ"
    if not private_example.is_file():
        raise TestFailure("Отсутствует безопасный пример .private/settings.typ.")
    private_example_text = private_example.read_text(encoding="utf-8")
    for employee_id in (
        "musikhin", "guzeev", "fedorov", "ilyasov", "khimichev",
        "brusnicin", "ozornin", "buhartdinov", "tkachenko",
    ):
        if f"{employee_id}:" not in private_example_text:
            raise TestFailure(
                f"В примере private settings отсутствует сотрудник {employee_id}."
            )

    tasks = json.loads((ROOT / ".vscode" / "tasks.json").read_text(encoding="utf-8"))
    task_labels = {task.get("label") for task in tasks.get("tasks", [])}
    if "Scientia: собрать PDF с приватными данными" in task_labels:
        raise TestFailure("Отдельная private build task больше не должна использоваться.")
    if (ROOT / ".template" / "development" / "tools" / "build-with-private-assets.ps1").exists():
        raise TestFailure("Устаревший ZIP installer должен быть удалён.")

    for company_id in ("scientia", "technology", "too"):
        company_path = ROOT / ".template" / "companies" / company_id / "data.json"
        company = json.loads(company_path.read_text(encoding="utf-8"))
        if company.get("sign_image") is not None or company.get("stamp_image") is not None:
            raise TestFailure(
                f"Публичный профиль {company_id} не должен содержать подпись или печать."
            )

    public_docs = (
        "README.md",
        "documents.md",
        "formatting.md",
        "vscode.md",
        "git.md",
        "private-assets.md",
        "writing-style.md",
        "troubleshooting.md",
    )
    for name in public_docs:
        if not (ROOT / "docs" / name).is_file():
            raise TestFailure(f"Публичная документация неполна: docs/{name}.")

    for starter in ("report", "letter", "commercial-offer", "contract"):
        starter_root = ROOT / "docs" / "examples" / "documents" / starter
        if not (starter_root / "main.typ").is_file() or not (starter_root / "chapters").is_dir():
            raise TestFailure(f"Публичный пример {starter} неполон.")

    for obsolete in ("document.typ", "draft.typ", "clean-copy.typ"):
        if (ROOT / obsolete).exists():
            raise TestFailure(f"Лишняя точка входа в корне: {obsolete}.")


def run(command: list[str], *, cwd: Path = ROOT) -> str:
    result = subprocess.run(
        command,
        cwd=cwd,
        text=True,
        encoding="utf-8",
        errors="replace",
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )
    if result.returncode != 0:
        rendered = subprocess.list2cmdline(command)
        raise TestFailure(f"Команда завершилась с кодом {result.returncode}:\n{rendered}\n{result.stdout}")
    return result.stdout


def find_executable(
    name: str,
    *,
    env_var: str | None = None,
    required: bool = True,
) -> Path | None:
    if env_var and os.environ.get(env_var):
        candidate = Path(os.environ[env_var])
        if candidate.is_file():
            return candidate

    located = shutil.which(name)
    if located:
        return Path(located)

    executable = name + (".exe" if os.name == "nt" else "")
    candidates: list[Path] = []

    poppler_bin = os.environ.get("POPPLER_BIN")
    if poppler_bin:
        candidates.append(Path(poppler_bin) / executable)

    if os.name == "nt":
        candidates.append(
            Path.home()
            / ".cache"
            / "codex-runtimes"
            / "codex-primary-runtime"
            / "dependencies"
            / "native"
            / "poppler"
            / "Library"
            / "bin"
            / executable
        )

    for candidate in candidates:
        if candidate.is_file():
            return candidate

    if required:
        raise TestFailure(
            f"Не найден исполняемый файл {name}. Добавьте его в PATH"
            + (f" или задайте {env_var}" if env_var else "")
            + ("/POPPLER_BIN." if name.startswith("pdf") else ".")
        )
    return None


def parse_version(value: str) -> tuple[int, ...]:
    match = re.search(r"(\d+)\.(\d+)\.(\d+)", value)
    if not match:
        raise TestFailure(f"Не удалось определить версию из строки: {value!r}")
    return tuple(int(part) for part in match.groups())


def load_manifest() -> dict:
    return json.loads(MANIFEST_PATH.read_text(encoding="utf-8"))


def compile_fixture(
    typst: Path,
    entry: Path,
    output: Path,
    creation_timestamp: int,
    inputs: dict[str, str] | None = None,
) -> str:
    command = [
        str(typst),
        "compile",
        "--root",
        str(ROOT),
        "--creation-timestamp",
        str(creation_timestamp),
    ]
    for key, value in sorted((inputs or {}).items()):
        command.extend(("--input", f"{key}={value}"))
    command.extend((str(entry), str(output)))
    return run(command)


def compile_expected_failure(
    typst: Path,
    entry: Path,
    output: Path,
    creation_timestamp: int,
    required_errors: list[str],
    inputs: dict[str, str] | None = None,
) -> None:
    command = [
        str(typst),
        "compile",
        "--root",
        str(ROOT),
        "--creation-timestamp",
        str(creation_timestamp),
    ]
    for key, value in sorted((inputs or {}).items()):
        command.extend(("--input", f"{key}={value}"))
    command.extend((str(entry), str(output)))
    result = subprocess.run(
        command,
        cwd=ROOT,
        text=True,
        encoding="utf-8",
        errors="replace",
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )
    if result.returncode == 0:
        raise TestFailure(f"{entry.relative_to(ROOT)}: ожидалась ошибка компиляции.")
    for marker in required_errors:
        if marker not in result.stdout:
            raise TestFailure(
                f"{entry.relative_to(ROOT)}: в diagnostic не найден маркер {marker!r}.\n"
                f"Фактический вывод:\n{result.stdout}"
            )


def run_unit_test(typst: Path, module: str, creation_timestamp: int) -> None:
    expression = f'import "/{module.replace(os.sep, "/")}": run; run()'
    output = run(
        [
            str(typst),
            "eval",
            "--root",
            str(ROOT),
            "--creation-timestamp",
            str(creation_timestamp),
            expression,
        ]
    ).strip()
    if json.loads(output) != "ok":
        raise TestFailure(f"Unit-модуль {module} вернул неожиданное значение: {output}")


def inspect_pdf(pdfinfo: Path, pdf: Path) -> tuple[int, str]:
    info = run([str(pdfinfo), str(pdf)])
    pages_match = re.search(r"^Pages:\s+(\d+)$", info, flags=re.MULTILINE)
    size_match = re.search(r"^Page size:.*\(([^)]+)\)$", info, flags=re.MULTILINE)
    if not pages_match or not size_match:
        raise TestFailure(f"Не удалось разобрать pdfinfo для {pdf}:\n{info}")
    return int(pages_match.group(1)), size_match.group(1).strip()


def extract_text(pdftotext: Path | None, pdf: Path, target: Path) -> str:
    if pdftotext:
        run([str(pdftotext), "-layout", str(pdf), str(target)])
        return target.read_text(encoding="utf-8", errors="replace")

    try:
        from pypdf import PdfReader
    except ImportError as exc:
        raise TestFailure(
            "Для проверки текста требуется pdftotext или Python-пакет pypdf."
        ) from exc

    reader = PdfReader(str(pdf))
    return "\n".join((page.extract_text() or "") for page in reader.pages)


def count_pdf_links(pdf: Path) -> tuple[int, int]:
    try:
        from pypdf import PdfReader
    except ImportError as exc:
        raise TestFailure("Для проверки PDF-ссылок требуется Python-пакет pypdf.") from exc

    external = internal = 0
    for page in PdfReader(str(pdf)).pages:
        for raw_annotation in page.get("/Annots", ()):
            annotation = raw_annotation.get_object()
            if annotation.get("/Subtype") != "/Link":
                continue
            action = annotation.get("/A")
            if action is not None and action.get_object().get("/URI") is not None:
                external += 1
            else:
                internal += 1
    return external, internal


def render_page(pdftoppm: Path, pdf: Path, page: int, dpi: int, target: Path) -> Path:
    target.parent.mkdir(parents=True, exist_ok=True)
    run(
        [
            str(pdftoppm),
            "-png",
            "-r",
            str(dpi),
            "-f",
            str(page),
            "-l",
            str(page),
            "-singlefile",
            str(pdf),
            str(target.with_suffix("")),
        ]
    )
    return target


def image_difference(expected: Path, actual: Path) -> float:
    try:
        from PIL import Image, ImageChops
    except ImportError as exc:
        raise TestFailure("Для visual regression требуется пакет Pillow.") from exc

    with Image.open(expected).convert("RGB") as left, Image.open(actual).convert("RGB") as right:
        if left.size != right.size:
            return 1.0
        diff = ImageChops.difference(left, right)
        histogram = diff.histogram()
        total = sum(value * (index % 256) for index, value in enumerate(histogram))
        maximum = left.width * left.height * 3 * 255
        return total / maximum


def main() -> int:
    parser = argparse.ArgumentParser(description="Регрессионные тесты шаблона документов Scientia")
    parser.add_argument("--fixture", action="append", help="Запустить только fixture с указанным id")
    parser.add_argument("--compile-only", action="store_true", help="Не запускать PDF и visual проверки")
    parser.add_argument(
        "--update-snapshots",
        action="store_true",
        help="Явно обновить visual snapshots после ручного подтверждения",
    )
    args = parser.parse_args()

    manifest = load_manifest()
    typst = find_executable("typst", env_var="TYPST_BIN")
    version_output = run([str(typst), "--version"]).strip()
    if parse_version(version_output) < parse_version(manifest["typst_min"]):
        raise TestFailure(
            f"Требуется Typst {manifest['typst_min']}+, найден {version_output}."
        )

    pdfinfo = pdftotext = pdftoppm = None
    if not args.compile_only:
        pdfinfo = find_executable("pdfinfo")
        pdftotext = find_executable("pdftotext", required=False)
        pdftoppm = find_executable("pdftoppm")

    creation_timestamp = int(manifest["creation_timestamp"])
    selected = set(args.fixture or [])
    known = {
        item["id"]
        for item in (
            *manifest["fixtures"],
            *manifest.get("negative_fixtures", []),
            *manifest.get("compile_matrices", []),
        )
    }
    unknown = selected - known
    if unknown:
        raise TestFailure(f"Неизвестные fixtures: {', '.join(sorted(unknown))}")

    if not selected:
        print("[static] architecture")
        check_architecture()
        print("[static] workspace")
        check_workspace()

    for unit in manifest.get("unit_tests", []):
        if selected:
            continue
        print(f"[unit] {unit['id']}")
        run_unit_test(typst, unit["module"], creation_timestamp)

    temp_dir = Path(tempfile.mkdtemp(prefix="scientia-tests-"))
    failed = False
    try:
        for matrix in manifest.get("compile_matrices", []):
            if selected and matrix["id"] not in selected:
                continue
            keys = tuple(matrix["parameters"])
            values = tuple(matrix["parameters"][key] for key in keys)
            for combination in itertools.product(*values):
                inputs = dict(zip(keys, combination))
                suffix = ",".join(f"{key}={value}" for key, value in inputs.items())
                print(f"[matrix] {matrix['id']}[{suffix}]")
                safe_suffix = re.sub(r"[^A-Za-z0-9_.-]+", "-", suffix)
                compile_fixture(
                    typst,
                    ROOT / matrix["entry"],
                    temp_dir / f"{matrix['id']}-{safe_suffix}.pdf",
                    creation_timestamp,
                    inputs=inputs,
                )

        for fixture in manifest.get("negative_fixtures", []):
            if selected and fixture["id"] not in selected:
                continue
            fixture_id = fixture["id"]
            print(f"[negative] {fixture_id}")
            compile_expected_failure(
                typst,
                ROOT / fixture["entry"],
                temp_dir / f"{fixture_id}.pdf",
                creation_timestamp,
                fixture.get("required_errors", []),
                inputs=fixture.get("inputs"),
            )

        for fixture in manifest["fixtures"]:
            if selected and fixture["id"] not in selected:
                continue

            fixture_id = fixture["id"]
            print(f"[compile] {fixture_id}")
            entry = ROOT / fixture["entry"]
            output = temp_dir / f"{fixture_id}.pdf"
            compile_fixture(
                typst,
                entry,
                output,
                creation_timestamp,
                inputs=fixture.get("inputs"),
            )

            if args.compile_only:
                continue

            assert pdfinfo and pdftoppm
            pages, page_size = inspect_pdf(pdfinfo, output)
            expected_pages = int(fixture["expected_pages"])
            if expected_pages and pages != expected_pages:
                raise TestFailure(
                    f"{fixture_id}: ожидалось {expected_pages} страниц, получено {pages}."
                )
            if page_size != fixture["expected_page_size"]:
                raise TestFailure(
                    f"{fixture_id}: ожидался формат {fixture['expected_page_size']}, получен {page_size}."
                )

            text = extract_text(pdftotext, output, temp_dir / f"{fixture_id}.txt")
            for marker in fixture.get("required_text", []):
                if marker not in text:
                    raise TestFailure(f"{fixture_id}: в PDF не найден текстовый маркер {marker!r}.")
            for marker, minimum in fixture.get("required_text_counts", {}).items():
                actual = text.count(marker)
                if actual < int(minimum):
                    raise TestFailure(
                        f"{fixture_id}: маркер {marker!r} найден {actual} раз; "
                        f"требуется не менее {minimum}."
                    )
            if "minimum_external_links" in fixture or "minimum_internal_links" in fixture:
                external_links, internal_links = count_pdf_links(output)
                minimum_external = int(fixture.get("minimum_external_links", 0))
                minimum_internal = int(fixture.get("minimum_internal_links", 0))
                if external_links < minimum_external:
                    raise TestFailure(
                        f"{fixture_id}: внешних PDF-ссылок {external_links}; "
                        f"требуется не менее {minimum_external}."
                    )
                if internal_links < minimum_internal:
                    raise TestFailure(
                        f"{fixture_id}: внутренних PDF-ссылок {internal_links}; "
                        f"требуется не менее {minimum_internal}."
                    )

            for page in fixture.get("snapshot_pages", []):
                if page > pages:
                    raise TestFailure(f"{fixture_id}: snapshot page {page} больше количества страниц {pages}.")
                actual = render_page(
                    pdftoppm,
                    output,
                    page,
                    int(manifest["dpi"]),
                    temp_dir / fixture_id / f"page-{page}.png",
                )
                expected = SNAPSHOT_ROOT / fixture_id / f"page-{page}.png"
                if args.update_snapshots:
                    expected.parent.mkdir(parents=True, exist_ok=True)
                    shutil.copy2(actual, expected)
                    print(f"[snapshot:update] {expected.relative_to(ROOT)}")
                    continue
                if not expected.is_file():
                    raise TestFailure(
                        f"Нет snapshot {expected.relative_to(ROOT)}. "
                        "После ручной проверки запустите --update-snapshots."
                    )
                difference = image_difference(expected, actual)
                threshold = float(fixture.get("visual_threshold", manifest["visual_threshold"]))
                if difference > threshold:
                    raise TestFailure(
                        f"{fixture_id}, страница {page}: visual diff {difference:.6f} "
                        f"превышает порог {threshold:.6f}. Фактический файл: {actual}"
                    )
                print(f"[snapshot:ok] {fixture_id}/page-{page} diff={difference:.6f}")

        print(f"OK: {version_output}")
        return 0
    except Exception:
        failed = True
        raise
    finally:
        if failed:
            print(f"Временные результаты сохранены: {temp_dir}", file=sys.stderr)
        else:
            shutil.rmtree(temp_dir, ignore_errors=True)


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except TestFailure as exc:
        print(f"FAIL: {exc}", file=sys.stderr)
        raise SystemExit(1)