Merge pull request #625 from F5Networks/ucs_support2

Author: wojtek0806

conftest.py

@@ -49,8 +49,49 @@ def pytest_addoption(parser):
 @pytest.fixture
 def fakeicontrolsession(monkeypatch):
     class Response(object):
+
         def json(self):
             return {'selfLink': 'https://localhost/mgmt/tm/sys?ver=11.6.0'}
+
+    fakesessionclass = mock.create_autospec(iControlRESTSession, spec_set=True)
+    fakesessioninstance = mock.create_autospec(iControlRESTSession('A', 'B'),
+                                               spec_set=True)
+    fakesessioninstance.get =\
+        mock.MagicMock(return_value=Response())
+    fakesessionclass.return_value = fakesessioninstance
+    monkeypatch.setattr('f5.bigip.iControlRESTSession', fakesessionclass)
+
+
+@pytest.fixture
+def fakeicontrolsessionfactory(monkeypatch):
+    class Response(object):
+        def __init__(self, **json_keys):
+            if 'selfLink' not in json_keys:
+                json_keys['selfLink'] =\
+                    'https://localhost/mgmt/tm/sys?ver=11.6.0'
+            self.params = json_keys
+
+        def json(self):
+            return self.params
+
+    def _session_factory(**json_keys):
+        fakesessionclass = mock.create_autospec(iControlRESTSession,
+                                                spec_set=True)
+        fakesessioninstance =\
+            mock.create_autospec(iControlRESTSession('A', 'B'), spec_set=True)
+        fakesessioninstance.get =\
+            mock.MagicMock(return_value=Response(**json_keys))
+        fakesessionclass.return_value = fakesessioninstance
+        monkeypatch.setattr('f5.bigip.iControlRESTSession', fakesessionclass)
+
+    return _session_factory
+
+
+@pytest.fixture
+def fakeicontrolsession_v12(monkeypatch):
+    class Response(object):
+        def json(self):
+            return {'selfLink': 'https://localhost/mgmt/tm/sys?ver=12.1.0'}
     fakesessionclass = mock.create_autospec(iControlRESTSession, spec_set=True)
     fakesessioninstance =\
         mock.create_autospec(iControlRESTSession('A', 'B'), spec_set=True)

f5/bigip/cm/autodeploy/test/test_software_image_uploads.py

@@ -24,7 +24,8 @@
 CHUNKSIZE = 20
 
 
-def test_software_image_uploads_80a(tmpdir, fakeicontrolsession):
+def test_software_image_uploads_80a(tmpdir, fakeicontrolsessionfactory):
+    fakeicontrolsessionfactory()
     filepath = tmpdir.mkdir('testdir').join('eightya.iso')
     filepath.write(80*'a')
     mr = ManagementRoot('FAKENETLOC', 'FAKENAME', 'FAKEPASSWORD')
@@ -37,7 +38,8 @@ def test_software_image_uploads_80a(tmpdir, fakeicontrolsession):
         assert d == 'a'*CHUNKSIZE
 
 
-def test_software_image_uploads_70a(tmpdir, fakeicontrolsession):
+def test_software_image_uploads_70a(tmpdir, fakeicontrolsessionfactory):
+    fakeicontrolsessionfactory()
     filepath = tmpdir.mkdir('testdir').join('seventya.iso')
     filepath.write(70*'a')
     mr = ManagementRoot('FAKENETLOC', 'FAKENAME', 'FAKEPASSWORD')
@@ -53,7 +55,8 @@ def test_software_image_uploads_70a(tmpdir, fakeicontrolsession):
     assert 10*'a' == lchunk
 
 
-def test_non_ISO_extension(tmpdir, fakeicontrolsession):
+def test_non_ISO_extension(tmpdir, fakeicontrolsessionfactory):
+    fakeicontrolsessionfactory()
     filepath = tmpdir.mkdir('testdir').join('wrong.name')
     mr = ManagementRoot('FAKENETLOC', 'FAKENAME', 'FAKEPASSWORD')
     sius = mr.cm.autodeploy.software_image_uploads

f5/bigip/mixins.py

@@ -126,6 +126,24 @@ def _check_supported_versions(self, container, attribute):
                         minimum)
             raise UnsupportedTmosVersion(error)
 
+    def _is_version_supported_method(container, method_version):
+        """Helper method
+
+         To use in instances where class methods on some resources
+         require a specific TMOS version to run.
+
+        Raises::
+                UnsupportedTmosVersion
+        """
+        tmos_v = container._meta_data['bigip'].tmos_version
+        if LooseVersion(tmos_v) < LooseVersion(method_version):
+            error = "There was an attempt to use a method which " \
+                    "has not been implemented or supported " \
+                    "in the device's TMOS version: %s. " \
+                    "Minimum TMOS version supported is %s" % (
+                        tmos_v, method_version)
+            raise UnsupportedTmosVersion(error)
+
 
 class ExclusiveAttributesMixin(object):
     """Overrides ``__setattr__`` to remove exclusive attrs from the object."""

f5/bigip/tm/sys/__init__.py

@@ -42,6 +42,7 @@
 from f5.bigip.tm.sys.performance import Performances
 from f5.bigip.tm.sys.software import Software
 from f5.bigip.tm.sys.sshd import Sshd
+from f5.bigip.tm.sys.ucs import Ucs
 
 
 class Sys(OrganizingCollection):
@@ -61,5 +62,6 @@ def __init__(self, tm):
             Dns,
             Sshd,
             Httpd,
-            Software
+            Software,
+            Ucs
         ]

f5/bigip/tm/sys/test/test_ucs.py

@@ -0,0 +1,68 @@
+# coding=utf-8
+#
+# Copyright 2016 F5 Networks Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+import mock
+import pytest
+
+from f5.bigip import ManagementRoot
+from f5.bigip.mixins import UnsupportedTmosVersion
+from f5.bigip.tm.sys.ucs import Ucs
+
+
+@pytest.fixture
+def FakeUcs():
+    fake_sys = mock.MagicMock()
+    fake_ucs = Ucs(fake_sys)
+    fake_ucs._meta_data['bigip'].tmos_version = '12.0.0'
+    return fake_ucs
+
+
+@pytest.fixture
+def FakeiControl(fakeicontrolsession_v12):
+    mr = ManagementRoot('FAKENETLOC', 'FAKENAME', 'FAKEPASSWORD')
+    mock_session = mock.MagicMock()
+    mock_session.post.return_value.json.return_value = {}
+    mr._meta_data['icr_session'] = mock_session
+    return mr.tm.sys.ucs
+
+
+class TestUCSCommand(object):
+    def test_command_ucs_load(self, FakeiControl):
+        FakeiControl.exec_cmd('load', name='foo.ucs')
+        session = FakeiControl._meta_data['bigip']._meta_data['icr_session']
+        assert session.post.call_args == mock.call(
+            'https://FAKENETLOC:443/mgmt/tm/sys/ucs/',
+            json={'name': 'foo.ucs', 'command': 'load'}
+        )
+
+    def test_command_ucs_save(self, FakeiControl):
+        FakeiControl.exec_cmd('save', name='foo.ucs')
+        session = FakeiControl._meta_data['bigip']._meta_data['icr_session']
+        assert session.post.call_args == mock.call(
+            'https://FAKENETLOC:443/mgmt/tm/sys/ucs/',
+            json={'name': 'foo.ucs', 'command': 'save'}
+        )
+
+    def test_list_ucs_wrong_tmos_version(self, FakeUcs):
+        with pytest.raises(UnsupportedTmosVersion) as EIO:
+            FakeUcs.load()
+            assert EIO.value.message == \
+                "There was an attempt to use a method which " \
+                "has not been implemented or supported " \
+                "in the device's TMOS version: 12.0.0. " \
+                "Minimum TMOS version supported is 12.1.0"

f5/bigip/tm/sys/ucs.py

@@ -0,0 +1,101 @@
+# coding=utf-8
+#
+# Copyright 2016 F5 Networks Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""BIG-IP® system config module
+
+REST URI
+    ``http://localhost/mgmt/tm/sys/ucs`
+
+GUI Path
+    N/A
+
+REST Kind
+    ``tm:sys:ucs:*``
+"""
+
+from f5.bigip.mixins import CommandExecutionMixin
+from f5.bigip.mixins import InvalidCommand
+from f5.bigip.resource import UnnamedResource
+from requests.exceptions import HTTPError
+
+
+class Ucs(UnnamedResource, CommandExecutionMixin):
+    """BIG-IP® system UCS resource
+
+        .. note::
+        Given the fact that 11.6.0 UCS via rest is
+        broken, this feature will be supported in 12.0.0 and above.
+        Listing of installed UCS has been fixed in 12.1.0.
+        This resource is a collection which does not allow listing
+        of each ucs as a resource. 'Items' attribute of the loaded object
+        is used to access the list of installed UCS files.
+
+        Caveat:
+        Loading UCS will result in ICRD restarting, therefore
+        due to ID476518 502 Bad Gateway is generated, this is
+        working as intended, at least until some architecture
+        changes have been made.
+
+
+    """
+    def __init__(self, sys):
+        super(Ucs, self).__init__(sys)
+        self._meta_data['required_load_parameters'] = set()
+        self._meta_data['allowed_commands'].extend(['load', 'save'])
+        self._meta_data['required_json_kind'] = ''
+        self._meta_data['minimum_version'] = '12.0.0'
+
+    def exec_cmd(self, command, **kwargs):
+        """Due to ID476518 the load command need special treatment"""
+        cmds = self._meta_data['allowed_commands']
+
+        if command not in self._meta_data['allowed_commands']:
+            error_message = "The command value {0} does not exist" \
+                            "Valid commands are {1}".format(command, cmds)
+            raise InvalidCommand(error_message)
+
+        if command == 'load':
+            kwargs['command'] = command
+            self._check_exclusive_parameters(**kwargs)
+            requests_params = self._handle_requests_params(kwargs)
+            self._check_command_parameters(**kwargs)
+            session = self._meta_data['bigip']._meta_data['icr_session']
+            try:
+
+                session.post(
+                    self._meta_data['uri'], json=kwargs, **requests_params)
+
+            except HTTPError as err:
+                if err.response.status_code != 502:
+                    raise
+                return
+
+        else:
+            return self._exec_cmd(command, **kwargs)
+
+    def load(self, **kwargs):
+        """Method to list the UCS on the system
+
+        Since this is only fixed in 12.1.0 and up
+        we implemented version check here
+        """
+
+        # Check if we are using 12.1.0 version or above when using this method
+        self._is_version_supported_method('12.1.0')
+        newinst = self._stamp_out_core()
+        newinst._refresh(**kwargs)
+
+        return newinst

test/functional/tm/sys/test_ucs.py

@@ -0,0 +1,41 @@
+# Copyright 2015-2016 F5 Networks Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from f5.bigip.mixins import LazyAttributesRequired
+import pytest
+import time
+
+
+@pytest.mark.skipif(pytest.config.getoption('--release') != '12.1.0',
+                    reason='Needs v12.1 TMOS to pass')
+class TestUcs(object):
+    def test_ucs_LR(self, bigip):
+        f = bigip.sys.ucs.load()
+
+        # Just in case our test unit does not have any ucs we create one
+        try:
+            f.items
+        except LazyAttributesRequired:
+            bigip.sys.ucs.exec_cmd('save', name='dummyucs.ucs')
+            time.sleep(1)
+            f.refresh()
+        finally:
+            ucs1 = len(f.items)
+        assert ucs1 >= 0
+        bigip.sys.ucs.exec_cmd('save', name='foobar.ucs')
+        time.sleep(1)
+        f.refresh()
+        ucs2 = len(f.items)
+        assert ucs2 > ucs1