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         'dns hostname',
 221     ])
 222
 223     def setUp(self):
 224         super().setUp()
 225         # create a minimal smb.conf file for testparm
 226         self.smbconf = os.path.join(self.tempdir, "paramtestsmb.conf")
 227         f = open(self.smbconf, 'w')
 228         try:
 229             f.write("""
 230 [test]
 231    path = /
 232 """)
 233         finally:
 234             f.close()
 235
 236         self.blankconf = os.path.join(self.tempdir, "emptytestsmb.conf")
 237         f = open(self.blankconf, 'w')
 238         try:
 239             f.write("")
 240         finally:
 241             f.close()
 242
 243         self.topdir = os.path.abspath(samba.source_tree_topdir())
 244
 245         try:
 246             self.documented = set(get_documented_parameters(self.topdir))
 247         except:
 248             self.fail("Unable to load documented parameters")
 249
 250         try:
 251             self.defaults = set(get_documented_tuples(self.topdir))
 252         except:
 253             self.fail("Unable to load parameters")
 254
 255         try:
 256             self.defaults_all = set(get_documented_tuples(self.topdir, False))
 257         except:
 258             self.fail("Unable to load parameters")
 259
 260     def tearDown(self):
 261         super().tearDown()
 262         os.unlink(self.smbconf)
 263         os.unlink(self.blankconf)
 264
 265     def test_default_s3(self):
 266         self._test_default(['bin/testparm'])
 267         self._set_defaults(['bin/testparm'])
 268
 269         # registry shares appears to need sudo
 270         self._set_arbitrary(['bin/testparm'],
 271             exceptions = ['client lanman auth',
 272                           'client plaintext auth',
 273                           'registry shares',
 274                           'smb ports',
 275                           'rpc server dynamic port range',
 276                           'name resolve order',
 277                           'clustering'])
 278         self._test_empty(['bin/testparm'])
 279
 280     def test_default_s4(self):
 281         self._test_default(['bin/samba-tool', 'testparm'])
 282         self._set_defaults(['bin/samba-tool', 'testparm'])
 283         self._set_arbitrary(['bin/samba-tool', 'testparm'],
 284                             exceptions=['smb ports',
 285                                         'rpc server dynamic port range',
 286                                         'name resolve order'])
 287         self._test_empty(['bin/samba-tool', 'testparm'])
 288
 289     def _test_default(self, program):
 290
 291         if program[0] == 'bin/samba-tool' and os.getenv("PYTHON", None):
 292             program = [os.environ["PYTHON"]] + program
 293
 294         failset = set()
 295
 296         with concurrent.futures.ProcessPoolExecutor(max_workers=get_max_worker_count()) as executor:
 297             result_futures = []
 298
 299             for tuples in self.defaults:
 300                 param, default, context, param_type = tuples
 301
 302                 if param in self.special_cases:
 303                     continue
 304                 # bad, bad parametric options - we don't have their default values
 305                 if ':' in param:
 306                     continue
 307                 section = None
 308                 if context == "G":
 309                     section = "global"
 310                 elif context == "S":
 311                     section = "test"
 312                 else:
 313                     self.fail("%s has no valid context" % param)
 314
 315                 program_arg1 = ["--configfile=%s" % (self.smbconf)]
 316                 if (program[0] == 'bin/testparm'):
 317                     program_arg1 = ["--suppress-prompt", self.smbconf]
 318
 319                 cmdline = program + program_arg1 + [
 320                                      "--section-name",
 321                                      section,
 322                                      "--parameter-name",
 323                                      param]
 324
 325                 future = executor.submit(check_or_set_smbconf_default, cmdline, self.topdir, param, default)
 326                 result_futures.append(future)
 327
 328             for f in concurrent.futures.as_completed(result_futures):
 329                 if f.result():
 330                     result, param, default_param = f.result()
 331
 332                     doc_triple = "%s\n      Expected: %s" % (param, default_param)
 333                     failset.add("%s\n      Got: %s" % (doc_triple, result))
 334
 335         if len(failset) > 0:
 336             self.fail(self._format_message(failset,
 337                                            "Parameters that do not have matching defaults:"))
 338
 339     def _set_defaults(self, program):
 340
 341         if program[0] == 'bin/samba-tool' and os.getenv("PYTHON", None):
 342             program = [os.environ["PYTHON"]] + program
 343
 344         failset = set()
 345
 346         with concurrent.futures.ProcessPoolExecutor(max_workers=get_max_worker_count()) as executor:
 347             result_futures = []
 348
 349             for tuples in self.defaults:
 350                 param, default, context, param_type = tuples
 351
 352                 exceptions = set([
 353                     'printing',
 354                     'smbd max async dosmode',
 355                 ])
 356
 357                 if param in exceptions:
 358                     continue
 359
 360                 section = None
 361                 if context == "G":
 362                     section = "global"
 363                 elif context == "S":
 364                     section = "test"
 365                 else:
 366                     self.fail("%s has no valid context" % param)
 367
 368                 program_arg1 = ["--configfile=%s" % (self.smbconf)]
 369                 if (program[0] == 'bin/testparm'):
 370                     program_arg1 = ["--suppress-prompt", self.smbconf]
 371
 372                 cmdline = program + program_arg1 + [
 373                                      "--section-name",
 374                                      section,
 375                                      "--parameter-name",
 376                                      param,
 377                                      "--option",
 378                                      "%s = %s" % (param, default)]
 379                 future = executor.submit(check_or_set_smbconf_default, cmdline, self.topdir, param, default)
 380                 result_futures.append(future)
 381
 382             for f in concurrent.futures.as_completed(result_futures):
 383                 if f.result():
 384                     result, param, default_param = f.result()
 385
 386                     doc_triple = "%s\n      Expected: %s" % (param, default)
 387                     failset.add("%s\n      Got: %s" % (doc_triple, result))
 388
 389         if len(failset) > 0:
 390             self.fail(self._format_message(failset,
 391                                            "Parameters that do not have matching defaults:"))
 392
 393     def _set_arbitrary(self, program, exceptions=None):
 394
 395         if program[0] == 'bin/samba-tool' and os.getenv("PYTHON", None):
 396             program = [os.environ["PYTHON"]] + program
 397
 398         arbitrary = {'string': 'string', 'boolean': 'yes', 'integer': '5',
 399                      'boolean-rev': 'yes',
 400                      'cmdlist': 'a b c',
 401                      'bytes': '10',
 402                      'octal': '0123',
 403                      'ustring': 'ustring',
 404                      'enum': '', 'boolean-auto': '', 'char': 'a', 'list': 'a, b, c'}
 405         opposite_arbitrary = {'string': 'string2', 'boolean': 'no', 'integer': '6',
 406                               'boolean-rev': 'no',
 407                               'cmdlist': 'd e f',
 408                               'bytes': '11',
 409                               'octal': '0567',
 410                               'ustring': 'ustring2',
 411                               'enum': '', 'boolean-auto': '', 'char': 'b', 'list': 'd, e, f'}
 412
 413         failset = set()
 414
 415         with concurrent.futures.ProcessPoolExecutor(max_workers=get_max_worker_count()) as executor:
 416             result_futures1 = []
 417             result_futures2 = []
 418
 419             for tuples in self.defaults_all:
 420                 param, default, context, param_type = tuples
 421
 422                 if param in ['printing', 'copy', 'include', 'log level']:
 423                     continue
 424
 425                 # currently no easy way to set an arbitrary value for these
 426                 if param_type in ['enum', 'boolean-auto']:
 427                     continue
 428
 429                 if exceptions is not None:
 430                     if param in exceptions:
 431                         continue
 432
 433                 section = None
 434                 if context == "G":
 435                     section = "global"
 436                 elif context == "S":
 437                     section = "test"
 438                 else:
 439                     self.fail("%s has no valid context" % param)
 440
 441                 value_to_use = arbitrary.get(param_type)
 442                 if value_to_use is None:
 443                     self.fail("%s has an invalid type" % param)
 444
 445                 program_arg1 = ["--configfile=%s" % (self.smbconf)]
 446                 if (program[0] == 'bin/testparm'):
 447                     program_arg1 = ["--suppress-prompt", self.smbconf]
 448
 449                 cmdline = program + program_arg1 + [
 450                                      "--section-name",
 451                                      section,
 452                                      "--parameter-name",
 453                                      param,
 454                                      "--option",
 455                                      "%s = %s" % (param, value_to_use)]
 456
 457                 future = executor.submit(set_smbconf_arbitrary, cmdline, self.topdir, param, param_type, value_to_use)
 458                 result_futures1.append(future)
 459
 460                 opposite_value = opposite_arbitrary.get(param_type)
 461
 462                 cmdline = program + ["--suppress-prompt",
 463                                      "--option",
 464                                      "%s = %s" % (param, value_to_use)]
 465
 466                 future = executor.submit(set_smbconf_arbitrary_opposite, cmdline, self.topdir, self.tempdir,
 467                                          section, param, param_type, opposite_value, value_to_use)
 468                 result_futures2.append(future)
 469
 470             for f in concurrent.futures.as_completed(result_futures1):
 471                 if f.result():
 472                     result, param, value_to_use = f.result()
 473
 474                     doc_triple = "%s\n      Expected: %s" % (param, value_to_use)
 475                     failset.add("%s\n      Got: %s" % (doc_triple, result))
 476
 477             for f in concurrent.futures.as_completed(result_futures2):
 478                 if f.result():
 479                     param, value_to_use, value_found = f.result()
 480
 481                     doc_triple = "%s\n      Expected: %s" % (param, value_to_use)
 482                     failset.add("%s\n      Got: %s" % (doc_triple, value_found))
 483
 484         if len(failset) > 0:
 485             self.fail(self._format_message(failset,
 486                                            "Parameters that were unexpectedly not set:"))
 487
 488     def _test_empty(self, program):
 489
 490         if program[0] == 'bin/samba-tool' and os.getenv("PYTHON", None):
 491             program = [os.environ["PYTHON"]] + program
 492
 493         program_arg1 = ["--configfile=%s" % (self.blankconf), "--suppress-prompt"]
 494         if (program[0] == 'bin/testparm'):
 495             program_arg1 = ["--suppress-prompt", self.blankconf]
 496
 497         print(program + program_arg1)
 498         p = subprocess.Popen(program + program_arg1,
 499                              stdout=subprocess.PIPE,
 500                              stderr=subprocess.PIPE,
 501                              cwd=self.topdir).communicate()
 502         output = ""
 503
 504         for line in p[0].decode().splitlines():
 505             if line.strip().startswith('#'):
 506                 continue
 507             if line.strip().startswith("idmap config *"):
 508                 continue
 509             output += line.strip().lower() + '\n'
 510
 511         if output.strip() != '[global]' and output.strip() != '[globals]':
 512             self.fail("Testparm returned unexpected output on an empty smb.conf.")