import os
import tempfile
import unittest
import mock
import pytest
import yaml
from pymongo.collection import Collection
from pymongo.cursor import Cursor
from cowbird.database.mongodb import MongoDatabase
from cowbird.database.stores import MonitoringStore
from cowbird.monitoring.monitor import Monitor
from tests import utils
@pytest.mark.database
[docs]class MongodbServiceStoreTestCase(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": {"Catalog": {"active": True, "url": "http://catalog", "workspace_dir": "/workspace"}}}))
cls.app = utils.get_test_app(settings={"cowbird.config_path": cls.cfg_file.name})
@classmethod
[docs] def tearDownClass(cls):
utils.clear_handlers_instances()
os.unlink(cls.cfg_file.name)
[docs] def setUp(self):
self.monitor_params = {"path": "/", "recursive": False, "callback": "cowbird.handlers.impl.catalog.Catalog"}
self.monitor_params_bad_path = {"path": "", "recursive": False,
"callback": "cowbird.handlers.impl.catalog.Catalog"}
self.monitor_params_bad_callback = {"path": "/", "recursive": False, "callback": ""}
self.monitor = Monitor(**self.monitor_params)
@pytest.mark.database
[docs] def test_get_store(self):
collection_mock = mock.Mock(spec=Collection)
store_mock = MonitoringStore(collection=collection_mock)
db = MongoDatabase({})
store_t1 = db.get_store(store_mock)
store_t2 = db.get_store(MonitoringStore)
store_t3 = db.get_store(MonitoringStore.type)
assert isinstance(store_t1, MonitoringStore)
assert isinstance(store_t2, MonitoringStore)
assert isinstance(store_t3, MonitoringStore)
[docs] def test_save_monitor(self):
collection_mock = mock.Mock(spec=Collection)
collection_mock.count_documents.return_value = 1
store = MonitoringStore(collection=collection_mock)
store.save_monitor(self.monitor)
collection_mock.delete_one.assert_called_with(self.monitor.key)
collection_mock.insert_one.assert_called_with(self.monitor.params())
[docs] def test_delete_monitor(self):
collection_mock = mock.Mock(spec=Collection)
store = MonitoringStore(collection=collection_mock)
store.delete_monitor(self.monitor)
collection_mock.delete_one.assert_called_with(self.monitor.key)
[docs] def test_list_monitors(self):
cursor_mock = mock.Mock(spec=Cursor)
cursor_mock.sort.return_value = [self.monitor_params,
self.monitor_params_bad_path,
self.monitor_params_bad_callback]
collection_mock = mock.Mock(spec=Collection)
collection_mock.find.return_value = cursor_mock
store = MonitoringStore(collection=collection_mock)
monitors = store.list_monitors()
assert len(monitors) == 1
assert monitors[0].params() == self.monitor_params
# Store should remove monitors with bad parameters from database
collection_mock.delete_one.assert_has_calls([mock.call(self.monitor_params_bad_path),
mock.call(self.monitor_params_bad_callback)])