Merge pull request #1526 from jasonrahm/enh.external_monitor

Issue #1521 - Add sys/file external monitor suport

f5/bigip/tm/sys/file.py

@@ -1,6 +1,6 @@
 # coding=utf-8
 #
-# Copyright 2016 F5 Networks Inc.
+# Copyright 2019 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.
@@ -39,6 +39,7 @@ def __init__(self, sys):
         super(File, self).__init__(sys)
         self._meta_data['allowed_lazy_attributes'] = [
             Data_Groups,
+            External_Monitors,
             Ifiles,
             Ssl_Certs,
             Ssl_Csrs,
@@ -72,6 +73,25 @@ def update(self, **kwargs):
         return self._update(**kwargs)
 
 
+class External_Monitors(Collection):
+    """BIG-IP® System sys file data-groups collection."""
+    def __init__(self, File):
+        super(External_Monitors, self).__init__(File)
+        self._meta_data['allowed_lazy_attributes'] = [External_Monitor]
+        self._meta_data['attribute_registry'] =\
+            {'tm:sys:file:external-monitor:external-monitorstate': External_Monitor}
+
+
+class External_Monitor(Resource):
+    """BIG-IP® System sys file data-groups resource."""
+    def __init__(self, external_monitors):
+        super(External_Monitor, self).__init__(external_monitors)
+        self._meta_data['required_json_kind'] =\
+            'tm:sys:file:external-monitor:external-monitorstate'
+        self._meta_data['required_creation_parameters'].update(
+            ('name', 'sourcePath'))
+
+
 class Ifiles(Collection):
     """BIG-IP® System sys file iFiles collection."""
     def __init__(self, File):

f5/bigip/tm/sys/test/functional/test_file.py

@@ -90,7 +90,7 @@ def test_CURDL_datagroup(request, mgmt_root):
     # Create
     ntf = NamedTemporaryFile(delete=False)
     ntf_basename = os.path.basename(ntf.name)
-    ntf.write('"name1" := "value1",')
+    ntf.write(b'"name1" := "value1",')
     ntf.seek(0)
     # Upload the file
     mgmt_root.shared.file_transfer.uploads.upload_file(ntf.name)
@@ -103,7 +103,7 @@ def test_CURDL_datagroup(request, mgmt_root):
     assert dg1.name == dg2.name
 
     # Rewrite the contents and update the object
-    ntf.write('"name2" := "value2",')
+    ntf.write(b'"name2" := "value2",')
     ntf.seek(0)
     mgmt_root.shared.file_transfer.uploads.upload_file(ntf.name)
 
@@ -120,6 +120,55 @@ def test_CURDL_datagroup(request, mgmt_root):
     assert dg1.revision != dg4.revision
 
 
+def setup_externalmonitor_test(request, mgmt_root, name, sourcepath, **kwargs):
+    em1 = mgmt_root.tm.sys.file.external_monitors.external_monitor.create(name=name,
+                                                                          sourcePath=sourcepath,
+                                                                          **kwargs)
+
+    def teardown():
+        # Remove the ifile.
+        try:
+            em1.delete()
+        except HTTPError as err:
+            if err.response.status_code != 404:
+                raise
+
+    request.addfinalizer(teardown)
+
+    return em1
+
+
+def test_CURDL_externalmonitor(request, mgmt_root):
+    # Create
+    ntf = NamedTemporaryFile(delete=False)
+    ntf_basename = os.path.basename(ntf.name)
+    ntf.write(b'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)
+    em1 = setup_externalmonitor_test(request, mgmt_root, ntf_basename, tpath_name)
+    assert em1.name == ntf_basename
+
+    # Load Object
+    em2 = mgmt_root.tm.sys.file.external_monitors.external_monitor.load(name=ntf_basename)
+    assert em1.name == em2.name
+
+    # Rewrite file contents and Update Object
+    ntf.write(b'this is still a test file')
+    ntf.seek(0)
+    mgmt_root.shared.file_transfer.uploads.upload_file(ntf.name)
+
+    em3 = mgmt_root.tm.sys.file.external_monitors.external_monitor.load(name=ntf_basename)
+    em3.update(sourcePath=tpath_name)
+    assert em1.revision != em3.revision
+
+    # Refresh em2 and make sure revision matches em3
+    em2.refresh()
+    assert em2.revision == em3.revision
+
+
 def setup_ifile_test(request, mgmt_root, name, sourcepath, **kwargs):
     if1 = mgmt_root.tm.sys.file.ifiles.ifile.create(name=name,
                                                     sourcePath=sourcepath,
@@ -141,22 +190,21 @@ def test_CURDL_ifile(request, mgmt_root):
     # Create
     ntf = NamedTemporaryFile(delete=False)
     ntf_basename = os.path.basename(ntf.name)
-    ntf.write('this is a test file')
+    ntf.write(b'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,
-                           )
+    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.write(b'this is still a test file')
     ntf.seek(0)
     mgmt_root.shared.file_transfer.uploads.upload_file(ntf.name)
 

f5/bigip/tm/sys/test/unit/test_file.py

@@ -17,6 +17,7 @@
 import pytest
 
 from f5.bigip.tm.sys.file import Data_Group
+from f5.bigip.tm.sys.file import External_Monitor
 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
@@ -41,8 +42,26 @@ def test_dg_create_no_args(FakeSysDatagroup):
 def test_dg_create_missing_arg(FakeSysDatagroup):
     with pytest.raises(MissingRequiredCreationParameter) as ex:
         FakeSysDatagroup.create(name='test_dg')
-        assert 'sourcePath' in ex.value.message
-        assert 'type' in ex.value.message
+    assert 'sourcePath' in str(ex.value)
+    assert 'type' in str(ex.value)
+
+
+@pytest.fixture
+def FakeSysExternalMonitor():
+    fake_em_s = mock.MagicMock()
+    fake_em = External_Monitor(fake_em_s)
+    return fake_em
+
+
+def test_em_create_no_args(FakeSysExternalMonitor):
+    with pytest.raises(MissingRequiredCreationParameter):
+        FakeSysExternalMonitor.create()
+
+
+def test_em_create_missing_arg(FakeSysExternalMonitor):
+    with pytest.raises(MissingRequiredCreationParameter) as ex:
+        FakeSysExternalMonitor.create(name='test_em')
+    assert 'sourcePath' in str(ex.value)
 
 
 @pytest.fixture
@@ -60,7 +79,7 @@ def test_ifile_create_no_args(FakeSysIfile):
 def test_ifile_create_missing_arg(FakeSysIfile):
     with pytest.raises(MissingRequiredCreationParameter) as ex:
         FakeSysIfile.create(name='test_ifile')
-        assert 'sourcePath' in ex.value.message
+    assert 'sourcePath' in str(ex.value)
 
 
 def test_ifile_modify(FakeSysIfile):
@@ -83,7 +102,7 @@ def test_cert_create_no_args(FakeSysCert):
 def test_cert_create_missing_arg(FakeSysCert):
     with pytest.raises(MissingRequiredCreationParameter) as ex:
         FakeSysCert.create(name='test_cert')
-        assert 'sourcePath' in ex.value.message
+    assert 'sourcePath' in str(ex.value)
 
 
 def test_cert_modify(FakeSysCert):
@@ -106,7 +125,7 @@ def test_crl_create_no_args(FakeSysCrl):
 def test_crl_create_missing_arg(FakeSysCrl):
     with pytest.raises(MissingRequiredCreationParameter) as ex:
         FakeSysCrl.create(name='test_crl')
-        assert 'sourcePath' in ex.value.message
+    assert 'sourcePath' in str(ex.value)
 
 
 def test_crl_modify(FakeSysCrl):
@@ -134,7 +153,7 @@ def test_csr_modify(FakeSysCsr):
 def test_csr_create_missing_arg(FakeSysCsr):
     with pytest.raises(MissingRequiredCreationParameter) as ex:
         FakeSysCsr.create(name='test_csr')
-        assert 'sourcePath' in ex.value.message
+    assert 'sourcePath' in str(ex.value)
 
 
 @pytest.fixture
@@ -152,7 +171,7 @@ def test_key_create_no_args(FakeSysKey):
 def test_key_create_missing_arg(FakeSysKey):
     with pytest.raises(MissingRequiredCreationParameter) as ex:
         FakeSysKey.create(name='test_key')
-        assert 'sourcePath' in ex.value.message
+    assert 'sourcePath' in str(ex.value)
 
 
 def test_key_modify(FakeSysKey):