← Back to branch summary

~cloud-init-dev/cloud-init/trunk

« back to all changes in this revision

Viewing changes to tests/unittests/test_handler/test_handler_rsyslog.py

  • Committer: Scott Moser
  • Date: 2016-08-10 15:06:15 UTC
  • Revision ID: smoser@ubuntu.com-20160810150615-ma2fv107w3suy1ma
README: Mention move of revision control to git.

cloud-init development has moved its revision control to git.
It is available at 
  https://code.launchpad.net/cloud-init

Clone with 
  git clone https://git.launchpad.net/cloud-init
or
  git clone git+ssh://git.launchpad.net/cloud-init

For more information see
  https://git.launchpad.net/cloud-init/tree/HACKING.rst

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/unittests/test_handler/test_handler_rsyslog.py
1
 
import os
2
 
import shutil
3
 
import tempfile
4
 
 
5
 
from cloudinit.config.cc_rsyslog import (
6
 
    apply_rsyslog_changes, DEF_DIR, DEF_FILENAME, DEF_RELOAD, load_config,
7
 
    parse_remotes_line, remotes_to_rsyslog_cfg)
8
 
from cloudinit import util
9
 
 
10
 
from .. import helpers as t_help
11
 
 
12
 
 
13
 
class TestLoadConfig(t_help.TestCase):
14
 
    def setUp(self):
15
 
        super(TestLoadConfig, self).setUp()
16
 
        self.basecfg = {
17
 
            'config_filename': DEF_FILENAME,
18
 
            'config_dir': DEF_DIR,
19
 
            'service_reload_command': DEF_RELOAD,
20
 
            'configs': [],
21
 
            'remotes': {},
22
 
        }
23
 
 
24
 
    def test_legacy_full(self):
25
 
        found = load_config({
26
 
            'rsyslog': ['*.* @192.168.1.1'],
27
 
            'rsyslog_dir': "mydir",
28
 
            'rsyslog_filename': "myfilename"})
29
 
        self.basecfg.update({
30
 
            'configs': ['*.* @192.168.1.1'],
31
 
            'config_dir': "mydir",
32
 
            'config_filename': 'myfilename',
33
 
            'service_reload_command': 'auto'}
34
 
        )
35
 
 
36
 
        self.assertEqual(found, self.basecfg)
37
 
 
38
 
    def test_legacy_defaults(self):
39
 
        found = load_config({
40
 
            'rsyslog': ['*.* @192.168.1.1']})
41
 
        self.basecfg.update({
42
 
            'configs': ['*.* @192.168.1.1']})
43
 
        self.assertEqual(found, self.basecfg)
44
 
 
45
 
    def test_new_defaults(self):
46
 
        self.assertEqual(load_config({}), self.basecfg)
47
 
 
48
 
    def test_new_configs(self):
49
 
        cfgs = ['*.* myhost', '*.* my2host']
50
 
        self.basecfg.update({'configs': cfgs})
51
 
        self.assertEqual(
52
 
            load_config({'rsyslog': {'configs': cfgs}}),
53
 
            self.basecfg)
54
 
 
55
 
 
56
 
class TestApplyChanges(t_help.TestCase):
57
 
    def setUp(self):
58
 
        self.tmp = tempfile.mkdtemp()
59
 
        self.addCleanup(shutil.rmtree, self.tmp)
60
 
 
61
 
    def test_simple(self):
62
 
        cfgline = "*.* foohost"
63
 
        changed = apply_rsyslog_changes(
64
 
            configs=[cfgline], def_fname="foo.cfg", cfg_dir=self.tmp)
65
 
 
66
 
        fname = os.path.join(self.tmp, "foo.cfg")
67
 
        self.assertEqual([fname], changed)
68
 
        self.assertEqual(
69
 
            util.load_file(fname), cfgline + "\n")
70
 
 
71
 
    def test_multiple_files(self):
72
 
        configs = [
73
 
            '*.* foohost',
74
 
            {'content': 'abc', 'filename': 'my.cfg'},
75
 
            {'content': 'filefoo-content',
76
 
             'filename': os.path.join(self.tmp, 'mydir/mycfg')},
77
 
        ]
78
 
 
79
 
        changed = apply_rsyslog_changes(
80
 
            configs=configs, def_fname="default.cfg", cfg_dir=self.tmp)
81
 
 
82
 
        expected = [
83
 
            (os.path.join(self.tmp, "default.cfg"),
84
 
             "*.* foohost\n"),
85
 
            (os.path.join(self.tmp, "my.cfg"), "abc\n"),
86
 
            (os.path.join(self.tmp, "mydir/mycfg"), "filefoo-content\n"),
87
 
        ]
88
 
        self.assertEqual([f[0] for f in expected], changed)
89
 
        actual = []
90
 
        for fname, _content in expected:
91
 
            util.load_file(fname)
92
 
            actual.append((fname, util.load_file(fname),))
93
 
        self.assertEqual(expected, actual)
94
 
 
95
 
    def test_repeat_def(self):
96
 
        configs = ['*.* foohost', "*.warn otherhost"]
97
 
 
98
 
        changed = apply_rsyslog_changes(
99
 
            configs=configs, def_fname="default.cfg", cfg_dir=self.tmp)
100
 
 
101
 
        fname = os.path.join(self.tmp, "default.cfg")
102
 
        self.assertEqual([fname], changed)
103
 
 
104
 
        expected_content = '\n'.join([c for c in configs]) + '\n'
105
 
        found_content = util.load_file(fname)
106
 
        self.assertEqual(expected_content, found_content)
107
 
 
108
 
    def test_multiline_content(self):
109
 
        configs = ['line1', 'line2\nline3\n']
110
 
 
111
 
        apply_rsyslog_changes(
112
 
            configs=configs, def_fname="default.cfg", cfg_dir=self.tmp)
113
 
 
114
 
        fname = os.path.join(self.tmp, "default.cfg")
115
 
        expected_content = '\n'.join([c for c in configs])
116
 
        found_content = util.load_file(fname)
117
 
        self.assertEqual(expected_content, found_content)
118
 
 
119
 
 
120
 
class TestParseRemotesLine(t_help.TestCase):
121
 
    def test_valid_port(self):
122
 
        r = parse_remotes_line("foo:9")
123
 
        self.assertEqual(9, r.port)
124
 
 
125
 
    def test_invalid_port(self):
126
 
        with self.assertRaises(ValueError):
127
 
            parse_remotes_line("*.* foo:abc")
128
 
 
129
 
    def test_valid_ipv6(self):
130
 
        r = parse_remotes_line("*.* [::1]")
131
 
        self.assertEqual("*.* @[::1]", str(r))
132
 
 
133
 
    def test_valid_ipv6_with_port(self):
134
 
        r = parse_remotes_line("*.* [::1]:100")
135
 
        self.assertEqual(r.port, 100)
136
 
        self.assertEqual(r.addr, "::1")
137
 
        self.assertEqual("*.* @[::1]:100", str(r))
138
 
 
139
 
    def test_invalid_multiple_colon(self):
140
 
        with self.assertRaises(ValueError):
141
 
            parse_remotes_line("*.* ::1:100")
142
 
 
143
 
    def test_name_in_string(self):
144
 
        r = parse_remotes_line("syslog.host", name="foobar")
145
 
        self.assertEqual("*.* @syslog.host # foobar", str(r))
146
 
 
147
 
 
148
 
class TestRemotesToSyslog(t_help.TestCase):
149
 
    def test_simple(self):
150
 
        # str rendered line must appear in remotes_to_ryslog_cfg return
151
 
        mycfg = "*.* myhost"
152
 
        myline = str(parse_remotes_line(mycfg, name="myname"))
153
 
        r = remotes_to_rsyslog_cfg({'myname': mycfg})
154
 
        lines = r.splitlines()
155
 
        self.assertEqual(1, len(lines))
156
 
        self.assertTrue(myline in r.splitlines())
157
 
 
158
 
    def test_header_footer(self):
159
 
        header = "#foo head"
160
 
        footer = "#foo foot"
161
 
        r = remotes_to_rsyslog_cfg(
162
 
            {'myname': "*.* myhost"}, header=header, footer=footer)
163
 
        lines = r.splitlines()
164
 
        self.assertTrue(header, lines[0])
165
 
        self.assertTrue(footer, lines[-1])
166
 
 
167
 
    def test_with_empty_or_null(self):
168
 
        mycfg = "*.* myhost"
169
 
        myline = str(parse_remotes_line(mycfg, name="myname"))
170
 
        r = remotes_to_rsyslog_cfg(
171
 
            {'myname': mycfg, 'removed': None, 'removed2': ""})
172
 
        lines = r.splitlines()
173
 
        self.assertEqual(1, len(lines))
174
 
        self.assertTrue(myline in r.splitlines())