"""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)