add xxh64
This commit is contained in:
@@ -0,0 +1,91 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Behavior tests for plot_probability_map.py."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image
|
||||
|
||||
import plot_probability_map
|
||||
|
||||
|
||||
class ProbabilityMapPlotTests(unittest.TestCase):
|
||||
def test_reads_bit_columns_and_ignores_operation_and_pairs(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temporary_directory:
|
||||
path = Path(temporary_directory) / "hash.csv"
|
||||
path.write_text(
|
||||
"operation,pairs,bit_0,bit_1\n"
|
||||
"replace,4,0.25,0.75\n"
|
||||
"delete,2,0.5,nan\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
rows = plot_probability_map.read_probability_map(path)
|
||||
|
||||
self.assertEqual(rows[0].operation, "replace")
|
||||
self.assertEqual(rows[0].pairs, 4)
|
||||
self.assertEqual(rows[0].probabilities, [0.25, 0.75])
|
||||
self.assertEqual(rows[1].operation, "delete")
|
||||
self.assertEqual(rows[1].pairs, 2)
|
||||
self.assertEqual(len(rows[1].probabilities), 2)
|
||||
|
||||
def test_mean_absolute_deviation_from_half_ignores_nan(self) -> None:
|
||||
result = plot_probability_map.mean_absolute_deviation(
|
||||
[0.25, 0.5, 0.75, float("nan")]
|
||||
)
|
||||
|
||||
self.assertEqual(result, 1 / 6)
|
||||
|
||||
def test_shared_scale_zooms_to_all_values_and_reference(self) -> None:
|
||||
scale = plot_probability_map.make_scale(
|
||||
[0.45, 0.48, 0.52, 0.55],
|
||||
reference=0.5,
|
||||
hard_limits=(0.0, 1.0),
|
||||
)
|
||||
|
||||
self.assertGreater(scale.minimum, 0.0)
|
||||
self.assertLess(scale.maximum, 1.0)
|
||||
self.assertLessEqual(scale.minimum, 0.45)
|
||||
self.assertGreaterEqual(scale.maximum, 0.55)
|
||||
self.assertIn(0.5, scale.ticks)
|
||||
|
||||
def test_deviation_scale_uses_data_range_instead_of_fixed_half(self) -> None:
|
||||
scale = plot_probability_map.make_scale(
|
||||
[0.05, 0.08], hard_limits=(0.0, 0.5)
|
||||
)
|
||||
|
||||
self.assertGreater(scale.minimum, 0.0)
|
||||
self.assertLess(scale.maximum, 0.5)
|
||||
self.assertLessEqual(scale.minimum, 0.05)
|
||||
self.assertGreaterEqual(scale.maximum, 0.08)
|
||||
|
||||
def test_generates_two_nonempty_png_files(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temporary_directory:
|
||||
root = Path(temporary_directory)
|
||||
source = root / "sample.csv"
|
||||
source.write_text(
|
||||
"operation,pairs,bit_0,bit_1,bit_2,bit_3\n"
|
||||
"replace,4,0.25,0.50,0.75,1.0\n"
|
||||
"delete,2,0.10,0.20,0.30,0.40\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
outputs = plot_probability_map.generate_plots(source, root / "plots")
|
||||
|
||||
self.assertEqual(len(outputs), 2)
|
||||
for output in outputs:
|
||||
self.assertTrue(output.is_file())
|
||||
with Image.open(output) as image:
|
||||
self.assertEqual(image.format, "PNG")
|
||||
self.assertGreater(image.width, 300)
|
||||
self.assertGreater(image.height, 200)
|
||||
colors = image.convert("RGB").getcolors(maxcolors=1_000_000)
|
||||
self.assertIsNotNone(colors)
|
||||
self.assertGreater(len(colors or []), 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Behavior tests for probability_map.py."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import io
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import redirect_stderr
|
||||
from pathlib import Path
|
||||
|
||||
import generate_input
|
||||
import probability_map
|
||||
|
||||
|
||||
class PrepareHashTests(unittest.TestCase):
|
||||
def test_compiles_c_source_when_binary_is_missing(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temporary_directory:
|
||||
root = Path(temporary_directory)
|
||||
hash_directory = root / "constant"
|
||||
hash_directory.mkdir()
|
||||
(hash_directory / "bin_hash.c").write_text(
|
||||
"#include <stdio.h>\n"
|
||||
"int main(void) { puts(\"00000001\"); return 0; }\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
executable = probability_map.prepare_hash("constant", root)
|
||||
|
||||
self.assertEqual(executable, hash_directory / "bin_hash")
|
||||
self.assertEqual(
|
||||
subprocess.check_output([executable, "word"], text=True).strip(),
|
||||
"00000001",
|
||||
)
|
||||
|
||||
def test_adds_python_shebang_and_execute_permission(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temporary_directory:
|
||||
root = Path(temporary_directory)
|
||||
hash_directory = root / "python_hash"
|
||||
hash_directory.mkdir()
|
||||
source = hash_directory / "bin_hash.py"
|
||||
source.write_text("print('00000002')\n", encoding="utf-8")
|
||||
|
||||
executable = probability_map.prepare_hash("python_hash", root)
|
||||
|
||||
self.assertEqual(executable, source)
|
||||
self.assertTrue(os.access(source, os.X_OK))
|
||||
self.assertTrue(
|
||||
source.read_text(encoding="utf-8").startswith(
|
||||
"#!/usr/bin/env python3\n"
|
||||
)
|
||||
)
|
||||
self.assertEqual(
|
||||
subprocess.check_output([executable, "word"], text=True).strip(),
|
||||
"00000002",
|
||||
)
|
||||
|
||||
|
||||
class ProbabilityTests(unittest.TestCase):
|
||||
def test_counts_changed_hash_bits_relative_to_source(self) -> None:
|
||||
probabilities = probability_map.bit_probabilities(
|
||||
source_hash=0b0000,
|
||||
changed_hashes=[0b0001, 0b0011, 0b0010, 0b0000],
|
||||
bits=4,
|
||||
)
|
||||
|
||||
self.assertEqual(probabilities, [0.5, 0.5, 0.0, 0.0])
|
||||
|
||||
def test_csv_table_has_operation_rows_and_bit_columns(self) -> None:
|
||||
stream = io.StringIO()
|
||||
|
||||
probability_map.write_csv_table(
|
||||
stream,
|
||||
{
|
||||
"replace": (4, [0.25, 0.75]),
|
||||
"delete": (2, [0.5, 0.0]),
|
||||
},
|
||||
)
|
||||
|
||||
rows = list(csv.reader(io.StringIO(stream.getvalue())))
|
||||
self.assertEqual(rows[0], ["operation", "pairs", "bit_0", "bit_1"])
|
||||
self.assertEqual(rows[1], ["replace", "4", "0.250000", "0.750000"])
|
||||
self.assertEqual(rows[2], ["delete", "2", "0.500000", "0.000000"])
|
||||
|
||||
def test_parser_accepts_multiple_source_words(self) -> None:
|
||||
arguments = probability_map.build_parser().parse_args(["first", "second"])
|
||||
|
||||
self.assertEqual(arguments.words, ["first", "second"])
|
||||
|
||||
def test_aggregates_samples_from_multiple_words_and_warns_on_shortfall(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temporary_directory:
|
||||
executable = Path(temporary_directory) / "hash.py"
|
||||
executable.write_text(
|
||||
"#!/usr/bin/env python3\n"
|
||||
"import sys\n"
|
||||
"print(f'{sum(sys.argv[1].encode()):08x}')\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
executable.chmod(0o755)
|
||||
warnings = io.StringIO()
|
||||
|
||||
with redirect_stderr(warnings):
|
||||
table = probability_map.build_probability_table(
|
||||
executable=executable,
|
||||
sources=["A", "B"],
|
||||
operations=["delete"],
|
||||
count=10,
|
||||
operation_count=1,
|
||||
alphabet=generate_input.DEFAULT_ALPHABET,
|
||||
seed=1,
|
||||
max_attempts=None,
|
||||
)
|
||||
|
||||
pair_count, probabilities = table["delete"]
|
||||
self.assertEqual(pair_count, 2)
|
||||
self.assertEqual(
|
||||
probabilities[:8],
|
||||
[0.5, 0.5, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0],
|
||||
)
|
||||
self.assertEqual(warnings.getvalue().count("delete"), 2)
|
||||
|
||||
|
||||
class GenerateWordsTests(unittest.TestCase):
|
||||
def test_count_is_an_upper_bound_when_unique_results_are_exhausted(self) -> None:
|
||||
words = generate_input.generate_words(
|
||||
source="abc",
|
||||
count=100,
|
||||
operation="delete",
|
||||
operation_count=1,
|
||||
seed=42,
|
||||
)
|
||||
|
||||
self.assertEqual(set(words), {"ab", "ac", "bc"})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,90 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Behavior tests for the standalone XXH64 hash CLI."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SOURCE = ROOT / "hash_funcs" / "xxh64" / "bin_hash.c"
|
||||
|
||||
|
||||
class Xxh64BinHashTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.temporary_directory = tempfile.TemporaryDirectory()
|
||||
cls.binary = Path(cls.temporary_directory.name) / "bin_hash"
|
||||
subprocess.run(
|
||||
[
|
||||
"cc",
|
||||
"-std=c11",
|
||||
"-O2",
|
||||
"-Wall",
|
||||
"-Wextra",
|
||||
"-Wpedantic",
|
||||
"-Werror",
|
||||
str(SOURCE),
|
||||
"-o",
|
||||
str(cls.binary),
|
||||
],
|
||||
check=True,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls) -> None:
|
||||
cls.temporary_directory.cleanup()
|
||||
|
||||
def run_hash(self, word: str | None = None, stdin: bytes | None = None) -> subprocess.CompletedProcess[bytes]:
|
||||
command = [str(self.binary)]
|
||||
if word is not None:
|
||||
command.append(word)
|
||||
return subprocess.run(command, input=stdin, capture_output=True, check=False)
|
||||
|
||||
def test_matches_official_xxh64_seed_zero_vectors(self) -> None:
|
||||
vectors = {
|
||||
"": b"ef46db3751d8e999\n",
|
||||
"hello": b"26c7827d889f6da3\n",
|
||||
"HashWord": b"3e26fc2935163fbe\n",
|
||||
}
|
||||
|
||||
for word, expected in vectors.items():
|
||||
with self.subTest(word=word):
|
||||
result = self.run_hash(word)
|
||||
self.assertEqual(result.returncode, 0)
|
||||
self.assertEqual(result.stdout, expected)
|
||||
self.assertRegex(result.stdout.decode(), r"^[0-9a-f]{16}\n$")
|
||||
|
||||
def test_argv_and_stdin_are_equivalent_and_strip_trailing_ascii_space(self) -> None:
|
||||
argv = self.run_hash("hello")
|
||||
stdin = self.run_hash(stdin=b"hello \t\r\n")
|
||||
|
||||
self.assertEqual(stdin.returncode, 0)
|
||||
self.assertEqual(stdin.stdout, argv.stdout)
|
||||
|
||||
def test_handles_long_ascii_input(self) -> None:
|
||||
payload = b"a" * 100_000
|
||||
result = self.run_hash(stdin=payload)
|
||||
reference = subprocess.run(
|
||||
["xxhsum", "-H64"], input=payload, capture_output=True, check=True
|
||||
).stdout.split()[0]
|
||||
|
||||
self.assertEqual(result.returncode, 0)
|
||||
self.assertEqual(result.stdout.strip(), reference)
|
||||
|
||||
def test_rejects_non_ascii_input(self) -> None:
|
||||
result = self.run_hash(stdin="ёж".encode())
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn(b"ASCII", result.stderr)
|
||||
self.assertEqual(result.stdout, b"")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user