[SEBA-412] Automated reformat of Python code
Passes of modernize, autopep8, black, then check with flake8
flake8 + manual fixes:
lib/xos-config
lib/xos-kafka
lib/xos-util
xos/coreapi
xos/api
xos/xos_client
Change-Id: Ib23cf84cb13beb3c6381fa0d79594dc9131dc815
diff --git a/lib/xos-genx/xos-genx-tests/test_xos_security.py b/lib/xos-genx/xos-genx-tests/test_xos_security.py
index 766e102..3bd4653 100644
--- a/lib/xos-genx/xos-genx-tests/test_xos_security.py
+++ b/lib/xos-genx/xos-genx-tests/test_xos_security.py
@@ -1,4 +1,3 @@
-
# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
@@ -21,25 +20,32 @@
"""The function below is for eliminating warnings arising due to the missing policy_output_enforcer,
which is generated and loaded dynamically.
"""
+
+
def policy_output_enforcer(x, y):
raise Exception("Security enforcer not generated. Test failed.")
return False
+
"""
-The tests below use the Python code target to generate
+The tests below use the Python code target to generate
Python security policies, set up an appropriate environment and execute the Python.
The security policies here deliberately made complex in order to stress the processor.
"""
+
+
class XProtoXOSSecurityTest(unittest.TestCase):
def setUp(self):
- self.target = XProtoTestHelpers.write_tmp_target("{{ xproto_fol_to_python_test('output',proto.policies.test_policy, None, '0') }}")
+ self.target = XProtoTestHelpers.write_tmp_target(
+ "{{ xproto_fol_to_python_test('output',proto.policies.test_policy, None, '0') }}"
+ )
"""
This is the security policy for controllers
"""
+
def test_controller_policy(self):
- xproto = \
-"""
+ xproto = """
policy test_policy < ctx.user.is_admin | exists Privilege: Privilege.accessor_id = ctx.user.id & Privilege.object_type = "Deployment" & Privilege.permission = "role:admin" & Privilege.object_id = obj.id >
"""
args = XOSProcessorArgs()
@@ -48,7 +54,7 @@
output = XOSProcessor.process(args)
- exec(output) # This loads the generated function, which should look like this:
+ exec(output) # This loads the generated function, which should look like this:
"""
def policy_output_enforcer(obj, ctx):
@@ -64,10 +70,10 @@
"""
This is the security policy for ControllerNetworks
"""
+
def test_controller_network_policy(self):
- xproto = \
-"""
- policy test_policy <
+ xproto = """
+ policy test_policy <
ctx.user.is_admin
| (exists Privilege:
Privilege.accessor_id = ctx.user.id
@@ -86,7 +92,7 @@
args.target = self.target
output = XOSProcessor.process(args)
- exec(output) # This loads the generated function, which should look like this:
+ exec(output) # This loads the generated function, which should look like this:
"""
def policy_output_enforcer(obj, ctx):
@@ -104,9 +110,9 @@
"""
This is the security policy for Slices
"""
+
def test_slice_policy(self):
- xproto = \
-"""
+ xproto = """
policy site_policy <
ctx.user.is_admin
| (ctx.write_access -> exists Privilege: Privilege.object_type = "Site" & Privilege.object_id = obj.id & Privilege.accessor_id = ctx.user.id & Privilege.permission_id = "role:admin") >
@@ -127,7 +133,7 @@
& Privilege.object_id = obj.site.id
& Privilege.permission = "role:admin"))
)>
-
+
"""
args = XOSProcessorArgs()
args.inputs = xproto
@@ -135,25 +141,25 @@
output = XOSProcessor.process(args)
- exec(output) # This loads the generated function, which should look like this:
+ exec(output) # This loads the generated function, which should look like this:
"""
def policy_output_enforcer(obj, ctx):
- i2 = ctx.user.is_admin
- i4 = policy_site_policy_enforcer(obj.site, ctx)
- i10 = ctx.write_access
- i11 = (not (not Privilege.objects.filter(Q(accessor_id=ctx.user.id), Q(accessor_type='User'), Q(object_type='Slice'), Q(object_id=obj.id), Q(permission='role:admin'))))
- i8 = (i10 and i11)
- i14 = ctx.write_access
- i12 = (not i14)
- i13 = (not (not Privilege.objects.filter(Q(accessor_id=ctx.user.id), Q(accessor_type='User'), Q(object_type='Slice'), Q(object_id=obj.id))))
- i9 = (i12 and i13)
- i6 = (i8 or i9)
- i7 = (not (not Privilege.objects.filter(Q(accessor_id=ctx.user.id), Q(accessor_type='User'), Q(object_type='Site'), Q(object_id=obj.site.id), Q(permission='role:admin'))))
- i5 = (i6 or i7)
- i3 = (i4 and i5)
- i1 = (i2 or i3)
- return i1
+ i2 = ctx.user.is_admin
+ i4 = policy_site_policy_enforcer(obj.site, ctx)
+ i10 = ctx.write_access
+ i11 = (not (not Privilege.objects.filter(Q(accessor_id=ctx.user.id), Q(accessor_type='User'), Q(object_type='Slice'), Q(object_id=obj.id), Q(permission='role:admin'))))
+ i8 = (i10 and i11)
+ i14 = ctx.write_access
+ i12 = (not i14)
+ i13 = (not (not Privilege.objects.filter(Q(accessor_id=ctx.user.id), Q(accessor_type='User'), Q(object_type='Slice'), Q(object_id=obj.id))))
+ i9 = (i12 and i13)
+ i6 = (i8 or i9)
+ i7 = (not (not Privilege.objects.filter(Q(accessor_id=ctx.user.id), Q(accessor_type='User'), Q(object_type='Site'), Q(object_id=obj.site.id), Q(permission='role:admin'))))
+ i5 = (i6 or i7)
+ i3 = (i4 and i5)
+ i1 = (i2 or i3)
+ return i1
"""
# FIXME: Test this policy by executing it
@@ -162,13 +168,13 @@
"""
This is the security policy for Users
"""
+
def test_user_policy(self):
- xproto = \
-"""
+ xproto = """
policy test_policy <
ctx.user.is_admin
| ctx.user.id = obj.id
- | (exists Privilege:
+ | (exists Privilege:
Privilege.accessor_id = ctx.user.id
& Privilege.accessor_type = "User"
& Privilege.permission = "role:admin"
@@ -181,7 +187,7 @@
output = XOSProcessor.process(args)
- exec(output) # This loads the generated function, which should look like this:
+ exec(output) # This loads the generated function, which should look like this:
"""
def policy_output_enforcer(obj, ctx):
@@ -196,5 +202,6 @@
# FIXME: Test this policy by executing it
self.assertTrue(policy_output_enforcer is not None)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
unittest.main()