Method test_ring_buffer_error_handling

test/test_ring_buffer.py:180–209

Test ring buffer error handling

(self)

        assert isinstance(metadata["sequence"], int), "Sequence should be integer"

    def test_ring_buffer_error_handling(self):
        """Test ring buffer error handling"""
        ring_name = "/test/ring_buffer_error"

        # Step 1: Create ring buffer
        cli_cmd = (
            f"test stats ring-buffer-gen {ring_name} 10 {INTERVAL_USEC} {RING_SIZE}"
        )
        result = self.vapi.cli(cli_cmd)
        assert "Generated" in result, f"CLI failed: {result}"

        # Step 2: Get ring buffer
        ring_buffer = self.statistics.get_ring_buffer(f"{ring_name}")

        # Step 3: Test invalid thread index
        config = ring_buffer.get_config()
        invalid_thread = config["n_threads"] + 1

        try:
            ring_buffer._get_thread_metadata(invalid_thread)
            assert False, "Should have raised IndexError for invalid thread"
        except IndexError:
            pass  # Expected

        # Step 4: Test invalid parameters
        data = ring_buffer.consume_data(thread_index=0, max_entries=0)
        assert data == [], "Zero max_entries should return empty list"

        data = ring_buffer.consume_data(thread_index=0, max_entries=-1)
        assert data == [], "Negative max_entries should return empty list"

    def test_ring_buffer_api_compatibility(self):
        """Test ring buffer API compatibility methods"""