JezK
Edit File: visualize_drs.py
# -*- coding: utf-8 -*- # Originally based on tests for samba.kcc.ldif_import_export. # Copyright (C) Andrew Bartlett 2015, 2018 # # by Douglas Bagnall <douglas.bagnall@catalyst.net.nz> # # 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/>. # """Tests for samba-tool visualize using the vampire DC and promoted DC environments. For most tests we assume we can't assert much about what state they are in, so we mainly check for command failure, but for others we try to grasp control of replication and make more specific assertions. """ import os import re import json import random import subprocess from samba.tests.samba_tool.base import SambaToolCmdTest VERBOSE = False ENV_DSAS = { 'promoted_dc': ['CN=PROMOTEDVDC,CN=Servers,CN=Default-First-Site-Name,CN=Sites,CN=Configuration,DC=samba,DC=example,DC=com', 'CN=LOCALDC,CN=Servers,CN=Default-First-Site-Name,CN=Sites,CN=Configuration,DC=samba,DC=example,DC=com'], 'vampire_dc': ['CN=LOCALDC,CN=Servers,CN=Default-First-Site-Name,CN=Sites,CN=Configuration,DC=samba,DC=example,DC=com', 'CN=LOCALVAMPIREDC,CN=Servers,CN=Default-First-Site-Name,CN=Sites,CN=Configuration,DC=samba,DC=example,DC=com'], } PARTITION_NAMES = [ "DOMAIN", "CONFIGURATION", "SCHEMA", "DNSDOMAIN", "DNSFOREST", ] def adjust_cmd_for_py_version(parts): if os.getenv("PYTHON", None): parts.insert(0, os.environ["PYTHON"]) return parts def set_auto_replication(dc, allow): credstring = '-U%s%%%s' % (os.environ["USERNAME"], os.environ["PASSWORD"]) on_or_off = '-' if allow else '+' for opt in ['DISABLE_INBOUND_REPL', 'DISABLE_OUTBOUND_REPL']: cmd = adjust_cmd_for_py_version(['bin/samba-tool', 'drs', 'options', credstring, dc, "--dsa-option=%s%s" % (on_or_off, opt)]) subprocess.check_call(cmd) def force_replication(src, dest, base): credstring = '-U%s%%%s' % (os.environ["USERNAME"], os.environ["PASSWORD"]) cmd = adjust_cmd_for_py_version(['bin/samba-tool', 'drs', 'replicate', dest, src, base, credstring, '--sync-forced']) subprocess.check_call(cmd) def get_utf8_matrix(s): # parse the graphical table *just* well enough for our tests # decolourise first s = re.sub("\033" r"\[[^m]+m", '', s) lines = s.split('\n') # matrix rows have '·' on the diagonal rows = [x.strip().replace('·', '0') for x in lines if '·' in x] names = [] values = [] for r in rows: parts = r.rsplit(None, len(rows)) k, v = parts[0], parts[1:] # we want the FOO in 'CN=FOO+' or 'CN=FOO,CN=x,DC=...' k = re.match(r'cn=([^+,]+)', k.lower()).group(1) names.append(k) if len(v) == 1: # this is a single-digit matrix, no spaces v = list(v[0]) values.append([int(x) if x.isdigit() else 1e999 for x in v]) d = {} for n1, row in zip(names, values): d[n1] = {} for n2, v in zip(names, row): d[n1][n2] = v return d class SambaToolVisualizeDrsTest(SambaToolCmdTest): def setUp(self): super(SambaToolVisualizeDrsTest, self).setUp() def test_ntdsconn(self): server = "ldap://%s" % os.environ["SERVER"] creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) (result, out, err) = self.runsubcmd("visualize", "ntdsconn", '-H', server, '-U', creds, '--color=no', '-S') self.assertCmdSuccess(result, out, err) def test_ntdsconn_remote(self): server = "ldap://%s" % os.environ["SERVER"] creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) (result, out, err) = self.runsubcmd("visualize", "ntdsconn", '-H', server, '-U', creds, '--color=no', '-S', '-r') self.assertCmdSuccess(result, out, err) def test_reps(self): server = "ldap://%s" % os.environ["SERVER"] creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) (result, out, err) = self.runsubcmd("visualize", "reps", '-H', server, '-U', creds, '--color=no', '-S') self.assertCmdSuccess(result, out, err) def test_uptodateness_all_partitions(self): creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) dc1 = os.environ["SERVER"] dc2 = os.environ["DC_SERVER"] # We will check that the visualisation works for the two # stopped DCs, but we can't make assertions that the output # will be the same because there may be replication between # the two calls. Stopping the replication on these ones is not # enough because there are other DCs about. (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=no', '-S') self.assertCmdSuccess(result, out, err) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc2, '-U', creds, '--color=no', '-S') self.assertCmdSuccess(result, out, err) def test_uptodateness_partitions(self): creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) dc1 = os.environ["SERVER"] for part in PARTITION_NAMES: (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=no', '-S', '--partition', part) self.assertCmdSuccess(result, out, err) def test_drs_uptodateness(self): """ Test cmd `drs uptodateness` It should print info like this: DNSDOMAIN failure: 4 median: 1.5 maximum: 2 SCHEMA failure: 4 median: 220.0 maximum: 439 DOMAIN failure: 1 median: 25 maximum: 25 CONFIGURATION failure: 1 median: 25 maximum: 25 DNSFOREST failure: 4 median: 1.5 maximum: 2 """ creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) dc1 = os.environ["SERVER"] dc2 = os.environ["DC_SERVER"] for dc in [dc1, dc2]: (result, out, err) = self.runsubcmd("drs", "uptodateness", '-H', "ldap://%s" % dc, '-U', creds) self.assertCmdSuccess(result, out, err) # each partition name should be in output for part_name in PARTITION_NAMES: self.assertIn(part_name, out, msg=out) for line in out.splitlines(): # check keyword in output for attr in ['maximum', 'median', 'failure']: self.assertIn(attr, line) def test_drs_uptodateness_partition(self): """ Test cmd `drs uptodateness --partition DOMAIN` It should print info like this: DOMAIN failure: 1 median: 25 maximum: 25 """ creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) dc1 = os.environ["SERVER"] dc2 = os.environ["DC_SERVER"] for dc in [dc1, dc2]: (result, out, err) = self.runsubcmd("drs", "uptodateness", '-H', "ldap://%s" % dc, '-U', creds, '--partition', 'DOMAIN') self.assertCmdSuccess(result, out, err) lines = out.splitlines() self.assertEqual(len(lines), 1) line = lines[0] self.assertTrue(line.startswith('DOMAIN')) def test_drs_uptodateness_json(self): """ Test cmd `drs uptodateness --json` Example output: { "DNSDOMAIN": { "failure": 0, "median": 0.0, "maximum": 0 }, ... "SCHEMA": { "failure": 0, "median": 0.0, "maximum": 0 } } """ creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) dc1 = os.environ["SERVER"] dc2 = os.environ["DC_SERVER"] for dc in [dc1, dc2]: (result, out, err) = self.runsubcmd("drs", "uptodateness", '-H', "ldap://%s" % dc, '-U', creds, '--json') self.assertCmdSuccess(result, out, err) # should be json format obj = json.loads(out) # each partition name should be in json obj for part_name in PARTITION_NAMES: self.assertIn(part_name, obj) summary_obj = obj[part_name] for attr in ['maximum', 'median', 'failure']: self.assertIn(attr, summary_obj) def test_drs_uptodateness_json_median(self): """ Test cmd `drs uptodateness --json --median` drs uptodateness --json --median { "DNSDOMAIN": { "median": 0.0 }, ... "SCHEMA": { "median": 0.0 } } """ creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) dc1 = os.environ["SERVER"] dc2 = os.environ["DC_SERVER"] for dc in [dc1, dc2]: (result, out, err) = self.runsubcmd("drs", "uptodateness", '-H', "ldap://%s" % dc, '-U', creds, '--json', '--median') self.assertCmdSuccess(result, out, err) # should be json format obj = json.loads(out) # each partition name should be in json obj for part_name in PARTITION_NAMES: self.assertIn(part_name, obj) summary_obj = obj[part_name] self.assertIn('median', summary_obj) self.assertNotIn('maximum', summary_obj) self.assertNotIn('failure', summary_obj) def assert_matrix_validity(self, matrix, dcs=()): for dc in dcs: self.assertIn(dc, matrix) for k, row in matrix.items(): self.assertEqual(row[k], 0) def test_uptodateness_stop_replication_domain(self): creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) dc1 = os.environ["SERVER"] dc2 = os.environ["DC_SERVER"] self.addCleanup(set_auto_replication, dc1, True) self.addCleanup(set_auto_replication, dc2, True) def display(heading, out): if VERBOSE: print("========", heading, "=========") print(out) samdb1 = self.getSamDB("-H", "ldap://%s" % dc1, "-U", creds) samdb2 = self.getSamDB("-H", "ldap://%s" % dc2, "-U", creds) domain_dn = samdb1.domain_dn() self.assertTrue(domain_dn == samdb2.domain_dn(), "We expected the same domain_dn across DCs") ou1 = "OU=dc1.%x,%s" % (random.randrange(1 << 64), domain_dn) ou2 = "OU=dc2.%x,%s" % (random.randrange(1 << 64), domain_dn) samdb1.add({ "dn": ou1, "objectclass": "organizationalUnit" }) samdb2.add({ "dn": ou2, "objectclass": "organizationalUnit" }) set_auto_replication(dc1, False) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("dc1 replication is now off", out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) force_replication(dc2, dc1, domain_dn) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("forced replication %s -> %s" % (dc2, dc1), out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) self.assertEqual(matrix[dc1][dc2], 0) force_replication(dc1, dc2, domain_dn) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("forced replication %s -> %s" % (dc2, dc1), out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) self.assertEqual(matrix[dc2][dc1], 0) dn1 = 'cn=u1.%%d,%s' % (ou1) dn2 = 'cn=u2.%%d,%s' % (ou2) for i in range(10): samdb1.add({ "dn": dn1 % i, "objectclass": "user" }) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("added 10 users on %s" % dc1, out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) # dc2's view of dc1 should now be 10 changes out of date self.assertEqual(matrix[dc2][dc1], 10) for i in range(10): samdb2.add({ "dn": dn2 % i, "objectclass": "user" }) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("added 10 users on %s" % dc2, out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) # dc1's view of dc2 is probably 11 changes out of date self.assertGreaterEqual(matrix[dc1][dc2], 10) for i in range(10, 101): samdb1.add({ "dn": dn1 % i, "objectclass": "user" }) samdb2.add({ "dn": dn2 % i, "objectclass": "user" }) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("added 91 users on both", out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) # the difference here should be ~101. self.assertGreaterEqual(matrix[dc1][dc2], 100) self.assertGreaterEqual(matrix[dc2][dc1], 100) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN', '--max-digits', '2') display("with --max-digits 2", out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) # visualising with 2 digits mean these overflow into infinity self.assertGreaterEqual(matrix[dc1][dc2], 1e99) self.assertGreaterEqual(matrix[dc2][dc1], 1e99) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN', '--max-digits', '1') display("with --max-digits 1", out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) # visualising with 1 digit means these overflow into infinity self.assertGreaterEqual(matrix[dc1][dc2], 1e99) self.assertGreaterEqual(matrix[dc2][dc1], 1e99) force_replication(dc2, dc1, samdb1.domain_dn()) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("forced replication %s -> %s" % (dc2, dc1), out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) self.assertEqual(matrix[dc1][dc2], 0) force_replication(dc1, dc2, samdb2.domain_dn()) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("forced replication %s -> %s" % (dc1, dc2), out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) self.assertEqual(matrix[dc2][dc1], 0) samdb1.delete(ou1, ['tree_delete:1']) samdb2.delete(ou2, ['tree_delete:1']) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("tree delete both ous on %s" % (dc1,), out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) self.assertGreaterEqual(matrix[dc1][dc2], 100) self.assertGreaterEqual(matrix[dc2][dc1], 100) set_auto_replication(dc1, True) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("replication is now on", out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) # We can't assert actual values after this because # auto-replication is on and things will change underneath us. (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc2, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("%s's view" % dc2, out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) force_replication(dc1, dc2, samdb2.domain_dn()) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("forced replication %s -> %s" % (dc1, dc2), out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) force_replication(dc2, dc1, samdb2.domain_dn()) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc1, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("forced replication %s -> %s" % (dc2, dc1), out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) (result, out, err) = self.runsubcmd("visualize", "uptodateness", "-r", '-H', "ldap://%s" % dc2, '-U', creds, '--color=yes', '--utf8', '-S', '--partition', 'DOMAIN') display("%s's view" % dc2, out) self.assertCmdSuccess(result, out, err) matrix = get_utf8_matrix(out) self.assert_matrix_validity(matrix, [dc1, dc2]) def test_reps_remote(self): server = "ldap://%s" % os.environ["SERVER"] creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) (result, out, err) = self.runsubcmd("visualize", "reps", '-H', server, '-U', creds, '--color=no', '-S', '-r') self.assertCmdSuccess(result, out, err) def test_ntdsconn_dot(self): server = "ldap://%s" % os.environ["SERVER"] creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) (result, out, err) = self.runsubcmd("visualize", "ntdsconn", '-H', server, '-U', creds, '--dot', '--color=no', '-S') self.assertCmdSuccess(result, out, err) def test_ntdsconn_remote_dot(self): server = "ldap://%s" % os.environ["SERVER"] creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) (result, out, err) = self.runsubcmd("visualize", "ntdsconn", '-H', server, '-U', creds, '--dot', '--color=no', '-S', '-r') self.assertCmdSuccess(result, out, err) def test_reps_dot(self): server = "ldap://%s" % os.environ["SERVER"] creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) (result, out, err) = self.runsubcmd("visualize", "reps", '-H', server, '-U', creds, '--dot', '--color=no', '-S') self.assertCmdSuccess(result, out, err) def test_reps_remote_dot(self): server = "ldap://%s" % os.environ["SERVER"] creds = "%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]) (result, out, err) = self.runsubcmd("visualize", "reps", '-H', server, '-U', creds, '--dot', '--color=no', '-S', '-r') self.assertCmdSuccess(result, out, err)