python/samba/tests/docs.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2007-2012
   3 #
   4 # Tests for documentation.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 3 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 # GNU General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 #
  19
  20 """Tests for presence of documentation."""
  21
  22 import samba
  23 import samba.tests
  24
  25 import os
  26 import subprocess
  27 import xml.etree.ElementTree as ET
  28 import multiprocessing
  29 import concurrent.futures
  30 import tempfile
  31
  32 class TestCase(samba.tests.TestCaseInTempDir):
  33
  34     def _format_message(self, parameters, message):
  35         parameters = list(parameters)
  36         parameters = list(map(str, parameters))
  37         parameters.sort()
  38         return message + '\n\n    %s' % ('\n    '.join(parameters))
  39
  40 def get_max_worker_count():
  41     cpu_count = multiprocessing.cpu_count()
  42
  43     # Always run two processes in parallel
  44     if cpu_count < 2:
  45         return 2
  46
  47     max_workers = int(cpu_count / 2)
  48     if max_workers < 2:
  49         return 2
  50
  51     return max_workers
  52
  53 def check_or_set_smbconf_default(cmdline, topdir, param, default_param):
  54     p = subprocess.Popen(cmdline,
  55                          stdout=subprocess.PIPE,
  56                          stderr=subprocess.PIPE,
  57                          cwd=topdir).communicate()
  58     result = p[0].decode().upper().strip()
  59     if result != default_param.upper():
  60         if not (result == "" and default_param == '""'):
  61             return result, param, default_param
  62
  63     return None
  64
  65 def set_smbconf_arbitrary(cmdline, topdir, param, param_type, value_to_use):
  66     p = subprocess.Popen(cmdline,
  67                          stdout=subprocess.PIPE,
  68                          stderr=subprocess.PIPE,
  69                          cwd=topdir).communicate()
  70     result = p[0].decode().upper().strip()
  71     if result != value_to_use.upper():
  72         # currently no way to distinguish command lists
  73         if param_type == 'list':
  74             if ", ".join(result.split()) == value_to_use.upper():
  75                 return None
  76
  77     # currently no way to identify octal
  78     if param_type == 'integer':
  79         try:
  80             if int(value_to_use, 8) == int(p[0].strip(), 8):
  81                 return None
  82         except:
  83             pass
  84
  85         return result, param, value_to_use
  86
  87     return None
  88
  89 def set_smbconf_arbitrary_opposite(cmdline, topdir, tempdir, section, param,
  90                                    param_type, opposite_value, value_to_use):
  91     g = tempfile.NamedTemporaryFile(mode='w', dir=tempdir, delete=False)
  92     try:
  93         towrite = section + "\n"
  94         towrite += param + " = " + opposite_value
  95         g.write(towrite)
  96     finally:
  97         g.close()
  98
  99     p = subprocess.Popen(cmdline + ["-s", g.name],
 100                          stdout=subprocess.PIPE,
 101                          stderr=subprocess.PIPE,
 102                          cwd=topdir).communicate()
 103     os.unlink(g.name)
 104
 105     # testparm doesn't display a value if they are equivalent
 106     if (value_to_use.lower() != opposite_value.lower()):
 107         for line in p[0].decode().splitlines():
 108             if not line.strip().startswith(param):
 109                 return None
 110
 111             value_found = line.split("=")[1].upper().strip()
 112             if value_found != value_to_use.upper():
 113                 # currently no way to distinguish command lists
 114                 if param_type == 'list':
 115                     if ", ".join(value_found.split()) == value_to_use.upper():
 116                         return None
 117
 118                 # currently no way to identify octal
 119                 if param_type == 'integer':
 120                     try:
 121                         if int(value_to_use, 8) == int(value_found, 8):
 122                             continue
 123                     except:
 124                         pass
 125
 126                     return param, value_to_use, value_found
 127
 128     return None
 129
 130 def get_documented_parameters(sourcedir):
 131     path = os.path.join(sourcedir, "bin", "default", "docs-xml", "smbdotconf")
 132     if not os.path.exists(os.path.join(path, "parameters.all.xml")):
 133         raise Exception("Unable to find parameters.all.xml")
 134     try:
 135         p = open(os.path.join(path, "parameters.all.xml"), 'r')
 136     except IOError as e:
 137         raise Exception("Error opening parameters file")
 138     out = p.read()
 139
 140     root = ET.fromstring(out)
 141     for parameter in root:
 142         name = parameter.attrib.get('name')
 143         if parameter.attrib.get('removed') == "1":
 144             continue
 145         yield name
 146         syn = parameter.findall('synonym')
 147         if syn is not None:
 148             for sy in syn:
 149                 yield sy.text
 150     p.close()
 151
 152
 153 def get_documented_tuples(sourcedir, omit_no_default=True):
 154     path = os.path.join(sourcedir, "bin", "default", "docs-xml", "smbdotconf")
 155     if not os.path.exists(os.path.join(path, "parameters.all.xml")):
 156         raise Exception("Unable to find parameters.all.xml")
 157     try:
 158         p = open(os.path.join(path, "parameters.all.xml"), 'r')
 159     except IOError as e:
 160         raise Exception("Error opening parameters file")
 161     out = p.read()
 162
 163     root = ET.fromstring(out)
 164     for parameter in root:
 165         name = parameter.attrib.get("name")
 166         param_type = parameter.attrib.get("type")
 167         if parameter.attrib.get('removed') == "1":
 168             continue
 169         values = parameter.findall("value")
 170         defaults = []
 171         for value in values:
 172             if value.attrib.get("type") == "default":
 173                 defaults.append(value)
 174
 175         default_text = None
 176         if len(defaults) == 0:
 177             if omit_no_default:
 178                 continue
 179         elif len(defaults) > 1:
 180             raise Exception("More than one default found for parameter %s" % name)
 181         else:
 182             default_text = defaults[0].text
 183
 184         if default_text is None:
 185             default_text = ""
 186         context = parameter.attrib.get("context")
 187         yield name, default_text, context, param_type
 188     p.close()
 189
 190
 191 class SmbDotConfTests(TestCase):
 192
 193     # defines the cases where the defaults may differ from the documentation
 194     #
 195     # Please pass the default via waf rather than adding to this
 196     # list if at all possible.
 197     special_cases = set([
 198         'log level',
 199         'path',
 200         'panic action',
 201         'homedir map',
 202         'NIS homedir',
 203         'server string',
 204         'netbios name',
 205         'socket options',
 206         'ctdbd socket',
 207         'printing',
 208         'printcap name',
 209         'queueresume command',
 210         'queuepause command',
 211         'lpresume command',
 212         'lppause command',
 213         'lprm command',
 214         'lpq command',
 215         'print command',
 216         'template homedir',
 217         'max open files',
 218         'include system krb5 conf',
 219         'smbd max async dosmode',
 220     ])
 221
 222     def setUp(self):
 223         super().setUp()
 224         # create a minimal smb.conf file for testparm
 225         self.smbconf = os.path.join(self.tempdir, "paramtestsmb.conf")
 226         f = open(self.smbconf, 'w')
 227         try:
 228             f.write("""
 229 [test]
 230    path = /
 231 """)
 232         finally:
 233             f.close()
 234
 235         self.blankconf = os.path.join(self.tempdir, "emptytestsmb.conf")
 236         f = open(self.blankconf, 'w')
 237         try:
 238             f.write("")
 239         finally:
 240             f.close()
 241
 242         self.topdir = os.path.abspath(samba.source_tree_topdir())
 243
 244         try:
 245             self.documented = set(get_documented_parameters(self.topdir))
 246         except:
 247             self.fail("Unable to load documented parameters")
 248
 249         try:
 250             self.defaults = set(get_documented_tuples(self.topdir))
 251         except:
 252             self.fail("Unable to load parameters")
 253
 254         try:
 255             self.defaults_all = set(get_documented_tuples(self.topdir, False))
 256         except:
 257             self.fail("Unable to load parameters")
 258
 259     def tearDown(self):
 260         super().tearDown()
 261         os.unlink(self.smbconf)
 262         os.unlink(self.blankconf)
 263
 264     def test_default_s3(self):
 265         self._test_default(['bin/testparm'])
 266         self._set_defaults(['bin/testparm'])
 267
 268         # registry shares appears to need sudo
 269         self._set_arbitrary(['bin/testparm'],
 270             exceptions = ['client lanman auth',
 271                           'client plaintext auth',
 272                           'registry shares',
 273                           'smb ports',
 274                           'rpc server dynamic port range',
 275                           'name resolve order',
 276                           'clustering'])
 277         self._test_empty(['bin/testparm'])
 278
 279     def test_default_s4(self):
 280         self._test_default(['bin/samba-tool', 'testparm'])
 281         self._set_defaults(['bin/samba-tool', 'testparm'])
 282         self._set_arbitrary(['bin/samba-tool', 'testparm'],
 283                             exceptions=['smb ports',
 284                                         'rpc server dynamic port range',
 285                                         'name resolve order'])
 286         self._test_empty(['bin/samba-tool', 'testparm'])
 287
 288     def _test_default(self, program):
 289
 290         if program[0] == 'bin/samba-tool' and os.getenv("PYTHON", None):
 291             program = [os.environ["PYTHON"]] + program
 292
 293         failset = set()
 294
 295         with concurrent.futures.ProcessPoolExecutor(max_workers=get_max_worker_count()) as executor:
 296             result_futures = []
 297
 298             for tuples in self.defaults:
 299                 param, default, context, param_type = tuples
 300
 301                 if param in self.special_cases:
 302                     continue
 303                 # bad, bad parametric options - we don't have their default values
 304                 if ':' in param:
 305                     continue
 306                 section = None
 307                 if context == "G":
 308                     section = "global"
 309                 elif context == "S":
 310                     section = "test"
 311                 else:
 312                     self.fail("%s has no valid context" % param)
 313
 314                 program_arg1 = ["--configfile=%s" % (self.smbconf)]
 315                 if (program[0] == 'bin/testparm'):
 316                     program_arg1 = ["--suppress-prompt", self.smbconf]
 317
 318                 cmdline = program + program_arg1 + [
 319                                      "--section-name",
 320                                      section,
 321                                      "--parameter-name",
 322                                      param]
 323
 324                 future = executor.submit(check_or_set_smbconf_default, cmdline, self.topdir, param, default)
 325                 result_futures.append(future)
 326
 327             for f in concurrent.futures.as_completed(result_futures):
 328                 if f.result():
 329                     result, param, default_param = f.result()
 330
 331                     doc_triple = "%s\n      Expected: %s" % (param, default_param)
 332                     failset.add("%s\n      Got: %s" % (doc_triple, result))
 333
 334         if len(failset) > 0:
 335             self.fail(self._format_message(failset,
 336                                            "Parameters that do not have matching defaults:"))
 337
 338     def _set_defaults(self, program):
 339
 340         if program[0] == 'bin/samba-tool' and os.getenv("PYTHON", None):
 341             program = [os.environ["PYTHON"]] + program
 342
 343         failset = set()
 344
 345         with concurrent.futures.ProcessPoolExecutor(max_workers=get_max_worker_count()) as executor:
 346             result_futures = []
 347
 348             for tuples in self.defaults:
 349                 param, default, context, param_type = tuples
 350
 351                 exceptions = set([
 352                     'printing',
 353                     'smbd max async dosmode',
 354                 ])
 355
 356                 if param in exceptions:
 357                     continue
 358
 359                 section = None
 360                 if context == "G":
 361                     section = "global"
 362                 elif context == "S":
 363                     section = "test"
 364                 else:
 365                     self.fail("%s has no valid context" % param)
 366
 367                 program_arg1 = ["--configfile=%s" % (self.smbconf)]
 368                 if (program[0] == 'bin/testparm'):
 369                     program_arg1 = ["--suppress-prompt", self.smbconf]
 370
 371                 cmdline = program + program_arg1 + [
 372                                      "--section-name",
 373                                      section,
 374                                      "--parameter-name",
 375                                      param,
 376                                      "--option",
 377                                      "%s = %s" % (param, default)]
 378                 future = executor.submit(check_or_set_smbconf_default, cmdline, self.topdir, param, default)
 379                 result_futures.append(future)
 380
 381             for f in concurrent.futures.as_completed(result_futures):
 382                 if f.result():
 383                     result, param, default_param = f.result()
 384
 385                     doc_triple = "%s\n      Expected: %s" % (param, default)
 386                     failset.add("%s\n      Got: %s" % (doc_triple, result))
 387
 388         if len(failset) > 0:
 389             self.fail(self._format_message(failset,
 390                                            "Parameters that do not have matching defaults:"))
 391
 392     def _set_arbitrary(self, program, exceptions=None):
 393
 394         if program[0] == 'bin/samba-tool' and os.getenv("PYTHON", None):
 395             program = [os.environ["PYTHON"]] + program
 396
 397         arbitrary = {'string': 'string', 'boolean': 'yes', 'integer': '5',
 398                      'boolean-rev': 'yes',
 399                      'cmdlist': 'a b c',
 400                      'bytes': '10',
 401                      'octal': '0123',
 402                      'ustring': 'ustring',
 403                      'enum': '', 'boolean-auto': '', 'char': 'a', 'list': 'a, b, c'}
 404         opposite_arbitrary = {'string': 'string2', 'boolean': 'no', 'integer': '6',
 405                               'boolean-rev': 'no',
 406                               'cmdlist': 'd e f',
 407                               'bytes': '11',
 408                               'octal': '0567',
 409                               'ustring': 'ustring2',
 410                               'enum': '', 'boolean-auto': '', 'char': 'b', 'list': 'd, e, f'}
 411
 412         failset = set()
 413
 414         with concurrent.futures.ProcessPoolExecutor(max_workers=get_max_worker_count()) as executor:
 415             result_futures1 = []
 416             result_futures2 = []
 417
 418             for tuples in self.defaults_all:
 419                 param, default, context, param_type = tuples
 420
 421                 if param in ['printing', 'copy', 'include', 'log level']:
 422                     continue
 423
 424                 # currently no easy way to set an arbitrary value for these
 425                 if param_type in ['enum', 'boolean-auto']:
 426                     continue
 427
 428                 if exceptions is not None:
 429                     if param in exceptions:
 430                         continue
 431
 432                 section = None
 433                 if context == "G":
 434                     section = "global"
 435                 elif context == "S":
 436                     section = "test"
 437                 else:
 438                     self.fail("%s has no valid context" % param)
 439
 440                 value_to_use = arbitrary.get(param_type)
 441                 if value_to_use is None:
 442                     self.fail("%s has an invalid type" % param)
 443
 444                 program_arg1 = ["--configfile=%s" % (self.smbconf)]
 445                 if (program[0] == 'bin/testparm'):
 446                     program_arg1 = ["--suppress-prompt", self.smbconf]
 447
 448                 cmdline = program + program_arg1 + [
 449                                      "--section-name",
 450                                      section,
 451                                      "--parameter-name",
 452                                      param,
 453                                      "--option",
 454                                      "%s = %s" % (param, value_to_use)]
 455
 456                 future = executor.submit(set_smbconf_arbitrary, cmdline, self.topdir, param, param_type, value_to_use)
 457                 result_futures1.append(future)
 458
 459                 opposite_value = opposite_arbitrary.get(param_type)
 460
 461                 cmdline = program + ["--suppress-prompt",
 462                                      "--option",
 463                                      "%s = %s" % (param, value_to_use)]
 464
 465                 future = executor.submit(set_smbconf_arbitrary_opposite, cmdline, self.topdir, self.tempdir,
 466                                          section, param, param_type, opposite_value, value_to_use)
 467                 result_futures2.append(future)
 468
 469             for f in concurrent.futures.as_completed(result_futures1):
 470                 if f.result():
 471                     result, param, value_to_use = f.result()
 472
 473                     doc_triple = "%s\n      Expected: %s" % (param, value_to_use)
 474                     failset.add("%s\n      Got: %s" % (doc_triple, result))
 475
 476             for f in concurrent.futures.as_completed(result_futures2):
 477                 if f.result():
 478                     param, value_to_use, value_found = f.result()
 479
 480                     doc_triple = "%s\n      Expected: %s" % (param, value_to_use)
 481                     failset.add("%s\n      Got: %s" % (doc_triple, value_found))
 482
 483         if len(failset) > 0:
 484             self.fail(self._format_message(failset,
 485                                            "Parameters that were unexpectedly not set:"))
 486
 487     def _test_empty(self, program):
 488
 489         if program[0] == 'bin/samba-tool' and os.getenv("PYTHON", None):
 490             program = [os.environ["PYTHON"]] + program
 491
 492         program_arg1 = ["--configfile=%s" % (self.blankconf), "--suppress-prompt"]
 493         if (program[0] == 'bin/testparm'):
 494             program_arg1 = ["--suppress-prompt", self.blankconf]
 495
 496         print(program + program_arg1)
 497         p = subprocess.Popen(program + program_arg1,
 498                              stdout=subprocess.PIPE,
 499                              stderr=subprocess.PIPE,
 500                              cwd=self.topdir).communicate()
 501         output = ""
 502
 503         for line in p[0].decode().splitlines():
 504             if line.strip().startswith('#'):
 505                 continue
 506             if line.strip().startswith("idmap config *"):
 507                 continue
 508             output += line.strip().lower() + '\n'
 509
 510         if output.strip() != '[global]' and output.strip() != '[globals]':
 511             self.fail("Testparm returned unexpected output on an empty smb.conf.")