diff options
Diffstat (limited to 'testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py')
-rw-r--r-- | testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py | 29 |
1 files changed, 29 insertions, 0 deletions
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py index 68313e6fb..32766b5c1 100644 --- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py +++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py @@ -1,4 +1,5 @@ import os +import re import sys import shutil import tempfile @@ -229,10 +230,12 @@ group: group:with:colons def setUp(self): Bcfg2TestCase.setUp(self) set_setup_default("probes_db") + set_setup_default("probes_allowed_groups", [re.compile(".*")]) self.datastore = None Bcfg2.Server.Cache.expire("Probes") def tearDown(self): + Bcfg2.Server.Cache.expire("Probes") if self.datastore is not None: shutil.rmtree(self.datastore) self.datastore = None @@ -275,8 +278,34 @@ group: group:with:colons def test_probes_db(self): """ Set and retrieve probe data with database enabled """ Bcfg2.Options.setup.probes_db = True + syncdb(TestProbesDB) self._perform_tests() + def test_allowed_cgroups(self): + """ Test option to only allow probes to set certain groups """ + probes = self.get_obj() + + test_text = """a couple lines +of freeform text +""" + test_groups = ["group", "group2", "group-with-dashes"] + test_probe_data = lxml.etree.Element("Probe", name="test") + test_probe_data.text = test_text + for group in test_groups: + test_probe_data.text += "group:%s\n" % group + + client = Mock() + groups, data = probes.ReceiveDataItem(client, test_probe_data) + self.assertItemsEqual(groups, test_groups) + self.assertEqual(data, test_text) + + old_allowed_groups = Bcfg2.Options.setup.probes_allowed_groups + Bcfg2.Options.setup.probes_allowed_groups = [re.compile(r'^group.?$')] + groups, data = probes.ReceiveDataItem(client, test_probe_data) + self.assertItemsEqual(groups, ['group', 'group2']) + self.assertEqual(data, test_text) + Bcfg2.Options.setup.probes_allowed_groups = old_allowed_groups + def _perform_tests(self): p = self.get_obj() |