python/samba/tests/samba_tool/join_lmdb_size.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Catalyst IT Ltd. 2019
   3 #
   4 # This program is free software; you can redistribute it and/or modify
   5 # it under the terms of the GNU General Public License as published by
   6 # the Free Software Foundation; either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 from samba.tests.samba_tool.base import SambaToolCmdTest
  19 import os
  20 import shutil
  21
  22
  23 class JoinLmdbSizeTestCase(SambaToolCmdTest):
  24     """Test setting of the lmdb map size during join"""
  25
  26     def setUp(self):
  27         super().setUp()
  28         self.tempsambadir = os.path.join(self.tempdir, "samba")
  29         os.mkdir(self.tempsambadir)
  30         (_, name) = os.path.split(self.tempdir)
  31         self.netbios_name = name
  32
  33     # join a domain and set the lmdb map size to size
  34     #
  35     # returns the tuple (ret, stdout, stderr)
  36     def join(self, size=None, role=None):
  37         command = (
  38             "samba-tool " +
  39             "domain join " +
  40             os.environ["REALM"] + " " +
  41             role + " " +
  42             ("-U%s%%%s " % (os.environ["USERNAME"], os.environ["PASSWORD"])) +
  43             ("--targetdir=%s " % self.tempsambadir) +
  44             ("--option=netbiosname=%s " % self.netbios_name) +
  45             "--backend-store=mdb "
  46         )
  47         if size:
  48             command += ("--backend-store-size=%s" % size)
  49
  50         (ret, stdout, stderr) = self.run_command(command)
  51         if ret == 0:
  52             self.cleanup_join(self.netbios_name)
  53
  54         return (ret, stdout, stderr)
  55
  56     def is_rodc(self):
  57         url = "ldb://%s/private/sam.ldb" % self.tempsambadir
  58         samdb = self.getSamDB("-H", url)
  59         return samdb.am_rodc()
  60
  61     #
  62     # Get the lmdb map size for the specified command
  63     #
  64     # While there is a python lmdb package available we use the lmdb command
  65     # line utilities to avoid introducing a dependency.
  66     #
  67     def get_lmdb_environment_size(self, path):
  68         (result, out, err) = self.run_command("mdb_stat -ne %s" % path)
  69         if result:
  70             self.fail("Unable to run mdb_stat\n")
  71         for line in out.split("\n"):
  72             line = line.strip()
  73             if line.startswith("Map size:"):
  74                 line = line.replace(" ", "")
  75                 (label, size) = line.split(":")
  76                 return int(size)
  77
  78     #
  79     # Check the lmdb files created by join and ensure that the map size
  80     # has been set to size.
  81     #
  82     # Currently this is all the *.ldb files in private/sam.ldb.d
  83     #
  84     def check_lmdb_environment_sizes(self, size):
  85         directory = os.path.join(self.tempsambadir, "private", "sam.ldb.d")
  86         for name in os.listdir(directory):
  87             if name.endswith(".ldb"):
  88                 path = os.path.join(directory, name)
  89                 s = self.get_lmdb_environment_size(path)
  90                 if s != size:
  91                     self.fail("File %s, size=%d larger than %d" %
  92                               (name, s, size))
  93
  94     #
  95     # Ensure that if --backend-store-size is not specified the default of
  96     # 8Gb is used
  97     def test_join_as_dc_default(self):
  98         (result, out, err) = self.join(role="DC")
  99         self.assertEqual(0, result)
 100         self.check_lmdb_environment_sizes(8 * 1024 * 1024 * 1024)
 101         self.assertFalse(self.is_rodc())
 102
 103     #
 104     # Join as an DC with the lmdb backend size set to 1Gb
 105     def test_join_as_dc(self):
 106         (result, out, err) = self.join("1Gb", "DC")
 107         self.assertEqual(0, result)
 108         self.check_lmdb_environment_sizes(1 * 1024 * 1024 * 1024)
 109         self.assertFalse(self.is_rodc())
 110
 111     #
 112     # Join as an RODC with the lmdb backend size set to 128Mb
 113     def test_join_as_rodc(self):
 114         (result, out, err) = self.join("128Mb", "RODC")
 115         self.assertEqual(0, result)
 116         self.check_lmdb_environment_sizes(128 * 1024 * 1024)
 117         self.assertTrue(self.is_rodc())
 118
 119     #
 120     # Join as an RODC with --backend-store-size
 121     def test_join_as_rodc_default(self):
 122         (result, out, err) = self.join(role="RODC")
 123         self.assertEqual(0, result)
 124         self.check_lmdb_environment_sizes(8 * 1024 * 1024 * 1024)
 125         self.assertTrue(self.is_rodc())
 126
 127     def test_no_unit_suffix(self):
 128         (result, out, err) = self.run_command(
 129             'samba-tool domain join --backend-store-size "2"')
 130         self.assertGreater(result, 0)
 131         self.assertRegex(err,
 132                          r"--backend-store-size invalid suffix ''")
 133
 134     def test_invalid_unit_suffix(self):
 135         (result, out, err) = self.run_command(
 136             'samba-tool domain join --backend-store-size "2 cd"')
 137         self.assertGreater(result, 0)
 138         self.assertRegex(err,
 139                          r"--backend-store-size invalid suffix 'cd'")
 140
 141     def test_non_numeric(self):
 142         (result, out, err) = self.run_command(
 143             'samba-tool domain join --backend-store-size "two Gb"')
 144         self.assertGreater(result, 0)
 145         self.assertRegex(
 146             err,
 147             r"backend-store-size option requires a numeric value, with an"
 148             " optional unit suffix")
 149
 150     def tearDown(self):
 151         super().tearDown()
 152         shutil.rmtree(self.tempsambadir)