Merge pull request #1316 from caphrim007/bugfix.allow-asm-tasks-to-import-to-partition

Adds the ability for ASM tasks to import to different partititions

Author: f5-rahm

f5/bigip/tm/asm/tasks.py

@@ -30,7 +30,10 @@
 from f5.bigip.resource import AsmTaskResource
 from f5.bigip.resource import Collection
 from f5.bigip.resource import OrganizingCollection
+from f5.sdk_exception import F5SDKError
+from f5.sdk_exception import RequiredOneOf
 from f5.sdk_exception import UnsupportedOperation
+from six import iterkeys
 
 
 class Tasks(OrganizingCollection):
@@ -125,12 +128,31 @@ class Import_Policy(AsmResource):
     def __init__(self, import_policy_s):
         super(Import_Policy, self).__init__(import_policy_s)
         self._meta_data['required_json_kind'] = 'tm:asm:tasks:import-policy:import-policy-taskstate'
-        self._meta_data['required_creation_parameters'] = {
-            'name'
-        }
+        self._meta_data['required_creation_parameters'] = set()
         self._meta_data['exclusive_attributes'] = ['filename', 'file']
         self._meta_data['minimum_additional_parameters'] = {'filename', 'file', 'policyTemplateReference'}
 
+    def create(self, **kwargs):
+        required_one_of = ['name', 'fullPath']
+        has_any = [x for x in iterkeys(kwargs) if x in required_one_of]
+
+        if 'partition' in kwargs and 'name' in kwargs and 'fullPath' not in kwargs:
+            partition = kwargs.pop('partition', None)
+            name = kwargs.pop('name', None)
+            if partition is not None and not partition.startswith('/'):
+                kwargs['fullPath'] = '/{0}/{1}'.format(partition, name)
+            elif partition is None:
+                raise F5SDKError(
+                    "The partition name, if specified, cannot be none"
+                )
+            elif partition.startswith('/'):
+                kwargs['fullPath'] = '{0}/{1}'.format(partition, name)
+
+        if len(has_any) == 1:
+            return self._create(**kwargs)
+
+        raise RequiredOneOf(required_one_of)
+
     def modify(self, **kwargs):
         """Modify is not supported for Apply Policy resource
 

f5/bigip/tm/asm/test/functional/test_tasks.py

@@ -55,6 +55,26 @@ def file_read():
     return result
 
 
+def remove_policies(mgmt_root, policy=None):
+    policies = mgmt_root.tm.asm.policies_s.get_collection()
+    if policy is None:
+        resources = policies
+    else:
+        resources = [p for p in policies if p.name == policy]
+    for resource in resources:
+        resource.delete()
+
+
+@pytest.fixture(scope='function')
+def partition(mgmt_root):
+    file = tempfile.NamedTemporaryFile()
+    name = os.path.basename(file.name)
+    partitions = mgmt_root.tm.auth.partitions.partition
+    local = partitions.create(name=name)
+    yield local
+    local.delete()
+
+
 @pytest.fixture(scope='function')
 def check_sig(mgmt_root):
     task = mgmt_root.tm.asm.tasks.check_signatures_s.check_signature.fetch()
@@ -221,6 +241,43 @@ def import_policy(mgmt_root):
     task.delete()
 
 
+@pytest.fixture(scope='function')
+def import_partitioned_policy(mgmt_root, partition):
+    content = file_read()
+    file = tempfile.NamedTemporaryFile()
+    name = os.path.basename(file.name)
+    task = mgmt_root.tm.asm.tasks.import_policy_s.import_policy.create(
+        file=content,
+        fullPath='/{0}/{1}'.format(partition.name, name)
+    )
+    while True:
+        task.refresh()
+        if task.status in ['COMPLETED', 'FAILURE']:
+            break
+        time.sleep(1)
+    yield task
+    task.delete()
+
+
+@pytest.fixture(scope='function')
+def import_partitioned_policy2(mgmt_root, partition):
+    content = file_read()
+    file = tempfile.NamedTemporaryFile()
+    name = os.path.basename(file.name)
+    task = mgmt_root.tm.asm.tasks.import_policy_s.import_policy.create(
+        file=content,
+        name=name,
+        partition=partition.name
+    )
+    while True:
+        task.refresh()
+        if task.status in ['COMPLETED', 'FAILURE']:
+            break
+        time.sleep(1)
+    yield task
+    task.delete()
+
+
 @pytest.fixture(scope='function')
 def import_vuln(mgmt_root, set_policy):
     reference = {'link': set_policy}
@@ -423,6 +480,26 @@ def test_create_import_template(self, import_policy_template):
         assert imp1.kind == 'tm:asm:tasks:import-policy:import-policy-taskstate'
         assert imp1.isBase64 is False
 
+    def test_create_import_partitioned(self, mgmt_root, import_partitioned_policy):
+        imp1 = import_partitioned_policy
+        endpoint = str(imp1.id)
+        base_uri = 'https://localhost/mgmt/tm/asm/tasks/import-policy/'
+        final_uri = base_uri + endpoint
+        assert imp1.selfLink.startswith(final_uri)
+        assert imp1.kind == 'tm:asm:tasks:import-policy:import-policy-taskstate'
+        assert imp1.isBase64 is False
+        remove_policies(mgmt_root, os.path.basename(imp1.fullPath))
+
+    def test_create_import_partitioned2(self, mgmt_root, import_partitioned_policy2):
+        imp1 = import_partitioned_policy2
+        endpoint = str(imp1.id)
+        base_uri = 'https://localhost/mgmt/tm/asm/tasks/import-policy/'
+        final_uri = base_uri + endpoint
+        assert imp1.selfLink.startswith(final_uri)
+        assert imp1.kind == 'tm:asm:tasks:import-policy:import-policy-taskstate'
+        assert imp1.isBase64 is False
+        remove_policies(mgmt_root, os.path.basename(imp1.fullPath))
+
     def test_create_import_fails(self, import_policy_base64):
         imp1 = import_policy_base64
         endpoint = str(imp1.id)

f5/bigip/tm/asm/test/unit/test_tasks.py

@@ -23,6 +23,7 @@
 from f5.bigip.tm.asm.tasks import Import_Vulnerabilities
 from f5.bigip.tm.asm.tasks import Update_Signature
 from f5.sdk_exception import MissingRequiredCreationParameter
+from f5.sdk_exception import RequiredOneOf
 from f5.sdk_exception import UnsupportedOperation
 from f5.sdk_exception import UnsupportedTmosVersion
 
@@ -249,7 +250,7 @@ def test_create_two(self, fakeicontrolsession):
         assert t1 is t2
 
     def test_create_no_args(self, FakeImportPolicy):
-        with pytest.raises(MissingRequiredCreationParameter):
+        with pytest.raises(RequiredOneOf):
             FakeImportPolicy.create()
 
     def test_collection(self, fakeicontrolsession):