"""Deterministic teaching experiments, not production implementations. Run with Python 3.10+: python3 engineering_models.py No services, credentials, network calls, or third-party packages are needed. """ from dataclasses import dataclass from hashlib import sha256 from random import Random from pathlib import Path import re import unittest SCALE = 1_000_000 @dataclass class Bucket: capacity: int rate: int tokens: int last_us: int def take(self, now_us: int, cost: int = 1) -> bool: if self.capacity <= 0 or self.rate <= 0 or cost <= 0: raise ValueError("positive capacity, rate, and cost required") now_us = max(now_us, self.last_us) self.tokens = min(self.capacity * SCALE, self.tokens + (now_us - self.last_us) * self.rate) self.last_us = now_us debit = cost * SCALE if self.tokens < debit: return False self.tokens -= debit return True @dataclass class FencedResource: generation: int = 0 value: str = "initial" def write(self, generation: int, value: str) -> bool: # One indivisible transition in this sequential model. if generation < self.generation: return False self.generation, self.value = generation, value return True def visible(versions, snapshot): """Versions are (sequence, value); None represents a tombstone.""" eligible = [(seq, value) for seq, value in versions if seq <= snapshot] return max(eligible, default=(0, None), key=lambda pair: pair[0])[1] def leading_zero_bits(digest: bytes, bits: int) -> bool: if not 0 <= bits <= len(digest) * 8: raise ValueError("difficulty outside digest width") whole, partial = divmod(bits, 8) return (all(byte == 0 for byte in digest[:whole]) and (partial == 0 or digest[whole] >> (8 - partial) == 0)) class Experiments(unittest.TestCase): def test_window_boundary_is_not_a_rolling_limit(self): times = [990_000] * 10 + [1_010_000] * 10 windows = {} for time in times: window = time // SCALE windows[window] = windows.get(window, 0) + 1 self.assertEqual(max(windows.values()), 10) self.assertEqual(len(times), 20) self.assertLess(times[-1] - times[0], SCALE) def test_fractional_refill_survives_denial(self): bucket = Bucket(4, 2, 4 * SCALE, 0) self.assertEqual([bucket.take(0) for _ in range(5)], [True, True, True, True, False]) self.assertFalse(bucket.take(250_000)) self.assertTrue(bucket.take(500_000)) self.assertEqual(bucket.tokens, 0) def test_weighted_request_and_backward_time(self): bucket = Bucket(20, 10, 20 * SCALE, 0) self.assertTrue(bucket.take(0, 15)) self.assertFalse(bucket.take(999_999, 15)) self.assertTrue(bucket.take(1_000_000, 15)) self.assertFalse(bucket.take(0)) self.assertFalse(bucket.take(2_000_000, 21)) def test_admission_envelope_over_many_intervals(self): rng = Random(17) bucket = Bucket(7, 3, 7 * SCALE, 0) now = 0 admitted = [] for _ in range(250): now += rng.randrange(0, 500_000) cost = rng.randrange(1, 5) if bucket.take(now, cost): admitted.append((now, cost)) for start in range(len(admitted)): total = 0 for end in range(start, len(admitted)): total += admitted[end][1] elapsed = admitted[end][0] - admitted[start][0] self.assertLessEqual(total * SCALE, 7 * SCALE + 3 * elapsed) def test_stale_worker_is_rejected_but_equal_generation_is_not_dedup(self): resource = FencedResource() self.assertTrue(resource.write(7, "A")) self.assertTrue(resource.write(8, "B")) self.assertFalse(resource.write(7, "stale A")) self.assertEqual(resource.value, "B") self.assertTrue(resource.write(8, "second B operation")) def test_snapshot_tombstone(self): versions = [(2, "bob"), (6, None), (12, "bea")] self.assertEqual(visible(versions, 5), "bob") self.assertIsNone(visible(versions, 8)) self.assertEqual(visible(versions, 12), "bea") self.assertEqual(visible([(2, "bob")], 8), "bob") def test_mean_of_means_loses_weight(self): groups = [(1, 100), (99, 0)] # count and mean wrong = sum(mean for _, mean in groups) / len(groups) correct = sum(count * mean for count, mean in groups) / sum(c for c, _ in groups) self.assertEqual(wrong, 50) self.assertEqual(correct, 1) def test_insert_sequence_is_not_commit_order(self): # Transaction 1 allocated 41 but is still open; 42 committed first. committed = {42} cursor = max(committed) committed.add(41) self.assertNotIn(41, [x for x in committed if x > cursor]) def test_partial_bit_pow_check(self): self.assertTrue(leading_zero_bits(bytes.fromhex('00003f'), 18)) self.assertFalse(leading_zero_bits(bytes.fromhex('000040'), 18)) self.assertTrue(leading_zero_bits(bytes.fromhex('000040'), 16)) self.assertTrue(leading_zero_bits(b'\x00', 8)) self.assertTrue(leading_zero_bits(bytes.fromhex('00000f'), 20)) self.assertFalse(leading_zero_bits(bytes.fromhex('000010'), 20)) with self.assertRaises(ValueError): leading_zero_bits(b'\x00', 9) def test_runnable_lesson_code(self): # Exercise the code readers copy, as well as the independent models above. lessons = Path(__file__).resolve().parents[1] / 'lessons' snippets = {} for number in ['08', '12', '37']: lesson = next(lessons.glob(number + '-*.md')) code = re.search(r'```python\n(.*?)\n```', lesson.read_text(), re.S).group(1) namespace = {} exec(compile(code, str(lesson), 'exec'), namespace) snippets[number] = namespace bucket = snippets['08']['Bucket'](4, 2, 0, 0) self.assertFalse(bucket.take(250_000)) self.assertTrue(bucket.take(500_000)) owner = snippets['12']['owner'] self.assertEqual(owner(30, [10, 30], ['A', 'B']), 'B') self.assertEqual(owner(31, [10, 30], ['A', 'B']), 'A') with self.assertRaises(ValueError): owner(1, [], []) check = snippets['37']['leading_zero_bits'] self.assertTrue(check(bytes.fromhex('00003f'), 18)) self.assertFalse(check(bytes.fromhex('000040'), 18)) self.assertTrue(check(bytes.fromhex('00000f'), 20)) self.assertFalse(check(bytes.fromhex('000010'), 20)) def test_small_pow_and_challenge_spending(self): challenge = b'teaching-only|signup|generation-1' nonce = 0 while not leading_zero_bits(sha256(challenge + b'|' + str(nonce).encode()).digest(), 8): nonce += 1 spent = set() self.assertNotIn(challenge, spent) spent.add(challenge) self.assertIn(challenge, spent) # any second nonce must still be refused if __name__ == '__main__': unittest.main(verbosity=2)