Test ring buffer overwrite behavior
(self)
| 55 | # print(f"Received {len(received)} messages from ring buffer '{RING_NAME}'") |
| 56 | |
| 57 | def test_ring_buffer_overwrite(self): |
| 58 | """Test ring buffer overwrite behavior""" |
| 59 | ring_size = 16 |
| 60 | total_messages = 3 * ring_size |
| 61 | interval_usec = 100 # Fast, to fill quickly |
| 62 | ring_name = "/test/ring_buffer_overwrite" |
| 63 | |
| 64 | # Step 1: Generate more messages than the ring size |
| 65 | cli_cmd = f"test stats ring-buffer-gen {ring_name} {total_messages} {interval_usec} {ring_size}" |
| 66 | result = self.vapi.cli(cli_cmd) |
| 67 | assert "Generated" in result, f"CLI failed: {result}" |
| 68 | |
| 69 | # Ensure all messages are written before polling |
| 70 | time.sleep(0.1) |
| 71 | |
| 72 | # Step 2: Connect to stats segment and get the ring buffer |
| 73 | ring_buffer = self.statistics.get_ring_buffer(f"{ring_name}") |
| 74 | |
| 75 | # Step 3: Poll for all messages (should only get the last 'ring_size' messages) |
| 76 | received = {} |
| 77 | start = time.time() |
| 78 | while len(received) < ring_size and (time.time() - start) < 10: |
| 79 | data = ring_buffer.poll_for_data(thread_index=0, timeout=0.5) |
| 80 | for entry in data: |
| 81 | seq, ts = struct.unpack("<Qd", entry) |
| 82 | if seq not in received: |
| 83 | received[seq] = ts |
| 84 | time.sleep(0.01) |
| 85 | |
| 86 | # Debug: print the received sequence numbers |
| 87 | print(f"Received sequence numbers: {sorted(received.keys())}") |
| 88 | |
| 89 | # Step 4: Validate |
| 90 | assert ( |
| 91 | len(received) == ring_size |
| 92 | ), f"Expected {ring_size} messages, got {len(received)}" |
| 93 | expected_start = total_messages - ring_size |
| 94 | for i, seq in enumerate(sorted(received)): |
| 95 | ts = received[seq] |
| 96 | assert ( |
| 97 | seq == expected_start + i |
| 98 | ), f"Sequence mismatch at {i}: got {seq}, expected {expected_start + i}" |
| 99 | assert ts > 0, f"Timestamp should be positive, got {ts}" |
| 100 | |
| 101 | def test_ring_buffer_batch_operations(self): |
| 102 | """Test ring buffer batch operations""" |
nothing calls this directly
no test coverage detected