Test ring buffer API compatibility methods
(self)
| 209 | assert data == [], "Negative max_entries should return empty list" |
| 210 | |
| 211 | def test_ring_buffer_api_compatibility(self): |
| 212 | """Test ring buffer API compatibility methods""" |
| 213 | ring_name = "/test/ring_buffer_compat" |
| 214 | |
| 215 | # Step 1: Create ring buffer |
| 216 | cli_cmd = ( |
| 217 | f"test stats ring-buffer-gen {ring_name} 10 {INTERVAL_USEC} {RING_SIZE}" |
| 218 | ) |
| 219 | result = self.vapi.cli(cli_cmd) |
| 220 | assert "Generated" in result, f"CLI failed: {result}" |
| 221 | |
| 222 | # Step 2: Get ring buffer |
| 223 | ring_buffer = self.statistics.get_ring_buffer(f"{ring_name}") |
| 224 | |
| 225 | # Step 3: Test compatibility methods |
| 226 | count = ring_buffer.get_count(thread_index=0) |
| 227 | is_empty = ring_buffer.is_empty(thread_index=0) |
| 228 | is_full = ring_buffer.is_full(thread_index=0) |
| 229 | |
| 230 | # These methods return simplified values since writer doesn't track reader state |
| 231 | assert count == 0, "Count should be 0 (writer doesn't track reader)" |
| 232 | assert is_empty == True, "Is empty should be True (writer doesn't track reader)" |
| 233 | assert is_full == False, "Is full should be False (writer doesn't track reader)" |
| 234 | |
| 235 | # Step 4: Test string representation |
| 236 | repr_str = repr(ring_buffer) |
| 237 | assert ( |
| 238 | "StatsRingBuffer" in repr_str |
| 239 | ), "String representation should contain StatsRingBuffer" |
| 240 | assert ( |
| 241 | "entry_size" in repr_str |
| 242 | ), "String representation should contain entry_size" |
| 243 | assert "ring_size" in repr_str, "String representation should contain ring_size" |
| 244 | assert "n_threads" in repr_str, "String representation should contain n_threads" |
| 245 | |
| 246 | def test_ring_buffer_performance(self): |
| 247 | """Test ring buffer performance characteristics""" |
nothing calls this directly
no test coverage detected