cleveragents-core/robot/helper_resource_handler_content_hash.py

"""Robot Framework helper for ResourceHandler content_hash integration tests."""

from __future__ import annotations

import importlib
import os
import sys
import tempfile
from pathlib import Path

_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.resource.handlers._base import (  # noqa: E402
    EMPTY_CONTENT_HASH,
    BaseResourceHandler,
)
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler  # noqa: E402


def _make_resource(location: str) -> object:
    """Create a minimal resource-like object with a location attribute."""

    class _FakeResource:
        def __init__(self, loc: str) -> None:
            self.location = loc
            self.resource_id = "01TEST00000000000000000001"
            self.type_name = "fs-directory"
            self.resource_kind = "physical"

    return _FakeResource(location)


def _run_deterministic() -> None:
    """Same content produces same hash twice."""
    with tempfile.TemporaryDirectory() as tmpdir:
        fpath = os.path.join(tmpdir, "test.txt")
        with open(fpath, "w") as f:
            f.write("hello world\n")

        handler = BaseResourceHandler()
        res = _make_resource(fpath)
        h1 = handler.content_hash(res)  # type: ignore[arg-type]
        h2 = handler.content_hash(res)  # type: ignore[arg-type]
        assert h1 == h2, f"Hashes differ: {h1} != {h2}"
        assert h1 != EMPTY_CONTENT_HASH, "Got empty hash for non-empty file"
        print("deterministic-ok")


def _run_empty_sentinel() -> None:
    """Missing resource returns EMPTY_CONTENT_HASH."""
    handler = BaseResourceHandler()
    res = _make_resource("/nonexistent/path/that/does/not/exist")
    h = handler.content_hash(res)  # type: ignore[arg-type]
    assert h == EMPTY_CONTENT_HASH, f"Expected sentinel, got {h}"
    print("empty-sentinel-ok")


def _run_content_change_detection() -> None:
    """Hash changes when file content changes."""
    with tempfile.TemporaryDirectory() as tmpdir:
        fpath = os.path.join(tmpdir, "test.txt")
        with open(fpath, "w") as f:
            f.write("version 1\n")

        handler = BaseResourceHandler()
        res = _make_resource(fpath)
        h1 = handler.content_hash(res)  # type: ignore[arg-type]

        with open(fpath, "w") as f:
            f.write("version 2\n")

        h2 = handler.content_hash(res)  # type: ignore[arg-type]
        assert h1 != h2, "Hash should change when content changes"
        print("content-change-detection-ok")


def _run_configurable_algorithm() -> None:
    """Algorithm parameter changes the hash output."""
    with tempfile.TemporaryDirectory() as tmpdir:
        fpath = os.path.join(tmpdir, "test.txt")
        with open(fpath, "w") as f:
            f.write("test content\n")

        handler = BaseResourceHandler()
        res = _make_resource(fpath)
        h_sha256 = handler.content_hash(res, algorithm="sha256")  # type: ignore[arg-type]
        h_md5 = handler.content_hash(res, algorithm="md5")  # type: ignore[arg-type]
        assert h_sha256 != h_md5, "Different algorithms should produce different hashes"
        assert len(h_sha256) == 64, (
            f"SHA-256 should be 64 hex chars, got {len(h_sha256)}"
        )
        assert len(h_md5) == 32, f"MD5 should be 32 hex chars, got {len(h_md5)}"
        print("configurable-algorithm-ok")


def _run_fs_directory_recursive() -> None:
    """FsDirectoryHandler hashes directory contents recursively."""
    with tempfile.TemporaryDirectory() as tmpdir:
        os.makedirs(os.path.join(tmpdir, "sub"))
        with open(os.path.join(tmpdir, "a.txt"), "w") as f:
            f.write("file a\n")
        with open(os.path.join(tmpdir, "sub", "b.txt"), "w") as f:
            f.write("file b\n")

        handler = FsDirectoryHandler()
        res = _make_resource(tmpdir)
        h = handler.content_hash(res)  # type: ignore[arg-type]
        assert h != EMPTY_CONTENT_HASH, "Directory hash should not be empty"
        assert len(h) == 64, f"Expected 64 hex chars, got {len(h)}"
        print("fs-directory-recursive-ok")


_COMMANDS = {
    "deterministic": _run_deterministic,
    "empty-sentinel": _run_empty_sentinel,
    "content-change-detection": _run_content_change_detection,
    "configurable-algorithm": _run_configurable_algorithm,
    "fs-directory-recursive": _run_fs_directory_recursive,
}

if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "all"
    if cmd == "all":
        for fn in _COMMANDS.values():
            fn()
    elif cmd in _COMMANDS:
        _COMMANDS[cmd]()
    else:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)