JezK
Edit File: logfiles.py
# Unix SMB/CIFS implementation. # # Copyright (C) Catalyst.Net Ltd. 2022 # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import subprocess import os from samba.tests import TestCaseInTempDir from pprint import pprint HERE = os.path.dirname(__file__) S4_SERVER = os.path.join(HERE, '../../../../bin/test_s4_logging') S3_SERVER = os.path.join(HERE, '../../../../bin/test_s3_logging') CLASS_LIST = ["all", "tdb", "printdrivers", "lanman", "smb", "rpc_parse", "rpc_srv", "rpc_cli", "passdb", "sam", "auth", "winbind", "vfs", "idmap", "quota", "acls", "locking", "msdfs", "dmapi", "registry", "scavenger", "dns", "ldb", "tevent", "auth_audit", "auth_json_audit", "kerberos", "drs_repl", "smb2", "smb2_credits", "dsdb_audit", "dsdb_json_audit", "dsdb_password_audit", "dsdb_password_json_audit", "dsdb_transaction_audit", "dsdb_transaction_json_audit", "dsdb_group_audit", "dsdb_group_json_audit"] CLASS_CODES = {k: i for i, k in enumerate(CLASS_LIST)} class S4LoggingTests(TestCaseInTempDir): server = S4_SERVER def _write_smb_conf(self, default_level=2, default_file="default", mapping=()): self.smbconf = os.path.join(self.tempdir, "smb.conf") with open(self.smbconf, "w") as f: f.write('[global]\n') if default_file is not None: dest = os.path.join(self.tempdir, default_file) f.write(f" log file = {dest}\n") f.write(" log level = ") if default_level: f.write(f"{default_level}") for dbg_class, log_level, log_file in mapping: f.write(' ') f.write(dbg_class) if log_level is not None: f.write(f':{log_level}') if log_file is not None: dest = os.path.join(self.tempdir, log_file) f.write(f'@{dest}') f.write('\n') self.addCleanup(os.unlink, self.smbconf) def _extract_log_level_line(self, new_level=2): # extricate the 'log level' line from the smb.conf, returning # the value, and replacing the log level line with something # innocuous. smbconf2 = self.smbconf + 'new' with open(self.smbconf) as f: with open(smbconf2, 'w') as f2: for line in f: if 'log level' in line: debug_arg = line.split('=', 1)[1].strip() if new_level is not None: f2.write(f' log level = {new_level}\n') else: f2.write(line) os.replace(smbconf2, self.smbconf) return debug_arg def _get_expected_strings(self, mapping, level_filter, default_file='default', file_filter=None): default = os.path.join(self.tempdir, default_file) expected = {default: []} # this kind of thing: # " logging for 'dns' [21], at level 4" for dbg_class, log_level, log_file in mapping: if log_file is None: log_file = default_file f = os.path.join(self.tempdir, log_file) expected.setdefault(f, []) if log_level < level_filter: continue if file_filter not in (None, log_file): continue s = (f" logging for '{dbg_class}' [{CLASS_CODES[dbg_class]}], " f"at level {level_filter}") expected[f].append(s) return expected def _run_s4_logger(self, log_level, *extra_args): cmd = [self.server, '-s', self.smbconf, '-L', str(log_level), *extra_args] p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) self.assertEqual(p.returncode, 0, f"'{' '.join(cmd)}' failed ({p.returncode})") return p.stdout.decode(), p.stderr.decode() def assert_string_contains(self, string, expected_lines, filename=None): expected_lines = set(expected_lines) string_lines = set(string.split('\n')) present_lines = string_lines & expected_lines if present_lines != expected_lines: if filename: print(filename) print("expected %d lines, found %d" % (len(expected_lines), len(present_lines))) print("missing lines:") pprint(expected_lines - present_lines) raise AssertionError("missing lines") def assert_file_contains(self, filename, expected_lines): with open(filename) as f: string = f.read() self.assert_string_contains(string, expected_lines, filename) def assert_n_known_lines_string(self, string, n): count = string.count("logging for '") if count != n: raise AssertionError( f"string has {count} lines, expected {n}") def assert_n_known_lines(self, filename, n): with open(filename) as f: string = f.read() count = string.count(" logging for '") if count != n: raise AssertionError( f"{filename} has {count} lines, expected {n}") def assert_unlink_expected_strings(self, expected_strings): for k, v in expected_strings.items(): if not os.path.exists(k): self.fail(f"{k} does not exist") self.assert_file_contains(k, v) self.assert_n_known_lines(k, len(v)) os.unlink(k) def test_each_to_its_own(self): level = 4 mapping = [(x, level, x) for x in CLASS_LIST] expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=mapping) stdout, stderr = self._run_s4_logger(level) self.assert_unlink_expected_strings(expected_strings) def test_all_to_one(self): level = 4 dest = 'everything' mapping = [(x, level, dest) for x in CLASS_LIST] expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=mapping) stdout, stderr = self._run_s4_logger(level) self.assert_unlink_expected_strings(expected_strings) def test_bifurcate(self): level = 4 dests = ['even', 'odd'] mapping = [(x, level + 1, dests[i & 1]) for i, x in enumerate(CLASS_LIST)] expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=mapping) stdout, stderr = self._run_s4_logger(level) self.assert_unlink_expected_strings(expected_strings) def test_bifurcate_level_out_of_range(self): # nothing will be logged, because we're logging at a too high # level. level = 4 dests = ['even', 'odd'] mapping = [(x, level - 1, dests[i & 1]) for i, x in enumerate(CLASS_LIST)] expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=mapping) stdout, stderr = self._run_s4_logger(level) self.assert_unlink_expected_strings(expected_strings) def test_bifurcate_misc_log_level(self): # We are sending even numbers to default and odd numbers to # 'odd', at various levels, depending on mod 3. Like this: # # log level = 2 all:5 \ # tdb:4@odd \ # printdrivers:3 \ # lanman:5@odd \ # smb:4 \ # rpc_parse:3@odd \ # rpc_srv:5 ... # # Therefore, 'default' should get classes that are (0 or 4) % 6 # and 'odd' should get classes that are (1 or 3) % 6. level = 4 dests = [None, 'odd'] mapping = [] for i, x in enumerate(CLASS_LIST): parity = i & 1 log_level = level + 1 - (i % 3) mapping.append((x, log_level, dests[parity])) expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=mapping) stdout, stderr = self._run_s4_logger(level) self.assert_unlink_expected_strings(expected_strings) def test_all_different_ways_cmdline_d(self): level = 4 dests = [None, 'a', 'b', 'c'] mapping = [] seed = 123 for i, x in enumerate(CLASS_LIST): d = seed & 3 seed = seed * 17 + 1 log_level = seed % 10 seed &= 0xff mapping.append((x, log_level, dests[d])) expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=mapping) debug_arg = self._extract_log_level_line(26) stdout, stderr = self._run_s4_logger(level, '-d', debug_arg) self.assert_unlink_expected_strings(expected_strings) def test_all_different_ways_cmdline_d_interactive(self): level = 4 dests = [None, 'a', 'b', 'c'] mapping = [] seed = 1234 for i, x in enumerate(CLASS_LIST): d = seed & 3 seed = seed * 13 + 1 log_level = seed % 10 seed &= 0xff mapping.append((x, log_level, dests[d])) expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=mapping) debug_arg = self._extract_log_level_line(None) stdout, stderr = self._run_s4_logger(level, '-d', debug_arg, '-i') expected_lines = [] for v in expected_strings.values(): # stderr doesn't end up with leading ' ' expected_lines.extend([x.strip() for x in v]) self.assert_string_contains(stderr, expected_lines) self.assert_n_known_lines_string(stderr, len(expected_lines)) def test_only_some_level_0(self): # running the logger with -L 0 makes the log messages run at # level 0 (i.e DBG_ERR), so we always see them in default, # even though smb.conf doesn't ask. mapping = [(x, 3, ['default', 'bees']['b' in x]) for x in CLASS_LIST] expected_strings = self._get_expected_strings(mapping, 0) self._write_smb_conf(mapping=[x for x in mapping if x[2] == 'bees']) stdout, stderr = self._run_s4_logger(0) self.assert_unlink_expected_strings(expected_strings) def test_only_some_level_3(self): # here, we're expecting the unmentioned non-b classes to just # disappear. level = 3 mapping = [(x, level, 'bees') for x in CLASS_LIST if 'b' in x] expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=[x for x in mapping if x[2] == 'bees']) stdout, stderr = self._run_s4_logger(level) self.assert_unlink_expected_strings(expected_strings) def test_none(self): level = 4 mapping = [] expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=mapping) stdout, stderr = self._run_s4_logger(level) self.assert_unlink_expected_strings(expected_strings) def test_none_high_default(self): # We set the default level to 5 and do nothing else special, # which means we need a different mapping for the smb.conf # than the expected strings. level = 4 mapping = [(x, 5, 'default') for x in CLASS_LIST] expected_strings = self._get_expected_strings(mapping, level) # note the empty mapping in smb.conf self._write_smb_conf(mapping=[], default_level=5) stdout, stderr = self._run_s4_logger(level) self.assert_unlink_expected_strings(expected_strings) def test_none_high_cmdline_d(self): # We set the default level to 2, but run the 'server' with -d 10. level = 4 mapping = [(x, 10, 'default') for x in CLASS_LIST] expected_strings = self._get_expected_strings(mapping, level) # note the empty mapping in smb.conf self._write_smb_conf(mapping=[]) stdout, stderr = self._run_s4_logger(level, '-d', '10') self.assert_unlink_expected_strings(expected_strings) def test_interactive_high_default_simple(self): # running with -i should send everything to stderr. level = 4 mapping = [(x, 5, 'default') for x in CLASS_LIST] expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=[], default_level=5) stdout, stderr = self._run_s4_logger(level, '-i') expected_lines = [] for v in expected_strings.values(): # stderr doesn't end up with leading ' ' expected_lines.extend([x.strip() for x in v]) self.assert_string_contains(stderr, expected_lines) def test_interactive_complex_smb_conf(self): # running with -i should send everything to stderr. The # smb.conf will set the levels, but the target files are # overridden. # (this is the test_bifurcate_misc_log_level() smb.conf). level = 4 dests = [None, 'odd'] mapping = [] for i, x in enumerate(CLASS_LIST): parity = i & 1 log_level = level + 1 - (i % 3) mapping.append((x, log_level, dests[parity])) expected_strings = self._get_expected_strings(mapping, level) self._write_smb_conf(mapping=mapping) stdout, stderr = self._run_s4_logger(level, '-i') expected_lines = [] for v in expected_strings.values(): # stderr doesn't end up with leading ' ' expected_lines.extend([x.strip() for x in v]) self.assert_string_contains(stderr, expected_lines) class S3LoggingTests(S4LoggingTests): server = S3_SERVER # These tests were developed for testing the test_logger when # linked against CMDLINE_S4 (see lib/util/wscript_build), but can # also run when linked against CMDLINE_S3.