mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-14 13:05:09 +00:00
feat: identify the running code by source hash
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
"""Unit tests for the code fingerprint behind the banner and the /api/health `code_hash` field.
|
||||
It runs at import time, so it must always return a string and must never raise. It must change when
|
||||
the framework source changes and stay unchanged when anything else does."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import unshackle.core
|
||||
from unshackle.core import __code_hash__, code_files, code_hash
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
|
||||
def test_is_hex_and_stable() -> None:
|
||||
first = code_hash()
|
||||
assert len(first) == 32
|
||||
assert int(first, 16) >= 0
|
||||
assert first == code_hash()
|
||||
|
||||
|
||||
def test_exported_value_is_the_short_form() -> None:
|
||||
assert __code_hash__ == code_hash()[:7]
|
||||
|
||||
|
||||
def test_covers_the_framework_dirs_only() -> None:
|
||||
rels = code_files()
|
||||
assert "__main__.py" in rels
|
||||
assert any(rel.startswith("core/") for rel in rels)
|
||||
assert not any(rel.startswith("services/") for rel in rels)
|
||||
assert not any("__pycache__" in rel for rel in rels)
|
||||
assert all(rel.endswith(".py") for rel in rels)
|
||||
assert rels == sorted(rels) # order must not depend on the filesystem
|
||||
|
||||
|
||||
def test_content_changes_the_hash() -> None:
|
||||
rels = ["core/__init__.py", "commands/dl.py"]
|
||||
blobs = {"core/__init__.py": b"a", "commands/dl.py": b"b"}
|
||||
before = code_hash(files=rels, read=blobs.__getitem__)
|
||||
|
||||
blobs["commands/dl.py"] = b"b "
|
||||
assert code_hash(files=rels, read=blobs.__getitem__) != before
|
||||
|
||||
|
||||
def test_path_is_part_of_the_hash() -> None:
|
||||
"""A rename with identical bytes must not look like the same code."""
|
||||
same_bytes = {"core/a.py": b"x", "core/b.py": b"x"}
|
||||
assert code_hash(files=["core/a.py"], read=same_bytes.__getitem__) != code_hash(
|
||||
files=["core/b.py"], read=same_bytes.__getitem__
|
||||
)
|
||||
|
||||
|
||||
def test_file_order_does_not_matter_to_the_caller() -> None:
|
||||
"""code_files() sorts, so two orderings of the same set must agree once sorted."""
|
||||
blobs = {"core/a.py": b"x", "core/b.py": b"y"}
|
||||
assert code_hash(files=sorted(blobs), read=blobs.__getitem__) == code_hash(
|
||||
files=sorted(reversed(list(blobs))), read=blobs.__getitem__
|
||||
)
|
||||
|
||||
|
||||
def test_unreadable_source_yields_empty_string() -> None:
|
||||
def unreadable(rel: str) -> bytes:
|
||||
raise PermissionError(rel)
|
||||
|
||||
assert code_hash(files=["core/__init__.py"], read=unreadable) == ""
|
||||
|
||||
|
||||
def test_unwalkable_source_yields_empty_string(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A dir os.walk cannot enter must yield "", not a valid-looking hash of fewer files."""
|
||||
(tmp_path / "__main__.py").write_bytes(b"")
|
||||
monkeypatch.setattr(unshackle.core, "_PKG", tmp_path)
|
||||
assert code_hash() == ""
|
||||
@@ -1,57 +0,0 @@
|
||||
"""Unit tests for the git commit probe behind the banner and the /api/health `commit` field.
|
||||
The probe runs at import time, so it must always return a string and must never raise, whatever
|
||||
state git and the checkout are in."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import unshackle.core
|
||||
from unshackle.core import _git_commit
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
# Without a .git the probe returns before it runs git, so the subprocess stubs below prove nothing.
|
||||
needs_checkout = pytest.mark.skipif(
|
||||
not (Path(unshackle.core.__file__).parents[2] / ".git").exists(), reason="not running from a git checkout"
|
||||
)
|
||||
|
||||
|
||||
def test_returns_str_in_this_checkout() -> None:
|
||||
assert isinstance(_git_commit(), str)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"error",
|
||||
[
|
||||
FileNotFoundError("git"),
|
||||
PermissionError("git"),
|
||||
subprocess.TimeoutExpired(cmd="git", timeout=2),
|
||||
subprocess.SubprocessError("boom"),
|
||||
],
|
||||
)
|
||||
@needs_checkout
|
||||
def test_swallows_subprocess_errors(monkeypatch: pytest.MonkeyPatch, error: Exception) -> None:
|
||||
def raise_it(*args: object, **kwargs: object) -> None:
|
||||
raise error
|
||||
|
||||
monkeypatch.setattr(subprocess, "run", raise_it)
|
||||
assert _git_commit() == ""
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("returncode", "stdout", "expected"),
|
||||
[
|
||||
(0, "a24c9b9\n", "a24c9b9"),
|
||||
(128, "", ""), # empty repo, corrupt repo, or a stale worktree pointer
|
||||
(0, "", ""), # no output despite success
|
||||
],
|
||||
)
|
||||
@needs_checkout
|
||||
def test_maps_git_result(monkeypatch: pytest.MonkeyPatch, returncode: int, stdout: str, expected: str) -> None:
|
||||
completed = subprocess.CompletedProcess(args=["git"], returncode=returncode, stdout=stdout, stderr="noise")
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **k: completed)
|
||||
assert _git_commit() == expected
|
||||
Reference in New Issue
Block a user