import os
import tempfile
import unittest
from time import sleep
import pytest
import yaml
from cowbird.handlers.handler_factory import HandlerFactory
from cowbird.monitoring.fsmonitor import FSMonitor
from cowbird.monitoring.monitoring import Monitoring
from tests.utils import clear_handlers_instances, get_test_app
[docs]def file_io(filename, mv_filename):
# Create
with open(filename, "w", encoding="utf-8") as f:
f.write("Hello")
# Update
with open(filename, "a", encoding="utf-8") as f:
f.write(" world!")
# Update on file permissions should also trigger a modified event
os.chmod(filename, 0o777)
# Should create a 'delete' and a 'create' event
os.rename(filename, mv_filename)
# Delete
os.remove(mv_filename)
@pytest.mark.monitoring
[docs]class TestMonitoring(unittest.TestCase):
@classmethod
[docs] def setUpClass(cls):
cls.cfg_file = tempfile.NamedTemporaryFile(mode="w", suffix=".cfg", delete=False) # pylint: disable=R1732
with cls.cfg_file as f:
f.write(yaml.safe_dump({"handlers": {"": {}}})) # no specific handler required for these tests
cls.app = get_test_app(settings={"cowbird.config_path": cls.cfg_file.name})
# clear up monitor entries
Monitoring().unregister_all()
@classmethod
[docs] def tearDownClass(cls):
Monitoring().store.clear_services()
clear_handlers_instances()
os.unlink(cls.cfg_file.name)
[docs] def test_register_unregister_monitor(self):
with tempfile.TemporaryDirectory() as tmpdir:
test_file = os.path.join(tmpdir, "testfile")
test_subdir = os.path.join(tmpdir, "subdir")
test_subdir_file = os.path.join(test_subdir, "test_subdir_file")
mv_test_file = os.path.join(test_subdir, "moved_testfile")
mv_test_subdir_file = os.path.join(tmpdir, "moved_test_subdir_file")
os.mkdir(test_subdir)
# Test registering directly a callback instance
mon = TestMonitor()
internal_mon = Monitoring().register(tmpdir, False, mon)
assert internal_mon.callback_instance == mon
# Test registering a callback via class name
internal_mon2 = Monitoring().register(tmpdir, True, TestMonitor2)
mon2 = internal_mon2.callback_instance
internal_mon3 = Monitoring().register(test_subdir, False, TestMonitor)
mon3 = internal_mon3.callback_instance
assert isinstance(mon2, TestMonitor2)
assert isinstance(mon3, TestMonitor)
# Test collision when 2 monitors are registered using the same path and class name
assert not internal_mon3.recursive
internal_mon4 = Monitoring().register(test_subdir, True, TestMonitor)
assert internal_mon4 is internal_mon3
assert internal_mon3.recursive # Recursive should take precedence when a collision occurs
# monitors first level is distinct path to monitor : (tmp_dir and test_subdir)
assert len(Monitoring().monitors) == 2
# monitors second level is distinct callback, for tmpdir : (TestMonitor and TestMonitor2)
assert len(Monitoring().monitors[tmpdir]) == 2
# Do some io operations that should be picked by the monitors
file_io(test_file, mv_test_file)
file_io(test_subdir_file, mv_test_subdir_file)
sleep(1)
# Root dir non-recursive
assert len(mon.created) == 2
assert mon.created[0] == test_file
assert mon.created[1] == mv_test_subdir_file
assert mon.created == mon.deleted
assert sorted(set(mon.modified)) == [tmpdir, test_file]
assert len(mon.modified) == 9
# Root dir recursive
assert len(mon2.created) == 4
assert mon2.created[0] == test_file
assert mon2.created[1] == mv_test_file
assert mon2.created[2] == test_subdir_file
assert mon2.created[3] == mv_test_subdir_file
assert mon2.created == mon2.deleted
assert sorted(set(mon2.modified)) == [tmpdir, test_subdir,
test_subdir_file, test_file]
assert len(mon2.modified) == 18
# Subdir
assert len(mon3.created) == 2
assert mon3.created[0] == mv_test_file
assert mon3.created[1] == test_subdir_file
assert mon3.created == mon3.deleted
assert sorted(set(mon3.modified)) == [test_subdir,
test_subdir_file]
assert len(mon3.modified) == 9
# Validate cleanup
Monitoring().unregister(tmpdir, mon)
Monitoring().unregister(tmpdir, mon2)
# Here we try to unregister a path with a bad class type
assert not Monitoring().unregister(test_subdir, mon2)
# Here we have the good class type
Monitoring().unregister(test_subdir, mon3)
assert len(Monitoring().monitors) == 0
assert not Monitoring().unregister(tmpdir, mon)
# Test registering a callback via a qualified class name string
catalog_mon = \
Monitoring().register(tmpdir, False, "cowbird.handlers.impl.catalog.Catalog").callback_instance
assert catalog_mon == HandlerFactory().get_handler("Catalog")
[docs]class TestMonitor(FSMonitor):
[docs] __test__ = False # avoid invalid collect depending on specified input path/items to pytest
def __init__(self):
self.created = []
self.deleted = []
self.modified = []
@staticmethod
[docs] def get_instance():
return TestMonitor()
[docs] def on_created(self, path):
self.created.append(path)
[docs] def on_deleted(self, path):
self.deleted.append(path)
[docs] def on_modified(self, path):
self.modified.append(path)
[docs]class TestMonitor2(TestMonitor):
[docs] __test__ = False # avoid invalid collect depending on specified input path/items to pytest
# Allow a second full qualified class name to register as monitor
@staticmethod
[docs] def get_instance():
return TestMonitor2()