python/samba/tests/samba_tool/base.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Sean Dague <sdague@linux.vnet.ibm.com> 2011
   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 # This provides a wrapper around the cmd interface so that tests can
  18 # easily be built on top of it and have minimal code to run basic tests
  19 # of the commands. A list of the environmental variables can be found in
  20 # ~/selftest/selftest.pl
  21 #
  22 # These can all be accessed via os.environ["VARIABLENAME"] when needed
  23
  24 import os
  25 import sys
  26 import random
  27 import string
  28 from io import StringIO
  29
  30 import samba.getopt as options
  31 import samba.tests
  32 from samba.auth import system_session
  33 from samba.getopt import OptionParser
  34 from samba.netcmd.main import cmd_sambatool
  35 from samba.samdb import SamDB
  36
  37
  38 def truncate_string(s, cutoff=100):
  39     if len(s) < cutoff + 15:
  40         return s
  41     return s[:cutoff] + '[%d more characters]' % (len(s) - cutoff)
  42
  43
  44 class SambaToolCmdTest(samba.tests.BlackboxTestCase):
  45     # Use a class level reference to StringIO, which subclasses can
  46     # override if they need to (to e.g. add a lying isatty() method).
  47     stringIO = StringIO
  48
  49     @staticmethod
  50     def getSamDB(*argv):
  51         """a convenience function to get a samdb instance so that we can query it"""
  52
  53         parser = OptionParser()
  54         sambaopts = options.SambaOptions(parser)
  55         credopts = options.CredentialsOptions(parser)
  56         hostopts = options.HostOptions(parser)
  57         parser.parse_args(list(argv))
  58
  59         lp = sambaopts.get_loadparm()
  60         creds = credopts.get_credentials(lp, fallback_machine=True)
  61
  62         return SamDB(url=hostopts.H, session_info=system_session(),
  63                      credentials=creds, lp=lp)
  64
  65     @classmethod
  66     def _run(cls, *argv, verbose=False):
  67         """run a samba-tool command.
  68
  69         positional arguments are effectively what gets passed to
  70         bin/samba-tool.
  71
  72         Add verbose=True during development to see the expanded
  73         command and results.
  74         """
  75         cmd, args = cmd_sambatool()._resolve('samba-tool', *argv,
  76                                              outf=cls.stringIO(),
  77                                              errf=cls.stringIO())
  78         result = cmd._run(*args)
  79         out = cmd.outf.getvalue()
  80         err = cmd.errf.getvalue()
  81
  82         if verbose:
  83             print(f"bin/samba-tool {' '.join(argv)}\n\nstdout:\n"
  84                   f"{out}\n\nstderr:\n{err}\nresult: {result}\n",
  85                   file=sys.stderr)
  86
  87         return (result, out, err)
  88
  89     runcmd = _run
  90     runsubcmd = _run
  91
  92     def runsublevelcmd(self, name, sublevels, *args):
  93         """run a command with any number of sub command levels"""
  94         # This is a weird and clunky interface for running a
  95         # subcommand. Use self.runcmd() instead.
  96         return self._run(name, *sublevels, *args)
  97
  98     def assertCmdSuccess(self, exit, out, err, msg=""):
  99         # Make sure we allow '\n]\n' in stdout and stderr
 100         # without causing problems with the subunit protocol.
 101         # We just inject a space...
 102         msg = "exit[%s] stdout[%s] stderr[%s]: %s" % (exit, out, err, msg)
 103         self.assertIsNone(exit, msg=msg.replace("\n]\n", "\n] \n"))
 104
 105     def assertCmdFail(self, val, msg=""):
 106         self.assertIsNotNone(val, msg)
 107
 108     def assertMatch(self, base, string, msg=None):
 109         # Note: we should stop doing this and just use self.assertIn()
 110         if msg is None:
 111             msg = "%r is not in %r" % (truncate_string(string),
 112                                        truncate_string(base))
 113         self.assertIn(string, base, msg)
 114
 115     def randomName(self, count=8):
 116         """Create a random name, cap letters and numbers, and always starting with a letter"""
 117         name = random.choice(string.ascii_uppercase)
 118         name += ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for x in range(count - 1))
 119         return name
 120
 121     def randomXid(self):
 122         # pick some unused, high UID/GID range to avoid interference
 123         # from the system the test runs on
 124
 125         # initialize a list to store used IDs
 126         try:
 127             self.used_xids
 128         except AttributeError:
 129             self.used_xids = []
 130
 131         # try to get an unused ID
 132         failed = 0
 133         while failed < 50:
 134             xid = random.randint(4711000, 4799000)
 135             if xid not in self.used_xids:
 136                 self.used_xids += [xid]
 137                 return xid
 138             failed += 1
 139         assert False, "No Xid are available"
 140
 141     def assertWithin(self, val1, val2, delta, msg=""):
 142         """Assert that val1 is within delta of val2, useful for time computations"""
 143         self.assertTrue(((val1 + delta) > val2) and ((val1 - delta) < val2), msg)
 144
 145     def cleanup_join(self, netbios_name):
 146         (result, out, err) \
 147             = self.runsubcmd("domain",
 148                              "demote",
 149                              ("--remove-other-dead-server=%s " % netbios_name),
 150                              ("-U%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"])),
 151                              ("--server=%s" % os.environ["SERVER"]))
 152
 153         self.assertCmdSuccess(result, out, err)