Digital Drosophila Archive
124 lines
5.4 KiB
Python
124 lines
5.4 KiB
Python
import copy
|
|
import hashlib
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import seed_registry as m
|
|
|
|
|
|
class Seeds(unittest.TestCase):
|
|
def test_order_independent(self):
|
|
a, b = m.SeedRegistry(461), m.SeedRegistry(461)
|
|
keys = [("mode", "PAM15", str(replicate), str(epoch)) for replicate in range(5) for epoch in range(50)]
|
|
for key in keys:
|
|
a.add("training", key)
|
|
for key in reversed(keys):
|
|
b.add("training", key)
|
|
self.assertEqual(a.manifest(), b.manifest())
|
|
self.assertEqual(a.manifest()["independent_count"], 250)
|
|
self.assertTrue(m.validate_manifest(a.manifest()))
|
|
|
|
def test_splits_disjoint_and_no_sequential_overlap(self):
|
|
r = m.SeedRegistry(461)
|
|
for split in ("training", "evaluation", "design"):
|
|
for replicate in range(5):
|
|
for epoch in range(100):
|
|
r.add(split, ("shared-name", str(replicate), str(epoch)))
|
|
out = r.manifest()
|
|
seeds = [x["seed"] for x in out["independent_streams"]]
|
|
self.assertEqual(len(set(seeds)), 1500)
|
|
self.assertEqual(out["training_evaluation_seed_overlap"], 0)
|
|
self.assertTrue(all(0 <= s < 2**63 for s in seeds))
|
|
|
|
def test_alias_explicit_not_independent(self):
|
|
r = m.SeedRegistry(461)
|
|
seed = r.add("training", ("paired", "0"))
|
|
self.assertEqual(seed, r.alias("training", ("frozen", "0"), "training", ("paired", "0"), reason="Same prescheduled CS inputs, different learning policy"))
|
|
out = r.manifest()
|
|
self.assertEqual(out["independent_count"], 1)
|
|
self.assertEqual(out["alias_count"], 1)
|
|
self.assertFalse(out["aliases_count_as_independent_replicates"])
|
|
self.assertTrue(m.validate_manifest(out))
|
|
|
|
def test_no_cross_split_alias(self):
|
|
r = m.SeedRegistry(1)
|
|
r.add("training", ("a",))
|
|
with self.assertRaisesRegex(ValueError, "Cross-split"):
|
|
r.alias("evaluation", ("a",), "training", ("a",), reason="Not permitted")
|
|
|
|
def test_no_alias_chain(self):
|
|
r = m.SeedRegistry(1)
|
|
r.add("training", ("a",))
|
|
r.alias("training", ("b",), "training", ("a",), reason="Control")
|
|
with self.assertRaisesRegex(ValueError, "not another alias"):
|
|
r.alias("training", ("c",), "training", ("b",), reason="Invalid chain")
|
|
|
|
def test_no_silent_duplicate(self):
|
|
r = m.SeedRegistry(1)
|
|
r.add("training", ("a",))
|
|
with self.assertRaisesRegex(ValueError, "Duplicate"):
|
|
r.add("training", ("a",))
|
|
with self.assertRaisesRegex(ValueError, "Duplicate"):
|
|
r.alias("training", ("a",), "training", ("a",), reason="Invalid")
|
|
r.alias("training", ("b",), "training", ("a",), reason="Control")
|
|
with self.assertRaisesRegex(ValueError, "Duplicate"):
|
|
r.add("training", ("b",))
|
|
|
|
def test_collision_fails_without_resampling(self):
|
|
r = m.SeedRegistry(1)
|
|
with patch.object(m, "seed_for", return_value=(123, "a" * 64)):
|
|
r.add("training", ("a",))
|
|
with self.assertRaisesRegex(ValueError, "Seed collision"):
|
|
r.add("evaluation", ("b",))
|
|
self.assertEqual(len(r._streams), 1)
|
|
|
|
def test_mutations_detected(self):
|
|
r = m.SeedRegistry(1)
|
|
r.add("training", ("a",))
|
|
original = r.manifest()
|
|
for mutate in (lambda x: x["independent_streams"][0].update(seed=7),
|
|
lambda x: x.update(training_evaluation_seed_overlap=1),
|
|
lambda x: x.update(independent_count=2),
|
|
lambda x: x.update(payload_sha256="a" * 64),
|
|
lambda x: x.update(training_executed=True)):
|
|
corrupt = copy.deepcopy(original)
|
|
mutate(corrupt)
|
|
with self.assertRaises(ValueError):
|
|
m.validate_manifest(corrupt)
|
|
|
|
def test_manifest_copy_not_mutable_alias(self):
|
|
r = m.SeedRegistry(1)
|
|
r.add("training", ("a",))
|
|
out = r.manifest()
|
|
out["independent_streams"][0]["key"][0] = "changed"
|
|
self.assertEqual(r.manifest()["independent_streams"][0]["key"], ["a"])
|
|
self.assertTrue(m.validate_manifest(json.loads(json.dumps(r.manifest()))))
|
|
|
|
def test_json_numeric_boolean_type_changes_rejected_even_when_rehashed(self):
|
|
r = m.SeedRegistry(1)
|
|
r.add("training", ("a",))
|
|
for field, replacement in (("training_executed", 0), ("independent_count", 1.0)):
|
|
for rehash in (False, True):
|
|
corrupt = r.manifest()
|
|
corrupt[field] = replacement
|
|
if rehash:
|
|
payload = {key: value for key, value in corrupt.items() if key != "payload_sha256"}
|
|
corrupt["payload_sha256"] = hashlib.sha256(m._canonical(payload).encode("utf-8")).hexdigest()
|
|
with self.subTest(field=field, rehash=rehash), self.assertRaises(ValueError):
|
|
m.validate_manifest(corrupt)
|
|
|
|
def test_reject_invalid_values(self):
|
|
for master in (True, -1, 2**63, 1.0, "1"):
|
|
with self.assertRaises(ValueError):
|
|
m.SeedRegistry(master)
|
|
for key in ([], "a", [""], [" a"], ["a "]):
|
|
with self.assertRaises(ValueError):
|
|
m.seed_for(1, "training", key)
|
|
with self.assertRaises(ValueError):
|
|
m.seed_for(1, "unknown", ["a"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|