Run the main fuzz test with specified number of operations
(self, num_operations: int = 1000000)
| 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!") |
no test coverage detected