"""Tests for content-addressable HDC storage and its file format.""" import os import random import sys import tempfile import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from hdc.hypervec import hvc, random_hv, sim # noqa: E402 from hdc.memory import HDCStorage, pack, unpack # noqa: E402 D = 2000 class TestCells(unittest.TestCase): def test_write_read(self): s = HDCStorage(d=D) hv = random_hv(D, random.Random(1)) s.write("a", hv) self.assertEqual(s.read("a"), hv) self.assertIn("a", s) self.assertEqual(len(s), 1) def test_overwrite_does_not_grow(self): s = HDCStorage(d=D) s.write("a", 1) s.write("a", 2) self.assertEqual(len(s), 1) self.assertEqual(s.read("a"), 2) def test_missing_cell(self): s = HDCStorage(d=D) with self.assertRaises(KeyError): s.read("nope") self.assertEqual(s.read("nope", default=7), 7) def test_capacity_overflow(self): s = HDCStorage(d=200, capacity=2) s.write("x", 1) s.write("y", 2) with self.assertRaises(MemoryError): s.write("z", 3) def test_pack_unpack(self): s = random_hv(D, random.Random(2)) self.assertEqual(unpack(pack(s, D), D), s) self.assertEqual(len(pack(s, D)), (D + 7) // 8) class TestCleanup(unittest.TestCase): def test_nearest_symbol_exact(self): s = HDCStorage(d=D) s.write("alpha", 1, symbol=True) s.write("beta", 2, symbol=True) self.assertEqual(s.symbols(), ["alpha", "beta"]) hits = s.nearest_symbol(s.addr("beta"), k=1) self.assertEqual(hits[0][0], "beta") self.assertEqual(hits[0][1], 1.0) def test_nearest_symbol_noisy(self): s = HDCStorage(d=D) s.write("alpha", 1, symbol=True) s.write("beta", 2, symbol=True) probe = s.addr("alpha") ^ (1 << 4) # one bit of the address wrong name, score = s.nearest_symbol(probe, k=1)[0] self.assertEqual(name, "alpha") self.assertLess(score, 1.0) self.assertGreater(score, 0.9) def test_prefix_filter(self): s = HDCStorage(d=D) s.write("code/0", 1, symbol=True) s.write("tool/x", 2, symbol=True) hits = s.nearest_symbol(s.addr("code/0"), k=5, prefix="tool/") self.assertEqual([h[0] for h in hits], ["tool/x"]) class TestContentAddressing(unittest.TestCase): def setUp(self): self.s = HDCStorage(d=D) self.names = ["r0", "r1", "data/x", "data/y"] for i, n in enumerate(self.names): self.s.write(n, random_hv(D, random.Random(100 + i))) def test_exact_hit_confidence_is_one(self): name, conf = self.s.resolve(self.s.addr("data/x")) self.assertEqual(name, "data/x") self.assertEqual(conf, 1.0) def test_fuzzy_hit(self): addr = self.s.addr("data/y") rng = random.Random(7) # flip ~0.5% of the bits: a corrupted but still recognisable address for pos in rng.sample(range(D), 10): addr ^= 1 << pos got = self.s.resolve(addr) self.assertIsNotNone(got) name, conf = got self.assertEqual(name, "data/y") self.assertLess(conf, 1.0) self.assertGreater(conf, 0.9) self.assertEqual(self.s.fuzzy_reads, 1) def test_unrelated_address_returns_none(self): alien = random_hv(D, random.Random(999)) self.assertIsNone(self.s.resolve(alien)) def test_empty_store_returns_none(self): self.assertIsNone(HDCStorage(d=D).resolve(random_hv(D, random.Random(3)))) class TestRecall(unittest.TestCase): def test_sorted_descending_and_self_first(self): s = HDCStorage(d=D) vecs = {} for i in range(6): vecs["c%d" % i] = random_hv(D, random.Random(200 + i)) s.write("c%d" % i, vecs["c%d" % i]) hits = s.recall(vecs["c3"], k=4) self.assertEqual(len(hits), 4) self.assertEqual(hits[0][0], "c3") self.assertEqual(hits[0][2], 1.0) sims = [h[2] for h in hits] self.assertEqual(sims, sorted(sims, reverse=True)) def test_prefix_scopes_search(self): s = HDCStorage(d=D) probe = random_hv(D, random.Random(5)) s.write("code/0", probe) s.write("data/0", random_hv(D, random.Random(6))) hits = s.recall(probe, k=5, prefix="data/") self.assertEqual([h[0] for h in hits], ["data/0"]) def test_floor_filters(self): s = HDCStorage(d=D) s.write("a", random_hv(D, random.Random(7))) self.assertEqual(s.recall(random_hv(D, random.Random(8)), k=5, floor=0.5), []) class TestPersistence(unittest.TestCase): def test_save_load_roundtrip(self): s = HDCStorage(d=D, capacity=64, label="unit") for i in range(5): s.write("cell%d" % i, random_hv(D, random.Random(300 + i)), symbol=True) s.write("code/0", 0) with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "unit.hdc") size = s.save(path) self.assertGreater(size, 0) loaded = HDCStorage.load(path) self.assertEqual(loaded.d, D) self.assertEqual(dict(loaded.cells), dict(s.cells)) self.assertEqual(loaded.symbols(), s.symbols()) self.assertEqual(loaded.order, s.order) self.assertEqual(loaded.read("cell3"), s.read("cell3")) def test_reject_bad_magic(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "junk.hdc") with open(path, "wb") as fh: fh.write(b"NOTHDC!!" + b"\x00" * 32) with self.assertRaises(ValueError): HDCStorage.load(path) class TestStat(unittest.TestCase): def test_keys_and_counts(self): s = HDCStorage(d=D, capacity=10) s.write("a", random_hv(D, random.Random(9)), symbol=True) s.read("a") st = s.stat() for key in ( "label", "d", "cells_used", "cells_free", "occupancy", "symbols", "reads", "writes", "stored_bits", "total_popcount", ): self.assertIn(key, st) self.assertEqual(st["cells_used"], 1) self.assertEqual(st["cells_free"], 9) self.assertEqual(st["writes"], 1) self.assertEqual(st["reads"], 1) self.assertEqual(st["stored_bits"], D) self.assertAlmostEqual(st["occupancy"], 0.1) def test_clone_is_independent(self): s = HDCStorage(d=D) s.write("a", 1, symbol=True) c = s.clone() c.write("b", 2) self.assertIn("a", c) self.assertNotIn("b", s) self.assertEqual(s.symbols(), ["a"]) if __name__ == "__main__": unittest.main()