forked from HAL9000/cleveragents-core
hamza.khyari
18b9d61e35
Add content_hash(resource, *, algorithm='sha256') -> str to the ResourceHandler protocol and all handler implementations: - Protocol: new content_hash method on ResourceHandler (protocol.py) - BaseResourceHandler: default impl hashes file content or directory entry names; returns EMPTY_CONTENT_HASH sentinel for missing resources - GitCheckoutHandler: hashes git rev-parse HEAD through the requested algorithm for consistent digest format - FsDirectoryHandler: recursive walk hashing sorted relative paths and file contents (content-only, ignores metadata) - DevcontainerHandler: hashes devcontainer.json config file - DatabaseResourceHandler: hashes connection string; for SQLite file-based DBs, hashes the database file content - _DefaultHandler: delegates to BaseResourceHandler EMPTY_CONTENT_HASH sentinel is the SHA-256 of empty input (e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855). Hash algorithm is configurable via the algorithm parameter (default sha256, accepts any hashlib.new()-compatible name). Behave tests (10 scenarios): sentinel for missing/nonexistent, determinism, different content produces different hash, fs-directory recursive hash with change detection, git-checkout hash, configurable algorithm (sha256 vs sha512), protocol compliance for all 4 handlers. ISSUES CLOSED: #837
714 lines
25 KiB
Python
714 lines
25 KiB
Python
"""Step definitions for resource_handler_content_hash.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.domain.models.core.resource import PhysVirt, Resource
|
|
from cleveragents.resource.handlers._base import EMPTY_CONTENT_HASH, BaseResourceHandler
|
|
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
|
|
from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
|
|
|
|
|
|
def _make_resource(location: str | None = None) -> Resource:
|
|
"""Create a minimal Resource for testing."""
|
|
from ulid import ULID
|
|
|
|
return Resource(
|
|
resource_id=str(ULID()),
|
|
resource_type_name="fs-directory",
|
|
name="test-resource",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=location,
|
|
)
|
|
|
|
|
|
# ── Given steps ──────────────────────────────────────────────
|
|
|
|
|
|
@given("a resource with no location for content_hash")
|
|
def step_resource_no_location(context: object) -> None:
|
|
"""Create a resource with no location."""
|
|
context.ch_resource = _make_resource(location=None) # type: ignore[attr-defined]
|
|
|
|
|
|
@given('a resource with location "{path}" for content_hash')
|
|
def step_resource_with_location(context: object, path: str) -> None:
|
|
"""Create a resource with a specific location."""
|
|
context.ch_resource = _make_resource(location=path) # type: ignore[attr-defined]
|
|
|
|
|
|
@given('a temporary file with content "{content}" for content_hash')
|
|
def step_temp_file(context: object, content: str) -> None:
|
|
"""Create a temporary file with given content."""
|
|
fd, path = tempfile.mkstemp(suffix=".txt")
|
|
os.write(fd, content.encode())
|
|
os.close(fd)
|
|
context.ch_resource = _make_resource(location=path) # type: ignore[attr-defined]
|
|
context.ch_temp_path = path # type: ignore[attr-defined]
|
|
|
|
|
|
@given('a second temporary file with content "{content}" for content_hash')
|
|
def step_second_temp_file(context: object, content: str) -> None:
|
|
"""Create a second temporary file."""
|
|
fd, path = tempfile.mkstemp(suffix=".txt")
|
|
os.write(fd, content.encode())
|
|
os.close(fd)
|
|
context.ch_resource2 = _make_resource(location=path) # type: ignore[attr-defined]
|
|
|
|
|
|
@given("a temporary directory with files for content_hash")
|
|
def step_temp_dir_with_files(context: object) -> None:
|
|
"""Create a temp directory with test files."""
|
|
tmpdir = tempfile.mkdtemp(prefix="ch-test-")
|
|
Path(tmpdir, "file_a.txt").write_text("alpha")
|
|
Path(tmpdir, "file_b.txt").write_text("beta")
|
|
Path(tmpdir, "sub").mkdir()
|
|
Path(tmpdir, "sub", "file_c.txt").write_text("gamma")
|
|
context.ch_resource = _make_resource(location=tmpdir) # type: ignore[attr-defined]
|
|
context.ch_temp_dir = tmpdir # type: ignore[attr-defined]
|
|
|
|
|
|
@given("a temporary git repository for content_hash")
|
|
def step_temp_git_repo(context: object) -> None:
|
|
"""Create a temp git repo with a commit."""
|
|
tmpdir = tempfile.mkdtemp(prefix="ch-git-")
|
|
subprocess.run(
|
|
["git", "init"],
|
|
cwd=tmpdir,
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "test@test.com"],
|
|
cwd=tmpdir,
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "Test"],
|
|
cwd=tmpdir,
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
Path(tmpdir, "README.md").write_text("# Test")
|
|
subprocess.run(
|
|
["git", "add", "."],
|
|
cwd=tmpdir,
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "init"],
|
|
cwd=tmpdir,
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
context.ch_resource = _make_resource(location=tmpdir) # type: ignore[attr-defined]
|
|
|
|
|
|
# ── When steps ───────────────────────────────────────────────
|
|
|
|
|
|
@when("I compute the content_hash via base handler")
|
|
def step_compute_base(context: object) -> None:
|
|
"""Compute hash via BaseResourceHandler."""
|
|
handler = BaseResourceHandler()
|
|
handler._default_strategy = "none" # type: ignore[assignment]
|
|
handler._type_label = "test" # type: ignore[assignment]
|
|
context.ch_result = handler.content_hash(context.ch_resource) # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I compute the content_hash twice via base handler")
|
|
def step_compute_twice(context: object) -> None:
|
|
"""Compute hash twice to verify determinism."""
|
|
handler = BaseResourceHandler()
|
|
handler._default_strategy = "none" # type: ignore[assignment]
|
|
handler._type_label = "test" # type: ignore[assignment]
|
|
res = context.ch_resource # type: ignore[attr-defined]
|
|
context.ch_result = handler.content_hash(res) # type: ignore[attr-defined]
|
|
context.ch_result2 = handler.content_hash(res) # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I compute content_hash for both files")
|
|
def step_compute_both(context: object) -> None:
|
|
"""Compute hashes for two different resources."""
|
|
handler = BaseResourceHandler()
|
|
handler._default_strategy = "none" # type: ignore[assignment]
|
|
handler._type_label = "test" # type: ignore[assignment]
|
|
context.ch_result = handler.content_hash(context.ch_resource) # type: ignore[attr-defined]
|
|
context.ch_result2 = handler.content_hash(context.ch_resource2) # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I compute the content_hash via FsDirectoryHandler")
|
|
def step_compute_fsdir(context: object) -> None:
|
|
"""Compute hash via FsDirectoryHandler."""
|
|
handler = FsDirectoryHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource) # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I modify a file in the directory")
|
|
def step_modify_file(context: object) -> None:
|
|
"""Modify a file in the temp directory."""
|
|
context.ch_prev_result = context.ch_result # type: ignore[attr-defined]
|
|
tmpdir = context.ch_temp_dir # type: ignore[attr-defined]
|
|
Path(tmpdir, "file_a.txt").write_text("alpha-modified")
|
|
|
|
|
|
@when("I compute the content_hash via FsDirectoryHandler again")
|
|
def step_compute_fsdir_again(context: object) -> None:
|
|
"""Re-compute hash after modification."""
|
|
handler = FsDirectoryHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource) # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I compute the content_hash via GitCheckoutHandler")
|
|
def step_compute_git(context: object) -> None:
|
|
"""Compute hash via GitCheckoutHandler."""
|
|
handler = GitCheckoutHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource) # type: ignore[attr-defined]
|
|
|
|
|
|
@when('I compute the content_hash with algorithm "{algo}"')
|
|
def step_compute_with_algo(context: object, algo: str) -> None:
|
|
"""Compute hash with a specific algorithm."""
|
|
handler = BaseResourceHandler()
|
|
handler._default_strategy = "none" # type: ignore[assignment]
|
|
handler._type_label = "test" # type: ignore[assignment]
|
|
context.ch_result = handler.content_hash( # type: ignore[attr-defined]
|
|
context.ch_resource,
|
|
algorithm=algo, # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
# ── Then steps ───────────────────────────────────────────────
|
|
|
|
|
|
@then("the content_hash result should be the EMPTY sentinel")
|
|
def step_is_sentinel(context: object) -> None:
|
|
"""Assert the result equals EMPTY_CONTENT_HASH."""
|
|
assert context.ch_result == EMPTY_CONTENT_HASH, ( # type: ignore[attr-defined]
|
|
f"Expected sentinel, got {context.ch_result}" # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
@then("the content_hash result should not be the EMPTY sentinel")
|
|
def step_not_sentinel(context: object) -> None:
|
|
"""Assert the result is not the sentinel."""
|
|
assert context.ch_result != EMPTY_CONTENT_HASH, ( # type: ignore[attr-defined]
|
|
"Expected non-sentinel hash"
|
|
)
|
|
|
|
|
|
@then("the two content_hash results should be identical")
|
|
def step_results_identical(context: object) -> None:
|
|
"""Assert two hash results are the same."""
|
|
assert context.ch_result == context.ch_result2, ( # type: ignore[attr-defined]
|
|
f"Hashes differ: {context.ch_result} vs {context.ch_result2}" # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
@then("the two content_hash results should differ")
|
|
def step_results_differ(context: object) -> None:
|
|
"""Assert two hash results are different."""
|
|
assert context.ch_result != context.ch_result2, ( # type: ignore[attr-defined]
|
|
f"Hashes should differ but both are {context.ch_result}" # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
@then("the two directory hashes should differ")
|
|
def step_dir_hashes_differ(context: object) -> None:
|
|
"""Assert directory hash changed after modification."""
|
|
assert context.ch_prev_result != context.ch_result, ( # type: ignore[attr-defined]
|
|
"Directory hash should change after file modification"
|
|
)
|
|
|
|
|
|
@then("the content_hash length should be {length:d}")
|
|
def step_hash_length(context: object, length: int) -> None:
|
|
"""Assert the hash has the expected hex-string length."""
|
|
actual = len(context.ch_result) # type: ignore[attr-defined]
|
|
assert actual == length, f"Expected length {length}, got {actual}"
|
|
|
|
|
|
@then("{handler_name} should have a content_hash method")
|
|
def step_has_method(context: object, handler_name: str) -> None:
|
|
"""Assert the handler class has a content_hash method."""
|
|
handlers = {
|
|
"GitCheckoutHandler": GitCheckoutHandler,
|
|
"FsDirectoryHandler": FsDirectoryHandler,
|
|
}
|
|
# Lazy imports for handlers not already imported
|
|
if handler_name == "DevcontainerHandler":
|
|
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
|
|
|
|
handlers["DevcontainerHandler"] = DevcontainerHandler
|
|
elif handler_name == "DatabaseResourceHandler":
|
|
from cleveragents.resource.handlers.database import DatabaseResourceHandler
|
|
|
|
handlers["DatabaseResourceHandler"] = DatabaseResourceHandler
|
|
|
|
cls = handlers.get(handler_name)
|
|
assert cls is not None, f"Unknown handler: {handler_name}"
|
|
assert hasattr(cls, "content_hash"), f"{handler_name} missing content_hash method"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DevcontainerHandler content_hash steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a temporary devcontainer project for content_hash")
|
|
def step_devcontainer_project(context: Any) -> None:
|
|
tmpdir = tempfile.mkdtemp()
|
|
dc_dir = os.path.join(tmpdir, ".devcontainer")
|
|
os.makedirs(dc_dir)
|
|
config = os.path.join(dc_dir, "devcontainer.json")
|
|
with open(config, "w") as f:
|
|
f.write(
|
|
'{"name": "test", "image": "mcr.microsoft.com/devcontainers/base:ubuntu"}\n'
|
|
)
|
|
context.ch_resource = Resource(
|
|
resource_id="01DCTEST00000000000000DC01",
|
|
resource_type_name="devcontainer-file",
|
|
name="test-devcontainer",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=tmpdir,
|
|
)
|
|
context._ch_tmpdir = tmpdir
|
|
|
|
|
|
@when("I compute the content_hash via DevcontainerHandler")
|
|
def step_devcontainer_hash(context: Any) -> None:
|
|
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
|
|
|
|
handler = DevcontainerHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@when("I compute the content_hash via DevcontainerHandler for missing")
|
|
def step_devcontainer_hash_missing(context: Any) -> None:
|
|
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
|
|
|
|
handler = DevcontainerHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@then("the devcontainer hash should be from config file fallback")
|
|
def step_devcontainer_config_fallback(context: Any) -> None:
|
|
# If no running container, the hash comes from config file
|
|
# Verify it's not empty (config file was hashed)
|
|
assert context.ch_result != EMPTY_CONTENT_HASH
|
|
assert len(context.ch_result) == 64 # SHA-256
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DatabaseResourceHandler content_hash steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a temporary SQLite database for content_hash")
|
|
def step_sqlite_db(context: Any) -> None:
|
|
import sqlite3
|
|
|
|
fd, db_path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
|
|
conn.execute("INSERT INTO users VALUES (1, 'alice')")
|
|
conn.commit()
|
|
conn.close()
|
|
context.ch_resource = Resource(
|
|
resource_id="01DBTEST00000000000000DB01",
|
|
resource_type_name="sqlite",
|
|
name="test-db",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=db_path,
|
|
)
|
|
context._ch_db_path = db_path
|
|
|
|
|
|
@when("I compute the content_hash via DatabaseResourceHandler")
|
|
def step_database_hash(context: Any) -> None:
|
|
from cleveragents.resource.handlers.database import DatabaseResourceHandler
|
|
|
|
handler = DatabaseResourceHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@when("I add a table to the database")
|
|
def step_add_table(context: Any) -> None:
|
|
import sqlite3
|
|
|
|
conn = sqlite3.connect(context._ch_db_path)
|
|
conn.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, total REAL)")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
@when("I compute the content_hash via DatabaseResourceHandler again")
|
|
def step_database_hash_again(context: Any) -> None:
|
|
from cleveragents.resource.handlers.database import DatabaseResourceHandler
|
|
|
|
handler = DatabaseResourceHandler()
|
|
context.ch_result2 = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@then("the two database hashes should differ")
|
|
def step_database_hashes_differ(context: Any) -> None:
|
|
assert context.ch_result != context.ch_result2, (
|
|
f"Expected different hashes after schema change, both are {context.ch_result}"
|
|
)
|
|
|
|
|
|
@given("a remote database resource for content_hash")
|
|
def step_remote_db_resource(context: Any) -> None:
|
|
context.ch_resource = Resource(
|
|
resource_id="01DBTEST00000000000000DB02",
|
|
resource_type_name="postgres",
|
|
name="test-remote-db",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location="postgresql://host/db",
|
|
)
|
|
|
|
|
|
@when("I compute the content_hash via DatabaseResourceHandler for remote")
|
|
def step_database_hash_remote(context: Any) -> None:
|
|
from cleveragents.resource.handlers.database import DatabaseResourceHandler
|
|
|
|
handler = DatabaseResourceHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CloudResourceHandler content_hash steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a cloud resource with ARN location for content_hash")
|
|
def step_cloud_resource_with_location(context: Any) -> None:
|
|
context.ch_resource = Resource(
|
|
resource_id="01AATEST0000000000000CCC01",
|
|
resource_type_name="aws-s3",
|
|
name="test-bucket",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location="arn:aws:s3:::my-bucket",
|
|
)
|
|
|
|
|
|
@given("a cloud resource without location for content_hash")
|
|
def step_cloud_resource_no_location(context: Any) -> None:
|
|
context.ch_resource = Resource(
|
|
resource_id="01AATEST0000000000000CCC02",
|
|
resource_type_name="aws-s3",
|
|
name="test-bucket-noloc",
|
|
classification=PhysVirt.PHYSICAL,
|
|
)
|
|
|
|
|
|
@when("I compute the content_hash via CloudResourceHandler")
|
|
def step_cloud_hash(context: Any) -> None:
|
|
from cleveragents.resource.handlers.cloud import CloudResourceHandler
|
|
|
|
handler = CloudResourceHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@when("I compute the content_hash via CloudResourceHandler for empty")
|
|
def step_cloud_hash_empty(context: Any) -> None:
|
|
from cleveragents.resource.handlers.cloud import CloudResourceHandler
|
|
|
|
handler = CloudResourceHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edge case and error path coverage steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a temporary directory with only subdirectories for content_hash")
|
|
def step_dir_only_subdirs(context: Any) -> None:
|
|
tmpdir = tempfile.mkdtemp()
|
|
os.makedirs(os.path.join(tmpdir, "subdir1"))
|
|
os.makedirs(os.path.join(tmpdir, "subdir2"))
|
|
context.ch_resource = Resource(
|
|
resource_id="01EDGE00000000000000000001",
|
|
resource_type_name="fs-directory",
|
|
name="empty-subdirs",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=tmpdir,
|
|
)
|
|
|
|
|
|
@when("I compute the content_hash via base handler for dir")
|
|
def step_base_hash_dir(context: Any) -> None:
|
|
handler = BaseResourceHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@given("a temporary file pretending to be a directory for content_hash")
|
|
def step_file_as_dir(context: Any) -> None:
|
|
fd, fpath = tempfile.mkstemp(suffix=".txt")
|
|
os.close(fd)
|
|
context.ch_resource = Resource(
|
|
resource_id="01EDGE00000000000000000002",
|
|
resource_type_name="fs-directory",
|
|
name="not-a-dir",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=fpath,
|
|
)
|
|
|
|
|
|
@when("I compute the content_hash via FsDirectoryHandler for non-dir")
|
|
def step_fsdir_hash_nondir(context: Any) -> None:
|
|
handler = FsDirectoryHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@given("a temporary directory with an unreadable file for content_hash")
|
|
def step_dir_unreadable(context: Any) -> None:
|
|
tmpdir = tempfile.mkdtemp()
|
|
good = os.path.join(tmpdir, "good.txt")
|
|
bad = os.path.join(tmpdir, "bad.txt")
|
|
with open(good, "w") as f:
|
|
f.write("readable content")
|
|
with open(bad, "w") as f:
|
|
f.write("secret")
|
|
os.chmod(bad, 0o000)
|
|
context.ch_resource = Resource(
|
|
resource_id="01EDGE00000000000000000003",
|
|
resource_type_name="fs-directory",
|
|
name="has-unreadable",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=tmpdir,
|
|
)
|
|
context._ch_bad_file = bad
|
|
|
|
|
|
@when("I compute the content_hash via FsDirectoryHandler for unreadable")
|
|
def step_fsdir_hash_unreadable(context: Any) -> None:
|
|
handler = FsDirectoryHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
# Restore permissions for cleanup
|
|
os.chmod(context._ch_bad_file, 0o644)
|
|
|
|
|
|
@given("a temporary non-git directory for content_hash")
|
|
def step_nongit_dir(context: Any) -> None:
|
|
tmpdir = tempfile.mkdtemp()
|
|
# Create a corrupt .git file so git rev-parse HEAD fails even if
|
|
# the temp dir is inside a git worktree (CI environments).
|
|
with open(os.path.join(tmpdir, ".git"), "w") as f:
|
|
f.write("not a valid git dir\n")
|
|
with open(os.path.join(tmpdir, "file.txt"), "w") as f:
|
|
f.write("not a git repo")
|
|
context.ch_resource = Resource(
|
|
resource_id="01EDGE00000000000000000004",
|
|
resource_type_name="git-checkout",
|
|
name="not-git",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=tmpdir,
|
|
)
|
|
|
|
|
|
@when("I compute the content_hash via GitCheckoutHandler for non-git")
|
|
def step_git_hash_nongit(context: Any) -> None:
|
|
handler = GitCheckoutHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@given("a temporary file as git location for content_hash")
|
|
def step_file_as_git(context: Any) -> None:
|
|
fd, fpath = tempfile.mkstemp()
|
|
os.close(fd)
|
|
context.ch_resource = Resource(
|
|
resource_id="01EDGE00000000000000000005",
|
|
resource_type_name="git-checkout",
|
|
name="file-not-dir",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=fpath,
|
|
)
|
|
|
|
|
|
@when("I compute the content_hash via GitCheckoutHandler for file")
|
|
def step_git_hash_file(context: Any) -> None:
|
|
handler = GitCheckoutHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@given("a database resource with no location for content_hash")
|
|
def step_db_no_location(context: Any) -> None:
|
|
context.ch_resource = Resource(
|
|
resource_id="01EDGE00000000000000000006",
|
|
resource_type_name="sqlite",
|
|
name="db-noloc",
|
|
classification=PhysVirt.PHYSICAL,
|
|
)
|
|
|
|
|
|
@when("I compute the content_hash via DatabaseResourceHandler for noloc")
|
|
def step_db_hash_noloc(context: Any) -> None:
|
|
from cleveragents.resource.handlers.database import DatabaseResourceHandler
|
|
|
|
handler = DatabaseResourceHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@given("a SQLite database with non-standard extension for content_hash")
|
|
def step_sqlite_nonext(context: Any) -> None:
|
|
import sqlite3
|
|
|
|
fd, db_path = tempfile.mkstemp(suffix=".data")
|
|
os.close(fd)
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("CREATE TABLE t (id INTEGER)")
|
|
conn.commit()
|
|
conn.close()
|
|
context.ch_resource = Resource(
|
|
resource_id="01EDGE00000000000000000007",
|
|
resource_type_name="sqlite",
|
|
name="nonext-db",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=db_path,
|
|
)
|
|
|
|
|
|
@when("I compute the content_hash via DatabaseResourceHandler for nonext")
|
|
def step_db_hash_nonext(context: Any) -> None:
|
|
from cleveragents.resource.handlers.database import DatabaseResourceHandler
|
|
|
|
handler = DatabaseResourceHandler()
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@when("I check _find_running_container")
|
|
def step_check_find_container(context: Any) -> None:
|
|
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
|
|
|
|
handler = DevcontainerHandler()
|
|
context._ch_container_id = handler._find_running_container(context.ch_resource)
|
|
|
|
|
|
@then("the container ID should be None")
|
|
def step_container_id_none(context: Any) -> None:
|
|
assert context._ch_container_id is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mock-based branch coverage steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I compute content_hash with mocked docker exec returning file hashes")
|
|
def step_devcontainer_mock_exec(context: Any) -> None:
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
|
|
|
|
handler = DevcontainerHandler()
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
mock_result.stdout = (
|
|
"abc123 /workspaces/project/file1.py\ndef456 /workspaces/project/file2.py\n"
|
|
)
|
|
|
|
with (
|
|
patch("subprocess.run", return_value=mock_result),
|
|
patch.object(handler, "_find_running_container", return_value="abc123def456"),
|
|
):
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@when("I compute content_hash with docker ps raising OSError")
|
|
def step_devcontainer_oserror(context: Any) -> None:
|
|
from unittest.mock import patch
|
|
|
|
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
|
|
|
|
handler = DevcontainerHandler()
|
|
|
|
with patch(
|
|
"cleveragents.resource.handlers.devcontainer.subprocess.run",
|
|
side_effect=OSError("docker not found"),
|
|
):
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@then("the devcontainer falls back to config hash")
|
|
def step_devcontainer_fallback(context: Any) -> None:
|
|
# Should have fallen back to config file hash (not empty)
|
|
assert context.ch_result != EMPTY_CONTENT_HASH
|
|
assert len(context.ch_result) == 64
|
|
|
|
|
|
@given("a temporary git-checkout resource for content_hash timeout")
|
|
def step_git_timeout_resource(context: Any) -> None:
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.ch_resource = Resource(
|
|
resource_id="01EDGEMCK000000000000000MK",
|
|
resource_type_name="git-checkout",
|
|
name="mock-git",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=tmpdir,
|
|
)
|
|
|
|
|
|
@when("I compute content_hash with git rev-parse timing out")
|
|
def step_git_timeout(context: Any) -> None:
|
|
from unittest.mock import patch
|
|
|
|
handler = GitCheckoutHandler()
|
|
|
|
with patch(
|
|
"subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired(cmd="git", timeout=10),
|
|
):
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@when("I compute content_hash with git rev-parse raising OSError")
|
|
def step_git_oserror(context: Any) -> None:
|
|
from unittest.mock import patch
|
|
|
|
handler = GitCheckoutHandler()
|
|
|
|
with patch("subprocess.run", side_effect=OSError("git not found")):
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|
|
|
|
|
|
@given("a resource pointing to a non-file non-dir path for content_hash")
|
|
def step_special_file(context: Any) -> None:
|
|
|
|
# Use a real temp path but mock os.path to make it exist but be neither file nor dir
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.ch_resource = Resource(
|
|
resource_id="01EDGE00000000000000SPEC01",
|
|
resource_type_name="test",
|
|
name="special",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=tmpdir,
|
|
)
|
|
context._ch_special_tmpdir = tmpdir
|
|
|
|
|
|
@when("I compute the content_hash via base handler for special")
|
|
def step_base_hash_special(context: Any) -> None:
|
|
from unittest.mock import patch
|
|
|
|
handler = BaseResourceHandler()
|
|
|
|
# Mock: path exists, but is neither file nor dir (e.g., a socket)
|
|
with (
|
|
patch("os.path.exists", return_value=True),
|
|
patch("os.path.isfile", return_value=False),
|
|
patch("os.path.isdir", return_value=False),
|
|
):
|
|
context.ch_result = handler.content_hash(context.ch_resource)
|