docs_unittest.py

#!/usr/bin/python
#

# Copyright (C) 2009 Google Inc.
#
# 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 2 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, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.


"""Script for unittesting documentation"""

import unittest
import re
import itertools
import operator

from ganeti import _autoconf
from ganeti import utils
from ganeti import cmdlib
from ganeti import build
from ganeti import compat
from ganeti import mcpu
from ganeti import opcodes
from ganeti import constants
from ganeti.rapi import baserlib
from ganeti.rapi import rlib2
from ganeti.rapi import connector

import testutils


VALID_URI_RE = re.compile(r"^[-/a-z0-9]*$")

RAPI_OPCODE_EXCLUDE = frozenset([
  # Not yet implemented
  opcodes.OpBackupQuery,
  opcodes.OpBackupRemove,
  opcodes.OpClusterConfigQuery,
  opcodes.OpClusterRepairDiskSizes,
  opcodes.OpClusterVerify,
  opcodes.OpClusterVerifyDisks,
  opcodes.OpInstanceChangeGroup,
  opcodes.OpInstanceMove,
  opcodes.OpNodeQueryvols,
  opcodes.OpOobCommand,
  opcodes.OpTagsSearch,
  opcodes.OpClusterActivateMasterIp,
  opcodes.OpClusterDeactivateMasterIp,

  # Difficult if not impossible
  opcodes.OpClusterDestroy,
  opcodes.OpClusterPostInit,
  opcodes.OpClusterRename,
  opcodes.OpNodeAdd,
  opcodes.OpNodeRemove,

  # Helper opcodes (e.g. submitted by LUs)
  opcodes.OpClusterVerifyConfig,
  opcodes.OpClusterVerifyGroup,
  opcodes.OpGroupEvacuate,
  opcodes.OpGroupVerifyDisks,

  # Test opcodes
  opcodes.OpTestAllocator,
  opcodes.OpTestDelay,
  opcodes.OpTestDummy,
  opcodes.OpTestJqueue,
  ])


def _ReadDocFile(filename):
  return utils.ReadFile("%s/doc/%s" %
                        (testutils.GetSourceDir(), filename))


class TestHooksDocs(unittest.TestCase):
  HOOK_PATH_OK = frozenset([
    "master-ip-turnup",
    "master-ip-turndown",
    ])

  def test(self):
    """Check whether all hooks are documented.

    """
    hooksdoc = _ReadDocFile("hooks.rst")

    # Reverse mapping from LU to opcode
    lu2opcode = dict((lu, op)
                     for (op, lu) in mcpu.Processor.DISPATCH_TABLE.items())
    assert len(lu2opcode) == len(mcpu.Processor.DISPATCH_TABLE), \
      "Found duplicate entries"

    hooks_paths = frozenset(re.findall("^:directory:\s*(.+)\s*$", hooksdoc,
                                       re.M))
    self.assertTrue(self.HOOK_PATH_OK.issubset(hooks_paths),
                    msg="Whitelisted path not found in documentation")

    raw_hooks_ops = re.findall("^OP_(?!CODE$).+$", hooksdoc, re.M)
    hooks_ops = set()
    duplicate_ops = set()
    for op in raw_hooks_ops:
      if op in hooks_ops:
        duplicate_ops.add(op)
      else:
        hooks_ops.add(op)

    self.assertFalse(duplicate_ops,
                     msg="Found duplicate opcode documentation: %s" %
                         utils.CommaJoin(duplicate_ops))

    seen_paths = set()
    seen_ops = set()

    self.assertFalse(duplicate_ops,
                     msg="Found duplicated hook documentation: %s" %
                         utils.CommaJoin(duplicate_ops))

    for name in dir(cmdlib):
      lucls = getattr(cmdlib, name)

      if (isinstance(lucls, type) and
          issubclass(lucls, cmdlib.LogicalUnit) and
          hasattr(lucls, "HPATH")):
        if lucls.HTYPE is None:
          continue

        opcls = lu2opcode.get(lucls, None)

        if opcls:
          seen_ops.add(opcls.OP_ID)
          self.assertTrue(opcls.OP_ID in hooks_ops,
                          msg="Missing hook documentation for %s" %
                              opcls.OP_ID)
        self.assertTrue(lucls.HPATH in hooks_paths,
                        msg="Missing documentation for hook %s/%s" %
                            (lucls.HTYPE, lucls.HPATH))
        seen_paths.add(lucls.HPATH)

    missed_ops = hooks_ops - seen_ops
    missed_paths = hooks_paths - seen_paths - self.HOOK_PATH_OK

    self.assertFalse(missed_ops,
                     msg="Op documents hook not existing anymore: %s" %
                         utils.CommaJoin(missed_ops))

    self.assertFalse(missed_paths,
                     msg="Hook path does not exist in opcode: %s" %
                         utils.CommaJoin(missed_paths))


class TestRapiDocs(unittest.TestCase):
  def _CheckRapiResource(self, uri, fixup, handler):
    docline = "%s resource." % uri
    self.assertEqual(handler.__doc__.splitlines()[0].strip(), docline,
                     msg=("First line of %r's docstring is not %r" %
                          (handler, docline)))

    # Apply fixes before testing
    for (rx, value) in fixup.items():
      uri = rx.sub(value, uri)

    self.assertTrue(VALID_URI_RE.match(uri), msg="Invalid URI %r" % uri)

  def test(self):
    """Check whether all RAPI resources are documented.

    """
    rapidoc = _ReadDocFile("rapi.rst")

    node_name = re.escape("[node_name]")
    instance_name = re.escape("[instance_name]")
    group_name = re.escape("[group_name]")
    job_id = re.escape("[job_id]")
    disk_index = re.escape("[disk_index]")
    query_res = re.escape("[resource]")

    resources = connector.GetHandlers(node_name, instance_name, group_name,
                                      job_id, disk_index, query_res)

    handler_dups = utils.FindDuplicates(resources.values())
    self.assertFalse(handler_dups,
                     msg=("Resource handlers used more than once: %r" %
                          handler_dups))

    uri_check_fixup = {
      re.compile(node_name): "node1examplecom",
      re.compile(instance_name): "inst1examplecom",
      re.compile(group_name): "group4440",
      re.compile(job_id): "9409",
      re.compile(disk_index): "123",
      re.compile(query_res): "lock",
      }

    assert compat.all(VALID_URI_RE.match(value)
                      for value in uri_check_fixup.values()), \
           "Fixup values must be valid URIs, too"

    titles = []

    prevline = None
    for line in rapidoc.splitlines():
      if re.match(r"^\++$", line):
        titles.append(prevline)

      prevline = line

    prefix_exception = frozenset(["/", "/version", "/2"])

    undocumented = []
    used_uris = []

    for key, handler in resources.iteritems():
      # Regex objects
      if hasattr(key, "match"):
        self.assert_(key.pattern.startswith("^/2/"),
                     msg="Pattern %r does not start with '^/2/'" % key.pattern)
        self.assertEqual(key.pattern[-1], "$")

        found = False
        for title in titles:
          if title.startswith("``") and title.endswith("``"):
            uri = title[2:-2]
            if key.match(uri):
              self._CheckRapiResource(uri, uri_check_fixup, handler)
              used_uris.append(uri)
              found = True
              break

        if not found:
          # TODO: Find better way of identifying resource
          undocumented.append(key.pattern)

      else:
        self.assert_(key.startswith("/2/") or key in prefix_exception,
                     msg="Path %r does not start with '/2/'" % key)

        if ("``%s``" % key) in titles:
          self._CheckRapiResource(key, {}, handler)
          used_uris.append(key)
        else:
          undocumented.append(key)

    self.failIf(undocumented,
                msg=("Missing RAPI resource documentation for %s" %
                     utils.CommaJoin(undocumented)))

    uri_dups = utils.FindDuplicates(used_uris)
    self.failIf(uri_dups,
                msg=("URIs matched by more than one resource: %s" %
                     utils.CommaJoin(uri_dups)))

    self._FindRapiMissing(resources.values())
    self._CheckTagHandlers(resources.values())

  def _FindRapiMissing(self, handlers):
    used = frozenset(itertools.chain(*map(baserlib.GetResourceOpcodes,
                                          handlers)))

    unexpected = used & RAPI_OPCODE_EXCLUDE
    self.assertFalse(unexpected,
      msg=("Found RAPI resources for excluded opcodes: %s" %
           utils.CommaJoin(_GetOpIds(unexpected))))

    missing = (frozenset(opcodes.OP_MAPPING.values()) - used -
               RAPI_OPCODE_EXCLUDE)
    self.assertFalse(missing,
      msg=("Missing RAPI resources for opcodes: %s" %
           utils.CommaJoin(_GetOpIds(missing))))

  def _CheckTagHandlers(self, handlers):
    tag_handlers = filter(lambda x: issubclass(x, rlib2._R_Tags), handlers)
    self.assertEqual(frozenset(map(operator.attrgetter("TAG_LEVEL"),
                                   tag_handlers)),
                     constants.VALID_TAG_TYPES)


def _GetOpIds(ops):
  """Returns C{OP_ID} for all opcodes in passed sequence.

  """
  return sorted(opcls.OP_ID for opcls in ops)


class TestManpages(unittest.TestCase):
  """Manpage tests"""

  @staticmethod
  def _ReadManFile(name):
    return utils.ReadFile("%s/man/%s.rst" %
                          (testutils.GetSourceDir(), name))

  @staticmethod
  def _LoadScript(name):
    return build.LoadModule("scripts/%s" % name)

  def test(self):
    for script in _autoconf.GNT_SCRIPTS:
      self._CheckManpage(script,
                         self._ReadManFile(script),
                         self._LoadScript(script).commands.keys())

  def _CheckManpage(self, script, mantext, commands):
    missing = []

    for cmd in commands:
      pattern = r"^(\| )?\*\*%s\*\*" % re.escape(cmd)
      if not re.findall(pattern, mantext, re.DOTALL | re.MULTILINE):
        missing.append(cmd)

    self.failIf(missing,
                msg=("Manpage for '%s' missing documentation for %s" %
                     (script, utils.CommaJoin(missing))))


if __name__ == "__main__":
  testutils.GanetiTestProgram()