Test ring buffer batch operations
(self)
| 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""" |
| 103 | ring_name = "/test/ring_buffer_batch" |
| 104 | batch_size = 10 |
| 105 | total_messages = 50 |
| 106 | |
| 107 | # Step 1: Generate messages using VPP CLI |
| 108 | cli_cmd = f"test stats ring-buffer-gen {ring_name} {total_messages} {INTERVAL_USEC} {RING_SIZE}" |
| 109 | result = self.vapi.cli(cli_cmd) |
| 110 | assert "Generated" in result, f"CLI failed: {result}" |
| 111 | |
| 112 | # Step 2: Get ring buffer |
| 113 | ring_buffer = self.statistics.get_ring_buffer(f"{ring_name}") |
| 114 | |
| 115 | # Step 3: Test individual consume |
| 116 | individual_data = [] |
| 117 | start = time.time() |
| 118 | while len(individual_data) < total_messages and (time.time() - start) < 10: |
| 119 | data = ring_buffer.consume_data(thread_index=0, max_entries=batch_size) |
| 120 | individual_data.extend(data) |
| 121 | time.sleep(0.01) |
| 122 | |
| 123 | # Step 4: Reset and test batch consume |
| 124 | ring_buffer.local_tails[0] = 0 |
| 125 | ring_buffer.last_sequences[0] = None |
| 126 | |
| 127 | batch_data = [] |
| 128 | start = time.time() |
| 129 | while len(batch_data) < total_messages and (time.time() - start) < 10: |
| 130 | data = ring_buffer.consume_data_batch( |
| 131 | thread_index=0, max_entries=batch_size |
| 132 | ) |
| 133 | batch_data.extend(data) |
| 134 | time.sleep(0.01) |
| 135 | |
| 136 | # Step 5: Validate results are the same |
| 137 | assert len(individual_data) == len( |
| 138 | batch_data |
| 139 | ), "Individual and batch should return same number of entries" |
| 140 | assert ( |
| 141 | individual_data == batch_data |
| 142 | ), "Individual and batch should return same data" |
| 143 | |
| 144 | # Step 6: Validate data format |
| 145 | for entry in individual_data: |
| 146 | seq, ts = struct.unpack("<Qd", entry) |
| 147 | assert seq >= 0, f"Sequence should be non-negative, got {seq}" |
| 148 | assert ts > 0, f"Timestamp should be positive, got {ts}" |
| 149 | |
| 150 | def test_ring_buffer_configuration(self): |
| 151 | """Test ring buffer configuration access""" |
nothing calls this directly
no test coverage detected