Method test_ring_buffer_generation

test/test_ring_buffer.py:27–55

Test ring buffer generation

(self)

        super(TestRingBuffer, self).tearDown()

    def test_ring_buffer_generation(self):
        """Test ring buffer generation"""

        # Step 1: Generate messages using the VPPApiClient CLI
        cli_cmd = f"test stats ring-buffer-gen {RING_NAME} {COUNT} {INTERVAL_USEC} {RING_SIZE}"
        result = self.vapi.cli(cli_cmd)
        assert "Generated" in result, f"CLI failed: {result}"

        # Step 2: Connect to stats segment and get the ring buffer
        ring_buffer = self.statistics.get_ring_buffer(f"{RING_NAME}")

        # Step 3: Poll for all messages
        received = []
        start = time.time()
        while len(received) < COUNT and (time.time() - start) < 10:
            data = ring_buffer.poll_for_data(thread_index=0, timeout=0.5)
            for entry in data:
                # Unpack the struct: u64 seq, f64 timestamp
                seq, ts = struct.unpack("<Qd", entry)
                received.append((seq, ts))
            time.sleep(0.01)

        # Step 4: Validate
        assert len(received) == COUNT, f"Expected {COUNT} messages, got {len(received)}"
        for i, (seq, ts) in enumerate(received):
            assert seq == i, f"Sequence mismatch at {i}: got {seq}"
            assert ts > 0, f"Timestamp should be positive, got {ts}"

        # print(f"Received {len(received)} messages from ring buffer '{RING_NAME}'")

    def test_ring_buffer_overwrite(self):
        """Test ring buffer overwrite behavior"""