diff --git a/conftest.py b/conftest.py index b168259ba..9d5e528f7 100644 --- a/conftest.py +++ b/conftest.py @@ -16,10 +16,13 @@ from f5.bigip import BigIP from f5.bigip import ManagementRoot from f5.utils.testutils.registrytools import register_device +from icontrol.session import iControlRESTSession import logging import mock +import os import pytest import requests +from tempfile import NamedTemporaryFile logger = logging.getLogger() @@ -27,8 +30,6 @@ requests.packages.urllib3.disable_warnings() -from icontrol.session import iControlRESTSession - def pytest_addoption(parser): parser.addoption("--bigip", action="store", @@ -219,3 +220,16 @@ def teardown(): after_snapshot[item].delete() request.addfinalizer(teardown) return before_snapshot + + +@pytest.fixture +def IFILE(mgmt_root): + ntf = NamedTemporaryFile() + ntf_basename = os.path.basename(ntf.name) + ntf.write('this is a test file') + ntf.seek(0) + mgmt_root.shared.file_transfer.uploads.upload_file(ntf.name) + tpath_name = 'file:/var/config/rest/downloads/{0}'.format(ntf_basename) + i = mgmt_root.tm.sys.file.ifiles.ifile.create(name=ntf_basename, + sourcePath=tpath_name) + return i diff --git a/f5/bigip/__init__.py b/f5/bigip/__init__.py index 607e13121..b244cbc30 100644 --- a/f5/bigip/__init__.py +++ b/f5/bigip/__init__.py @@ -71,6 +71,10 @@ def hostname(self): def icontrol_version(self): return self._meta_data['icontrol_version'] + @property + def tmos_version(self): + return self._meta_data['tmos_version'] + def _get_tmos_version(self): connect = self._meta_data['bigip']._meta_data['icr_session'] base_uri = self._meta_data['uri'] + 'tm/sys/' diff --git a/f5/bigip/mixins.py b/f5/bigip/mixins.py index 6216156b5..0f04ec85a 100644 --- a/f5/bigip/mixins.py +++ b/f5/bigip/mixins.py @@ -114,7 +114,7 @@ def __getattr__(container, name): return attribute def _check_supported_versions(self, container, attribute): - tmos_v = container._meta_data['bigip']._meta_data['tmos_version'] + tmos_v = container._meta_data['bigip'].tmos_version minimum = attribute._meta_data['minimum_version'] if LooseVersion(tmos_v) < LooseVersion(minimum): error = "There was an attempt to access resource: \n{}\n which " \ diff --git a/f5/bigip/test/test_resource.py b/f5/bigip/test/test_resource.py index 72da53610..ece08c6eb 100644 --- a/f5/bigip/test/test_resource.py +++ b/f5/bigip/test/test_resource.py @@ -797,6 +797,8 @@ def test_load(self): }) ]} mock_session = mock.MagicMock(**attrs) + ver_mock = mock.PropertyMock(return_value='11.5.0') + type(r._meta_data['bigip']).tmos_version = ver_mock r._meta_data['bigip']._meta_data =\ {'icr_session': mock_session, 'hostname': 'TESTDOMAINNAME', diff --git a/f5/bigip/tm/ltm/__init__.py b/f5/bigip/tm/ltm/__init__.py index ef21221df..9c8786990 100644 --- a/f5/bigip/tm/ltm/__init__.py +++ b/f5/bigip/tm/ltm/__init__.py @@ -30,6 +30,7 @@ from f5.bigip.resource import OrganizingCollection from f5.bigip.tm.ltm.data_group import Data_Group +from f5.bigip.tm.ltm.ifile import Ifiles from f5.bigip.tm.ltm.monitor import Monitor from f5.bigip.tm.ltm.nat import Nats from f5.bigip.tm.ltm.node import Nodes @@ -51,6 +52,7 @@ def __init__(self, tm): super(Ltm, self).__init__(tm) self._meta_data['allowed_lazy_attributes'] = [ Data_Group, + Ifiles, Monitor, Nats, Nodes, diff --git a/f5/bigip/tm/ltm/ifile.py b/f5/bigip/tm/ltm/ifile.py new file mode 100644 index 000000000..1124759d4 --- /dev/null +++ b/f5/bigip/tm/ltm/ifile.py @@ -0,0 +1,47 @@ +# coding=utf-8 +# +# Copyright 2014-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Ā® LTM ifile submodule. + +REST URI + ``http://localhost/mgmt/tm/ltm/ifile/`` + +GUI Path + ``Local Traffic --> iRules --> iFiles`` + +REST Kind + ``tm:ltm:ifile*`` +""" + +from f5.bigip.resource import Collection +from f5.bigip.resource import Resource + + +class Ifiles(Collection): + def __init__(self, ltm): + super(Ifiles, self).__init__(ltm) + self._meta_data['allowed_lazy_attributes'] = [Ifile] + self._meta_data['attribute_registry'] =\ + {u'tm:ltm:ifile:ifilestate': Ifile} + + +class Ifile(Resource): + def __init__(self, ifile_s): + super(Ifile, self).__init__(ifile_s) + self._meta_data['required_json_kind'] = u'tm:ltm:ifile:ifilestate' + self._meta_data['required_creation_parameters'].update( + ('name', 'fileName') + ) diff --git a/f5/bigip/tm/ltm/test/test_ifile.py b/f5/bigip/tm/ltm/test/test_ifile.py new file mode 100644 index 000000000..84e8ac6d2 --- /dev/null +++ b/f5/bigip/tm/ltm/test/test_ifile.py @@ -0,0 +1,40 @@ +# Copyright 2015 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.resource import MissingRequiredCreationParameter +from f5.bigip.tm.ltm.ifile import Ifile + + +@pytest.fixture +def FakeIfile(): + fake_ifile_s = mock.MagicMock() + fake_ifile = Ifile(fake_ifile_s) + return fake_ifile + + +class TestCreate(object): + def test_create_two(self, fakeicontrolsession): + mgmt = ManagementRoot('172.16.44.15', 'admin', 'admin') + r1 = mgmt.tm.ltm.ifiles.ifile + r2 = mgmt.tm.ltm.ifiles.ifile + assert r1 is not r2 + + def test_create_no_args(self, FakeIfile): + with pytest.raises(MissingRequiredCreationParameter): + FakeIfile.create() diff --git a/f5/bigip/tm/sys/file.py b/f5/bigip/tm/sys/file.py index 20353e9f4..df9a5b4b0 100644 --- a/f5/bigip/tm/sys/file.py +++ b/f5/bigip/tm/sys/file.py @@ -25,6 +25,7 @@ REST Kind ``tm:sys:file:*`` """ + from f5.bigip.resource import Collection from f5.bigip.resource import OrganizingCollection from f5.bigip.resource import Resource @@ -34,19 +35,20 @@ class File(OrganizingCollection): def __init__(self, sys): super(File, self).__init__(sys) self._meta_data['allowed_lazy_attributes'] = [ - Data_Groups - ] + Data_Groups, + Ifiles, + Ssl_Certs, + Ssl_Csrs, + Ssl_Crls, + Ssl_Keys] class Data_Groups(Collection): def __init__(self, File): super(Data_Groups, self).__init__(File) self._meta_data['allowed_lazy_attributes'] = [Data_Group] - self._meta_data['required_json_kind'] = \ - u'tm:sys:file:data-group:data-groupcollectionstate' self._meta_data['attribute_registry'] =\ {u'tm:sys:file:data-group:data-groupstate': Data_Group} - self._meta_data['uri'] = self._meta_data['uri'].replace('_', '-') class Data_Group(Resource): @@ -55,5 +57,90 @@ def __init__(self, data_groups): self._meta_data['required_json_kind'] =\ u'tm:sys:file:data-group:data-groupstate' self._meta_data['required_creation_parameters'].update( - ('name', 'sourcePath', 'type') - ) + ('name', 'sourcePath', 'type')) + + +class Ifiles(Collection): + def __init__(self, File): + super(Ifiles, self).__init__(File) + self._meta_data['allowed_lazy_attributes'] = [Ifile] + self._meta_data['attribute_registry'] = \ + {u'tm:sys:file:ifile:ifilestate': Ifile} + + +class Ifile(Resource): + def __init__(self, ifiles): + super(Ifile, self).__init__(ifiles) + self._meta_data['required_json_kind'] =\ + u'tm:sys:file:ifile:ifilestate' + self._meta_data['required_creation_parameters'].update( + ('name', 'sourcePath')) + + +class Ssl_Certs(Collection): + def __init__(self, File): + super(Ssl_Certs, self).__init__(File) + self._meta_data['allowed_lazy_attributes'] = [Ssl_Cert] + self._meta_data['attribute_registry'] =\ + {u'tm:sys:file:ssl-cert:ssl-certstate': Ssl_Cert} + + +class Ssl_Cert(Resource): + def __init__(self, ssl_certs): + super(Ssl_Cert, self).__init__(ssl_certs) + self._meta_data['required_json_kind'] =\ + u'tm:sys:file:ssl-cert:ssl-certstate' + self._meta_data['required_creation_parameters'].update( + ('name', 'sourcePath')) + + +class Ssl_Crls(Collection): + def __init__(self, File): + super(Ssl_Crls, self).__init__(File) + self._meta_data['allowed_lazy_attributes'] = [Ssl_Crl] + self._meta_data['attribute_registry'] =\ + {u'tm:sys:file:ssl-crl:ssl-crlstate': Ssl_Crl} + + +class Ssl_Crl(Resource): + def __init__(self, ssl_crls): + super(Ssl_Crl, self).__init__(ssl_crls) + self._meta_data['required_json_kind'] =\ + u'tm:sys:file:ssl-crl:ssl-crlstate' + self._meta_data['required_creation_parameters'].update( + ('name', 'sourcePath')) + + +class Ssl_Csrs(Collection): + def __init__(self, File): + super(Ssl_Csrs, self).__init__(File) + self._meta_data['allowed_lazy_attributes'] = [Ssl_Csr] + self._meta_data['attribute_registry'] =\ + {u'tm:sys:file:ssl-csr:ssl-csrstate': Ssl_Csr} + self._meta_data['minimum_version'] = '12.0.0' + + +class Ssl_Csr(Resource): + def __init__(self, ssl_csrs): + super(Ssl_Csr, self).__init__(ssl_csrs) + self._meta_data['required_json_kind'] =\ + u'tm:sys:file:ssl-csr:ssl-csrstate' + self._meta_data['required_creation_parameters'].update( + ('name', 'sourcePath')) + + +class Ssl_Keys(Collection): + def __init__(self, File): + super(Ssl_Keys, self).__init__(File) + self._meta_data['allowed_lazy_attributes'] = [Ssl_Key] + self._meta_data['attribute_registry'] =\ + {u'tm:sys:file:ssl-key:ssl-keystate': Ssl_Key} + + +class Ssl_Key(Resource): + def __init__(self, ssl_keys): + super(Ssl_Key, self).__init__(ssl_keys) + self._meta_data['required_json_kind'] =\ + u'tm:sys:file:ssl-key:ssl-keystate' + self._meta_data['required_creation_parameters'].update( + ('name', 'sourcePath')) diff --git a/f5/bigip/tm/sys/test/test_db.py b/f5/bigip/tm/sys/test/test_db.py index ef4e9e234..a1c173855 100644 --- a/f5/bigip/tm/sys/test/test_db.py +++ b/f5/bigip/tm/sys/test/test_db.py @@ -23,7 +23,7 @@ def fake_dbs(): fake_sys = mock.MagicMock() dbs = Dbs(fake_sys) - dbs._meta_data['bigip']._meta_data = {'tmos_version': '11.6.0'} + dbs._meta_data['bigip'].tmos_version = '11.6.0' return dbs diff --git a/f5/bigip/tm/sys/test/test_file.py b/f5/bigip/tm/sys/test/test_file.py new file mode 100644 index 000000000..d7c197443 --- /dev/null +++ b/f5/bigip/tm/sys/test/test_file.py @@ -0,0 +1,127 @@ +# 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. +# + +from distutils.version import LooseVersion +import mock +import pytest + +from f5.bigip.resource import MissingRequiredCreationParameter +from f5.bigip.tm.sys.file import Ifile +from f5.bigip.tm.sys.file import Ssl_Cert +from f5.bigip.tm.sys.file import Ssl_Crl +from f5.bigip.tm.sys.file import Ssl_Csr +from f5.bigip.tm.sys.file import Ssl_Key + + +@pytest.fixture +def FakeSysIfile(): + fake_ifile_s = mock.MagicMock() + fake_ifile = Ifile(fake_ifile_s) + return fake_ifile + + +def test_ifile_create_no_args(FakeSysIfile): + with pytest.raises(MissingRequiredCreationParameter): + FakeSysIfile.create() + + +def test_ifile_create_missing_arg(FakeSysIfile): + with pytest.raises(MissingRequiredCreationParameter) as ex: + FakeSysIfile.create(name='test_ifile') + assert 'sourcePath' in ex.value.message + + +@pytest.fixture +def FakeSysCert(): + fake_cert_s = mock.MagicMock() + fake_cert = Ssl_Cert(fake_cert_s) + return fake_cert + + +def test_cert_create_no_args(FakeSysCert): + with pytest.raises(MissingRequiredCreationParameter): + FakeSysCert.create() + + +def test_cert_create_missing_arg(FakeSysCert): + with pytest.raises(MissingRequiredCreationParameter) as ex: + FakeSysCert.create(name='test_cert') + assert 'sourcePath' in ex.value.message + + +@pytest.fixture +def FakeSysCrl(): + fake_crl_s = mock.MagicMock() + fake_crl = Ssl_Crl(fake_crl_s) + return fake_crl + + +def test_crl_create_no_args(FakeSysCrl): + with pytest.raises(MissingRequiredCreationParameter): + FakeSysCrl.create() + + +def test_crl_create_missing_arg(FakeSysCrl): + with pytest.raises(MissingRequiredCreationParameter) as ex: + FakeSysCrl.create(name='test_crl') + assert 'sourcePath' in ex.value.message + + +@pytest.fixture +def FakeSysCsr(): + fake_csr_s = mock.MagicMock() + fake_csr = Ssl_Csr(fake_csr_s) + return fake_csr + + +@pytest.mark.skipif( + LooseVersion( + pytest.config.getoption('--release') + ) < LooseVersion('12.0.0'), + reason='csr management is only supported in 12.0.0 or greater.' +) +def test_csr_create_no_args(FakeSysCsr): + with pytest.raises(MissingRequiredCreationParameter): + FakeSysCsr.create() + + +@pytest.mark.skipif( + LooseVersion( + pytest.config.getoption('--release') + ) < LooseVersion('12.0.0'), + reason='csr management is only supported in 12.0.0 or greater.' +) +def test_csr_create_missing_arg(FakeSysCsr): + with pytest.raises(MissingRequiredCreationParameter) as ex: + FakeSysCsr.create(name='test_csr') + assert 'sourcePath' in ex.value.message + + +@pytest.fixture +def FakeSysKey(): + fake_key_s = mock.MagicMock() + fake_key = Ssl_Key(fake_key_s) + return fake_key + + +def test_key_create_no_args(FakeSysKey): + with pytest.raises(MissingRequiredCreationParameter): + FakeSysKey.create() + + +def test_key_create_missing_arg(FakeSysKey): + with pytest.raises(MissingRequiredCreationParameter) as ex: + FakeSysKey.create(name='test_key') + assert 'sourcePath' in ex.value.message diff --git a/f5/bigip/tm/sys/test/test_folder.py b/f5/bigip/tm/sys/test/test_folder.py index c41510b82..7c60775fd 100644 --- a/f5/bigip/tm/sys/test/test_folder.py +++ b/f5/bigip/tm/sys/test/test_folder.py @@ -24,7 +24,7 @@ def FakeFolders(): fake_sys = mock.MagicMock() folders = Folders(fake_sys) - folders._meta_data['bigip']._meta_data = {'tmos_version': '11.6.0'} + folders._meta_data['bigip'].tmos_version = '11.6.0' return folders diff --git a/f5/bigip/tm/sys/test/test_performance.py b/f5/bigip/tm/sys/test/test_performance.py index 47633bc7b..ab820a335 100644 --- a/f5/bigip/tm/sys/test/test_performance.py +++ b/f5/bigip/tm/sys/test/test_performance.py @@ -25,7 +25,7 @@ def FakePerformance(): fake_sys = mock.MagicMock() performances = Performances(fake_sys) - performances._meta_data['bigip']._meta_data = {'tmos_version': '11.6.0'} + performances._meta_data['bigip'].tmos_version = '11.6.0' return performances diff --git a/requirements.test.txt b/requirements.test.txt index 2fc262efd..0e1f5732b 100644 --- a/requirements.test.txt +++ b/requirements.test.txt @@ -8,4 +8,5 @@ mock==1.3.0 pytest==2.9.1 pytest-cov>=2.2.1 git+https://github.com/F5Networks/pytest-symbols.git -python-coveralls \ No newline at end of file +python-coveralls +pyopenssl diff --git a/test/functional/test/test__init__.py b/test/functional/test/test__init__.py index 06571d201..e5d33b37f 100644 --- a/test/functional/test/test__init__.py +++ b/test/functional/test/test__init__.py @@ -27,3 +27,10 @@ def test_invalid_args(opt_bigip, opt_username, opt_password, opt_port): def test_icontrol_version(opt_bigip, opt_username, opt_password, opt_port): m = ManagementRoot(opt_bigip, opt_username, opt_password, port=opt_port) assert hasattr(m, 'icontrol_version') + + +def test_tmos_version(mgmt_root): + assert mgmt_root.tmos_version == \ + mgmt_root._meta_data['bigip']._meta_data['tmos_version'] + assert mgmt_root.tmos_version is not None + assert mgmt_root._meta_data['bigip']._meta_data['tmos_version'] != '' diff --git a/test/functional/tm/ltm/test_ifile.py b/test/functional/tm/ltm/test_ifile.py new file mode 100644 index 000000000..22585bae1 --- /dev/null +++ b/test/functional/tm/ltm/test_ifile.py @@ -0,0 +1,91 @@ +# 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. +# + +from f5.bigip.resource import MissingRequiredCreationParameter +import pytest +from requests.exceptions import HTTPError + + +def delete_ifile(mgmt_root, name, partition, IFILE): + try: + ifile = mgmt_root.tm.ltm.ifiles.ifile.load(name=name, + partition=partition) + except HTTPError as err: + if err.response.status_code != 404: + raise + return + ifile.delete() + + # not testing this function here; but need to clean it up + try: + IFILE.delete() + except HTTPError as err: + if err.response.status_code != 404: + raise + return + + +def setup_create_test(request, mgmt_root, name, partition, IFILE): + def teardown(): + delete_ifile(mgmt_root, name, partition, IFILE) + + request.addfinalizer(teardown) + + +def setup_basic_test(request, mgmt_root, name, partition, IFILE): + def teardown(): + delete_ifile(mgmt_root, name, partition, IFILE) + + ifile1 = mgmt_root.tm.ltm.ifiles.ifile.create( + name='ifile1', partition='Common', fileName=IFILE.name) + + request.addfinalizer(teardown) + return ifile1 + + +def test_create_no_args(mgmt_root): + with pytest.raises(MissingRequiredCreationParameter): + mgmt_root.tm.ltm.ifiles.ifile.create() + + +def test_create_no_filename(mgmt_root): + with pytest.raises(MissingRequiredCreationParameter): + mgmt_root.tm.ltm.ifiles.ifile.create(name='ifile1', partition='Common') + + +def test_create(request, mgmt_root, IFILE): + setup_create_test(request, mgmt_root, 'ifile1', 'Common', IFILE) + ifile1 = mgmt_root.tm.ltm.ifiles.ifile.create( + name='ifile1', partition='Common', fileName=IFILE.name) + + assert ifile1.name == 'ifile1' + assert ifile1.partition == 'Common' + + +def test_delete(request, mgmt_root, IFILE): + ifile1 = setup_basic_test(request, mgmt_root, 'ifile1', 'Common', IFILE) + ifile1.delete() + with pytest.raises(HTTPError) as err: + mgmt_root.tm.ltm.ifiles.ifile.load( + name='ifile1', partition='Common') + assert err.response.status_code == 404 + + # not testing this function here; but need to clean it up + try: + IFILE.delete() + except HTTPError as err: + if err.response.status_code != 404: + raise + return diff --git a/test/functional/tm/sys/test_file.py b/test/functional/tm/sys/test_file.py new file mode 100644 index 000000000..83303a78c --- /dev/null +++ b/test/functional/tm/sys/test_file.py @@ -0,0 +1,314 @@ + +# 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. +# + +from distutils.version import LooseVersion +import pytest + + +from OpenSSL import crypto +import os +from requests import HTTPError +from tempfile import NamedTemporaryFile + + +def gen_key(): + # returns key pair in PKey object + key = crypto.PKey() + key.generate_key(crypto.TYPE_RSA, 2048) + return key + + +def gen_csr(key, **name): + # returns the certificate request in an X509Req object + req = crypto.X509Req() + subj = req.get_subject() + for (k, v) in name.items(): + setattr(subj, k, v) + req.set_pubkey(key) + req.sign(key, 'sha1') + return req + + +def gen_cert(req, (ca_cert, ca_key), serial): + # returns the signed certificate in an X509 object + cert = crypto.X509() + cert.set_serial_number(serial) + cert.gmtime_adj_notBefore(0) + cert.gmtime_adj_notAfter(86400) + cert.set_issuer(ca_cert.get_subject()) + cert.set_subject(req.get_subject()) + cert.set_pubkey(req.get_pubkey()) + cert.sign(ca_key, 'sha1') + return cert + + +def create_sslfiles(): + # Create a CA Key/Cert + ca_key = gen_key() + ca_csr = gen_csr(ca_key, CN='Certificate Authority') + ca_cert = gen_cert(ca_csr, (ca_csr, ca_key), 0) + + # Create Key/Cert/CSR for uploading to BIG-IP + key = gen_key() + csr = gen_csr(key, CN='mycert.test.local') + cert = gen_cert(csr, (ca_cert, ca_key), 1) + + return key, csr, cert + + +def setup_ifile_test(request, mgmt_root, name, sourcepath): + if1 = mgmt_root.tm.sys.file.ifiles.ifile.create(name=name, + sourcePath=sourcepath) + + def teardown(): + # Remove the ifile. + try: + if1.delete() + except HTTPError as err: + if err.response.status_code != 404: + raise + request.addfinalizer(teardown) + + return if1 + + +def test_CURDL_ifile(request, mgmt_root): + # Create + ntf = NamedTemporaryFile() + ntf_basename = os.path.basename(ntf.name) + ntf.write('this is a test file') + ntf.seek(0) + # Upload the file + mgmt_root.shared.file_transfer.uploads.upload_file(ntf.name) + + tpath_name = 'file:/var/config/rest/downloads/{0}'.format(ntf_basename) + if1 = setup_ifile_test(request, mgmt_root, ntf_basename, tpath_name) + assert if1.name == ntf_basename + + # Load Object + if2 = mgmt_root.tm.sys.file.ifiles.ifile.load(name=ntf_basename) + assert if1.name == if2.name + + # Rewrite file contents and Update Object + ntf.write('this is still a test file') + ntf.seek(0) + mgmt_root.shared.file_transfer.uploads.upload_file(ntf.name) + + if3 = mgmt_root.tm.sys.file.ifiles.ifile.load(name=ntf_basename) + if3.update(sourcePath=tpath_name) + assert if1.revision != if3.revision + + # Refresh if2 and make sure revision matches if3 + if2.refresh() + assert if2.revision == if3.revision + + +def setup_sslkey_test(request, mgmt_root, name, sourcepath): + key = mgmt_root.tm.sys.file.ssl_keys.ssl_key.create(name=name, + sourcePath=sourcepath) + + def teardown(): + # Remove the key. + try: + key.delete() + except HTTPError as err: + if err.response.status_code != 404: + raise + request.addfinalizer(teardown) + + return key + + +def setup_sslcsr_test(request, mgmt_root, name, sourcepath): + csr = mgmt_root.tm.sys.file.ssl_csrs.ssl_csr.create(name=name, + sourcePath=sourcepath) + + def teardown(): + # Remove the key. + try: + csr.delete() + except HTTPError as err: + if err.response.status_code != 404: + raise + + request.addfinalizer(teardown) + + return csr + + +def setup_sslcrt_test(request, mgmt_root, name, sourcepath): + cert = mgmt_root.tm.sys.file.ssl_certs.ssl_cert.create( + name=name, sourcePath=sourcepath) + + def teardown(): + # Remove the key. + try: + cert.delete() + except HTTPError as err: + if err.response.status_code != 404: + raise + + request.addfinalizer(teardown) + + return cert + + +def test_CURDL_sslkeyfile(request, mgmt_root): + # Create temporary Key File. + # Use extensions so tmui doesn't break in managing them. + ntf_key = NamedTemporaryFile(suffix='.key') + ntf_key_basename = os.path.basename(ntf_key.name) + ntf_key_sourcepath = 'file:/var/config/rest/downloads/{0}'.format( + ntf_key_basename) + + # Create a CA Key/Cert + key, csr, cert = create_sslfiles() + + # Write Data to Temporary File + ntf_key.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, key)) + ntf_key.seek(0) + + # Upload File to BIG-IP + mgmt_root.shared.file_transfer.uploads.upload_file(ntf_key.name) + + # Finally, Let's test something! + key1 = setup_sslkey_test(request, mgmt_root, ntf_key_basename, + ntf_key_sourcepath) + assert key1.name == ntf_key_basename + + key2 = mgmt_root.tm.sys.file.ssl_keys.ssl_key.load(name=ntf_key_basename) + assert key1.name == key2.name + + # Create new CA Key/Cert + key, csr, cert = create_sslfiles() + + # Write new data to Temporary File + ntf_key.seek(0) + ntf_key.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, key)) + ntf_key.truncate() + ntf_key.seek(0) + + # Upload File to BIG-IP + mgmt_root.shared.file_transfer.uploads.upload_file(ntf_key.name) + + # Update Key + key2.update() + assert key2.revision != key1.revision + + # Refresh Key + key1.refresh() + assert key2.revision == key1.revision + + +@pytest.mark.skipif( + LooseVersion( + pytest.config.getoption('--release') + ) < LooseVersion('12.0.0'), + reason='csr management is only supported in 12.0.0 or greater.' +) +def test_CURDL_sslcsrfile(request, mgmt_root): + # Create temporary CSR File. + # Use extensions so tmui doesn't break in managing them. + ntf_csr = NamedTemporaryFile(suffix='.csr') + ntf_csr_basename = os.path.basename(ntf_csr.name) + ntf_csr_sourcepath = 'file:/var/config/rest/downloads/{0}'.format( + ntf_csr_basename) + + # Create a CA Key/Cert + key, csr, cert = create_sslfiles() + + # Write Data to Temporary File + ntf_csr.write(crypto.dump_certificate_request(crypto.FILETYPE_PEM, csr)) + ntf_csr.seek(0) + + # Upload File to BIG-IP + mgmt_root.shared.file_transfer.uploads.upload_file(ntf_csr.name) + + # Finally, Let's test something! + csr1 = setup_sslcsr_test(request, mgmt_root, ntf_csr_basename, + ntf_csr_sourcepath) + assert csr1.name == ntf_csr_basename + + csr2 = mgmt_root.tm.sys.file.ssl_csrs.ssl_csr.load(name=ntf_csr_basename) + assert csr1.name == csr2.name + + # Create new CA Key/Cert + key, csr, cert = create_sslfiles() + + # Write new data to Temporary File + ntf_csr.seek(0) + ntf_csr.write(crypto.dump_certificate_request(crypto.FILETYPE_PEM, csr)) + ntf_csr.truncate() + ntf_csr.seek(0) + + # Upload File to BIG-IP + mgmt_root.shared.file_transfer.uploads.upload_file(ntf_csr.name) + + # Update Key + csr2.update() + assert csr2.revision != csr1.revision + + # Refresh Key + csr1.refresh() + assert csr2.revision == csr1.revision + + +def test_CURDL_sslcertfile(request, mgmt_root): + # Create temporary CSR File. + # Use extensions so tmui doesn't break in managing them. + ntf_cert = NamedTemporaryFile(suffix='.crt') + ntf_cert_basename = os.path.basename(ntf_cert.name) + ntf_cert_sourcepath = 'file:/var/config/rest/downloads/{0}'.format( + ntf_cert_basename) + + # Create a CA Key/Cert + key, csr, cert = create_sslfiles() + + # Write Data to Temporary File + ntf_cert.write(crypto.dump_certificate(crypto.FILETYPE_PEM, cert)) + ntf_cert.seek(0) + + # Upload File to BIG-IP + mgmt_root.shared.file_transfer.uploads.upload_file(ntf_cert.name) + + # Finally, Let's test something! + cert1 = setup_sslcrt_test(request, mgmt_root, ntf_cert_basename, + ntf_cert_sourcepath) + assert cert1.name == ntf_cert_basename + + cert2 = mgmt_root.tm.sys.file.ssl_certs.ssl_cert.load( + name=ntf_cert_basename) + assert cert1.name == cert2.name + + # Create new CA Key/Cert + key, csr, cert = create_sslfiles() + + # Write new data to Temporary File + ntf_cert.seek(0) + ntf_cert.write(crypto.dump_certificate(crypto.FILETYPE_PEM, cert)) + ntf_cert.truncate() + ntf_cert.seek(0) + + # Upload File to BIG-IP + mgmt_root.shared.file_transfer.uploads.upload_file(ntf_cert.name) + + # Update Key + cert2.update() + assert cert2.revision != cert1.revision + + # Refresh Key + cert1.refresh() + assert cert2.revision == cert1.revision