cni: _has_cap unit tests system independant

Originally the test was designed so that it would use the system CapBnd.
This was good because it would alert us soon if the procfs status
formatting changed. This however presents issues with severely
constrained environments like the one in the bug description.

Change-Id: Ie95d6cb35d9bee5c62e09a272001c68bb684ee73
Closes-Bug: 1753751
Signed-off-by: Antoni Segura Puimedon <antonisp@celebdor.com>
This commit is contained in:
Antoni Segura Puimedon 2018-03-06 15:04:02 +01:00
parent c16815d544
commit 7f77dc2380
No known key found for this signature in database
GPG Key ID: 9B08FFD846853B9D
2 changed files with 22 additions and 3 deletions

View File

@ -44,8 +44,17 @@ CAP_NET_ADMIN = 12 # Taken from linux/capabilities.h
EFFECTIVE_CAPS = 'CapEff:\t'
def _has_cap(capability, entry):
with open('/proc/self/status', 'r') as pstat:
def _has_cap(capability, entry, proc_status_path='/proc/self/status'):
"""Returns true iff the process has the specified capability.
:param capability: the bit number for the capability to check as seen
in linux/capabilities.h.
:param entry: Whether to check CapInh, CapEff or CapBnd.
:param proc_status_path: Which process status should be checked. If none
is passed, it will check the current process.
:return: Whether the specified process has the capability bit set
"""
with open(proc_status_path, 'r') as pstat:
for line in pstat:
if line.startswith(entry):
caps = int(line[len(entry):], 16)

View File

@ -15,6 +15,7 @@ from kuryr_kubernetes.cni import health
from kuryr_kubernetes.tests import base
import mock
import multiprocessing
import tempfile
from oslo_config import cfg
@ -91,4 +92,13 @@ class TestCNIHealthServer(base.TestCase):
class TestCNIHealthUtils(base.TestCase):
def test_has_cap(self):
self.assertTrue(health._has_cap(health.CAP_NET_ADMIN, 'CapBnd:\t'))
with tempfile.NamedTemporaryFile() as fake_status:
fake_status.write(b'CapInh:\t0000000000000000\n'
b'CapPrm:\t0000000000000000\n'
b'CapEff:\t0000000000000000\n'
b'CapBnd:\t0000003fffffffff\n')
fake_status.flush()
self.assertTrue(
health._has_cap(health.CAP_NET_ADMIN,
'CapBnd:\t',
fake_status.name))