1#!/usr/bin/env python3 2# Copyright (c) PLUMgrid, Inc. 3# Licensed under the Apache License, Version 2.0 (the "License") 4 5import os 6import ctypes as ct 7 8from bcc import BPF 9 10from unittest import main, TestCase, skipUnless 11from utils import kernel_version_ge 12 13@skipUnless(kernel_version_ge(4,20), "requires kernel >= 4.20") 14class TestQueueStack(TestCase): 15 16 def test_stack(self): 17 text = b""" 18 BPF_STACK(stack, u64, 10); 19 """ 20 b = BPF(text=text) 21 stack = b[b'stack'] 22 23 for i in range(10): 24 stack.push(ct.c_uint64(i)) 25 26 with self.assertRaises(Exception): 27 stack.push(ct.c_uint(10)) 28 29 assert stack.peek().value == 9 30 31 for i in reversed(range(10)): 32 assert stack.pop().value == i 33 34 with self.assertRaises(KeyError): 35 stack.peek() 36 37 with self.assertRaises(KeyError): 38 stack.pop() 39 40 for i in reversed(range(10)): 41 stack.push(ct.c_uint64(i)) 42 43 # testing itervalues() 44 for i,v in enumerate(stack.values()): 45 assert v.value == i 46 47 b.cleanup() 48 49 def test_queue(self): 50 text = b""" 51 BPF_QUEUE(queue, u64, 10); 52 """ 53 b = BPF(text=text) 54 queue = b[b'queue'] 55 56 for i in range(10): 57 queue.push(ct.c_uint64(i)) 58 59 with self.assertRaises(Exception): 60 queue.push(ct.c_uint(10)) 61 62 assert queue.peek().value == 0 63 64 for i in range(10): 65 assert queue.pop().value == i 66 67 with self.assertRaises(KeyError): 68 queue.peek() 69 70 with self.assertRaises(KeyError): 71 queue.pop() 72 73 for i in range(10): 74 queue.push(ct.c_uint64(i)) 75 76 # testing itervalues() 77 for i,v in enumerate(queue.values()): 78 assert v.value == i 79 80 b.cleanup() 81 82 83if __name__ == "__main__": 84 main() 85