MCPcopy Create free account
hub / github.com/KentBeck/BPlusTree3 / run_fuzz_test

Method run_fuzz_test

python/tests/fuzz_test.py:318–385  ·  view source on GitHub ↗

Run the main fuzz test with specified number of operations

(self, num_operations: int = 1000000)

Source from the content-addressed store, hash-verified

316 return True
317
318 def run_fuzz_test(self, num_operations: int = 1000000) -> bool:
319 """Run the main fuzz test with specified number of operations"""
320 print(f"Starting fuzz test with {num_operations} operations (seed={self.seed})")
321 print(f"B+ tree capacity: {self.capacity}")
322 if self.prepopulate > 0:
323 print(f"Pre-populated with {self.prepopulate} elements")
324
325 start_time = time.time()
326
327 # Define operation weights
328 operations = [
329 (self.do_insert_or_update, 50), # 50% inserts/updates
330 (self.do_delete, 35), # 35% deletes
331 (self.do_get, 15), # 15% gets
332 # Note: batch_delete removed - not implemented yet
333 # (self.do_compact, 5), # 5% compactions - removed as no-op
334 ]
335
336 # Create weighted operation list
337 weighted_ops = []
338 for op_func, weight in operations:
339 weighted_ops.extend([op_func] * weight)
340
341 # Perform operations
342 for i in range(num_operations):
343 if i % 100000 == 0 and i > 0:
344 elapsed = time.time() - start_time
345 print(
346 f"Completed {i} operations in {elapsed:.1f}s (rate: {i/elapsed:.0f} ops/s)"
347 )
348 print(f" Current tree size: {len(self.btree)} keys")
349
350 # Verify consistency periodically
351 if not self.verify_consistency():
352 print(f"CONSISTENCY ERROR at operation {i}")
353 self._save_failure_info(i)
354 return False
355
356 # Choose and execute random operation
357 operation = random.choice(weighted_ops)
358 try:
359 if not operation():
360 print(f"OPERATION ERROR at operation {i}")
361 self._save_failure_info(i)
362 return False
363 except Exception as e:
364 print(f"EXCEPTION at operation {i}: {e}")
365 self._save_failure_info(i)
366 return False
367
368 # Final consistency check
369 if not self.verify_consistency():
370 print("FINAL CONSISTENCY CHECK FAILED")
371 self._save_failure_info(num_operations)
372 return False
373
374 elapsed = time.time() - start_time
375 print(f"\n✅ Fuzz test PASSED!")

Callers 7

run_capacity_sweepFunction · 0.95
run_stress_testFunction · 0.95
run_edge_case_testsFunction · 0.95
run_quick_fuzz_testFunction · 0.95
run_full_fuzz_testFunction · 0.95

Calls 4

verify_consistencyMethod · 0.95
_save_failure_infoMethod · 0.95
_count_total_nodesMethod · 0.45
itemsMethod · 0.45

Tested by

no test coverage detected