lib/build/sphinx_ext.py

#
#

# Copyright (C) 2011, 2012, 2013 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.


"""Sphinx extension for building opcode documentation.

"""

import re
from cStringIO import StringIO

import docutils.statemachine
import docutils.nodes
import docutils.utils
import docutils.parsers.rst

import sphinx.errors
import sphinx.util.compat
import sphinx.roles
import sphinx.addnodes

s_compat = sphinx.util.compat

try:
  # Access to a protected member of a client class
  # pylint: disable=W0212
  orig_manpage_role = docutils.parsers.rst.roles._roles["manpage"]
except (AttributeError, ValueError, KeyError), err:
  # Normally the "manpage" role is registered by sphinx/roles.py
  raise Exception("Can't find reST role named 'manpage': %s" % err)

from ganeti import constants
from ganeti import compat
from ganeti import errors
from ganeti import utils
from ganeti import opcodes
from ganeti import ht
from ganeti import rapi
from ganeti import luxi
from ganeti import objects
from ganeti import http
from ganeti import _autoconf

import ganeti.rapi.rlib2 # pylint: disable=W0611
import ganeti.rapi.connector # pylint: disable=W0611


#: Regular expression for man page names
_MAN_RE = re.compile(r"^(?P<name>[-\w_]+)\((?P<section>\d+)\)$")

_TAB_WIDTH = 2

RAPI_URI_ENCODE_RE = re.compile("[^_a-z0-9]+", re.I)


class ReSTError(Exception):
  """Custom class for generating errors in Sphinx.

  """


def _GetCommonParamNames():
  """Builds a list of parameters common to all opcodes.

  """
  names = set(map(compat.fst, opcodes.OpCode.OP_PARAMS))

  # The "depends" attribute should be listed
  names.remove(opcodes.DEPEND_ATTR)

  return names


COMMON_PARAM_NAMES = _GetCommonParamNames()

#: Namespace for evaluating expressions
EVAL_NS = dict(compat=compat, constants=constants, utils=utils, errors=errors,
               rlib2=rapi.rlib2, luxi=luxi, rapi=rapi, objects=objects,
               http=http)

# Constants documentation for man pages
CV_ECODES_DOC = "ecodes"
# We don't care about the leak of variables _, name and doc here.
# pylint: disable=W0621
CV_ECODES_DOC_LIST = [(name, doc) for (_, name, doc) in constants.CV_ALL_ECODES]
DOCUMENTED_CONSTANTS = {
  CV_ECODES_DOC: CV_ECODES_DOC_LIST,
  }


class OpcodeError(sphinx.errors.SphinxError):
  category = "Opcode error"


def _SplitOption(text):
  """Split simple option list.

  @type text: string
  @param text: Options, e.g. "foo, bar, baz"

  """
  return [i.strip(",").strip() for i in text.split()]


def _ParseAlias(text):
  """Parse simple assignment option.

  @type text: string
  @param text: Assignments, e.g. "foo=bar, hello=world"
  @rtype: dict

  """
  result = {}

  for part in _SplitOption(text):
    if "=" not in part:
      raise OpcodeError("Invalid option format, missing equal sign")

    (name, value) = part.split("=", 1)

    result[name.strip()] = value.strip()

  return result


def _BuildOpcodeParams(op_id, include, exclude, alias):
  """Build opcode parameter documentation.

  @type op_id: string
  @param op_id: Opcode ID

  """
  op_cls = opcodes.OP_MAPPING[op_id]

  params_with_alias = \
    utils.NiceSort([(alias.get(name, name), name, default, test, doc)
                    for (name, default, test, doc) in op_cls.GetAllParams()],
                   key=compat.fst)

  for (rapi_name, name, default, test, doc) in params_with_alias:
    # Hide common parameters if not explicitly included
    if (name in COMMON_PARAM_NAMES and
        (not include or name not in include)):
      continue
    if exclude is not None and name in exclude:
      continue
    if include is not None and name not in include:
      continue

    has_default = default is not ht.NoDefault
    has_test = not (test is None or test is ht.NoType)

    buf = StringIO()
    buf.write("``%s``" % (rapi_name,))
    if has_default or has_test:
      buf.write(" (")
      if has_default:
        buf.write("defaults to ``%s``" % (default,))
        if has_test:
          buf.write(", ")
      if has_test:
        buf.write("must be ``%s``" % (test,))
      buf.write(")")
    yield buf.getvalue()

    # Add text
    for line in doc.splitlines():
      yield "  %s" % line


def _BuildOpcodeResult(op_id):
  """Build opcode result documentation.

  @type op_id: string
  @param op_id: Opcode ID

  """
  op_cls = opcodes.OP_MAPPING[op_id]

  result_fn = getattr(op_cls, "OP_RESULT", None)

  if not result_fn:
    raise OpcodeError("Opcode '%s' has no result description" % op_id)

  return "``%s``" % result_fn


class OpcodeParams(s_compat.Directive):
  """Custom directive for opcode parameters.

  See also <http://docutils.sourceforge.net/docs/howto/rst-directives.html>.

  """
  has_content = False
  required_arguments = 1
  optional_arguments = 0
  final_argument_whitespace = False
  option_spec = dict(include=_SplitOption, exclude=_SplitOption,
                     alias=_ParseAlias)

  def run(self):
    op_id = self.arguments[0]
    include = self.options.get("include", None)
    exclude = self.options.get("exclude", None)
    alias = self.options.get("alias", {})

    path = op_id
    include_text = "\n".join(_BuildOpcodeParams(op_id, include, exclude, alias))

    # Inject into state machine
    include_lines = docutils.statemachine.string2lines(include_text, _TAB_WIDTH,
                                                       convert_whitespace=1)
    self.state_machine.insert_input(include_lines, path)

    return []


class OpcodeResult(s_compat.Directive):
  """Custom directive for opcode result.

  See also <http://docutils.sourceforge.net/docs/howto/rst-directives.html>.

  """
  has_content = False
  required_arguments = 1
  optional_arguments = 0
  final_argument_whitespace = False

  def run(self):
    op_id = self.arguments[0]

    path = op_id
    include_text = _BuildOpcodeResult(op_id)

    # Inject into state machine
    include_lines = docutils.statemachine.string2lines(include_text, _TAB_WIDTH,
                                                       convert_whitespace=1)
    self.state_machine.insert_input(include_lines, path)

    return []


def PythonEvalRole(role, rawtext, text, lineno, inliner,
                   options={}, content=[]):
  """Custom role to evaluate Python expressions.

  The expression's result is included as a literal.

  """
  # pylint: disable=W0102,W0613,W0142
  # W0102: Dangerous default value as argument
  # W0142: Used * or ** magic
  # W0613: Unused argument

  code = docutils.utils.unescape(text, restore_backslashes=True)

  try:
    result = eval(code, EVAL_NS)
  except Exception, err: # pylint: disable=W0703
    msg = inliner.reporter.error("Failed to evaluate %r: %s" % (code, err),
                                 line=lineno)
    return ([inliner.problematic(rawtext, rawtext, msg)], [msg])

  node = docutils.nodes.literal("", unicode(result), **options)

  return ([node], [])


class PythonAssert(s_compat.Directive):
  """Custom directive for writing assertions.

  The content must be a valid Python expression. If its result does not
  evaluate to C{True}, the assertion fails.

  """
  has_content = True
  required_arguments = 0
  optional_arguments = 0
  final_argument_whitespace = False

  def run(self):
    # Handle combinations of Sphinx and docutils not providing the wanted method
    if hasattr(self, "assert_has_content"):
      self.assert_has_content()
    else:
      assert self.content

    code = "\n".join(self.content)

    try:
      result = eval(code, EVAL_NS)
    except Exception, err:
      raise self.error("Failed to evaluate %r: %s" % (code, err))

    if not result:
      raise self.error("Assertion failed: %s" % (code, ))

    return []


def BuildQueryFields(fields):
  """Build query fields documentation.

  @type fields: dict (field name as key, field details as value)

  """
  defs = [(fdef.name, fdef.doc)
           for (_, (fdef, _, _, _)) in utils.NiceSort(fields.items(),
                                                      key=compat.fst)]
  return BuildValuesDoc(defs)


def BuildValuesDoc(values):
  """Builds documentation for a list of values

  @type values: list of tuples in the form (value, documentation)

  """
  for name, doc in values:
    assert len(doc.splitlines()) == 1
    yield "``%s``" % (name,)
    yield "  %s" % (doc,)


def _ManPageNodeClass(*args, **kwargs):
  """Generates a pending XRef like a ":doc:`...`" reference.

  """
  # Type for sphinx/environment.py:BuildEnvironment.resolve_references
  kwargs["reftype"] = "doc"

  # Force custom title
  kwargs["refexplicit"] = True

  return sphinx.addnodes.pending_xref(*args, **kwargs)


class _ManPageXRefRole(sphinx.roles.XRefRole):
  def __init__(self):
    """Initializes this class.

    """
    sphinx.roles.XRefRole.__init__(self, nodeclass=_ManPageNodeClass,
                                   warn_dangling=True)

    assert not hasattr(self, "converted"), \
      "Sphinx base class gained an attribute named 'converted'"

    self.converted = None

  def process_link(self, env, refnode, has_explicit_title, title, target):
    """Specialization for man page links.

    """
    if has_explicit_title:
      raise ReSTError("Setting explicit title is not allowed for man pages")

    # Check format and extract name and section
    m = _MAN_RE.match(title)
    if not m:
      raise ReSTError("Man page reference '%s' does not match regular"
                      " expression '%s'" % (title, _MAN_RE.pattern))

    name = m.group("name")
    section = int(m.group("section"))

    wanted_section = _autoconf.MAN_PAGES.get(name, None)

    if not (wanted_section is None or wanted_section == section):
      raise ReSTError("Referenced man page '%s' has section number %s, but the"
                      " reference uses section %s" %
                      (name, wanted_section, section))

    self.converted = bool(wanted_section is not None and
                          env.app.config.enable_manpages)

    if self.converted:
      # Create link to known man page
      return (title, "man-%s" % name)
    else:
      # No changes
      return (title, target)


def _ManPageRole(typ, rawtext, text, lineno, inliner, # pylint: disable=W0102
                 options={}, content=[]):
  """Custom role for man page references.

  Converts man pages to links if enabled during the build.

  """
  xref = _ManPageXRefRole()

  assert ht.TNone(xref.converted)

  # Check if it's a known man page
  try:
    result = xref(typ, rawtext, text, lineno, inliner,
                  options=options, content=content)
  except ReSTError, err:
    msg = inliner.reporter.error(str(err), line=lineno)
    return ([inliner.problematic(rawtext, rawtext, msg)], [msg])

  assert ht.TBool(xref.converted)

  # Return if the conversion was successful (i.e. the man page was known and
  # conversion was enabled)
  if xref.converted:
    return result

  # Fallback if man page links are disabled or an unknown page is referenced
  return orig_manpage_role(typ, rawtext, text, lineno, inliner,
                           options=options, content=content)


def _EncodeRapiResourceLink(method, uri):
  """Encodes a RAPI resource URI for use as a link target.

  """
  parts = [RAPI_URI_ENCODE_RE.sub("-", uri.lower()).strip("-")]

  if method is not None:
    parts.append(method.lower())

  return "rapi-res-%s" % "+".join(filter(None, parts))


def _MakeRapiResourceLink(method, uri):
  """Generates link target name for RAPI resource.

  """
  if uri in ["/", "/2"]:
    # Don't link these
    return None

  elif uri == "/version":
    return _EncodeRapiResourceLink(method, uri)

  elif uri.startswith("/2/"):
    return _EncodeRapiResourceLink(method, uri[len("/2/"):])

  else:
    raise ReSTError("Unhandled URI '%s'" % uri)


def _GetHandlerMethods(handler):
  """Returns list of HTTP methods supported by handler class.

  @type handler: L{rapi.baserlib.ResourceBase}
  @param handler: Handler class
  @rtype: list of strings

  """
  return sorted(method
                for (method, op_attr, _, _) in rapi.baserlib.OPCODE_ATTRS
                # Only if handler supports method
                if hasattr(handler, method) or hasattr(handler, op_attr))


def _DescribeHandlerAccess(handler, method):
  """Returns textual description of required RAPI permissions.

  @type handler: L{rapi.baserlib.ResourceBase}
  @param handler: Handler class
  @type method: string
  @param method: HTTP method (e.g. L{http.HTTP_GET})
  @rtype: string

  """
  access = rapi.baserlib.GetHandlerAccess(handler, method)

  if access:
    return utils.CommaJoin(sorted(access))
  else:
    return "*(none)*"


class _RapiHandlersForDocsHelper(object):
  @classmethod
  def Build(cls):
    """Returns dictionary of resource handlers.

    """
    resources = \
      rapi.connector.GetHandlers("[node_name]", "[instance_name]",
                                 "[group_name]", "[network_name]", "[job_id]",
                                 "[disk_index]", "[resource]",
                                 translate=cls._TranslateResourceUri)

    return resources

  @classmethod
  def _TranslateResourceUri(cls, *args):
    """Translates a resource URI for use in documentation.

    @see: L{rapi.connector.GetHandlers}

    """
    return "".join(map(cls._UriPatternToString, args))

  @staticmethod
  def _UriPatternToString(value):
    """Converts L{rapi.connector.UriPattern} to strings.

    """
    if isinstance(value, rapi.connector.UriPattern):
      return value.content
    else:
      return value


_RAPI_RESOURCES_FOR_DOCS = _RapiHandlersForDocsHelper.Build()


def _BuildRapiAccessTable(res):
  """Build a table with access permissions needed for all RAPI resources.

  """
  for (uri, handler) in utils.NiceSort(res.items(), key=compat.fst):
    reslink = _MakeRapiResourceLink(None, uri)
    if not reslink:
      # No link was generated
      continue

    yield ":ref:`%s <%s>`" % (uri, reslink)

    for method in _GetHandlerMethods(handler):
      yield ("  | :ref:`%s <%s>`: %s" %
             (method, _MakeRapiResourceLink(method, uri),
              _DescribeHandlerAccess(handler, method)))


class RapiAccessTable(s_compat.Directive):
  """Custom directive to generate table of all RAPI resources.

  See also <http://docutils.sourceforge.net/docs/howto/rst-directives.html>.

  """
  has_content = False
  required_arguments = 0
  optional_arguments = 0
  final_argument_whitespace = False
  option_spec = {}

  def run(self):
    include_text = "\n".join(_BuildRapiAccessTable(_RAPI_RESOURCES_FOR_DOCS))

    # Inject into state machine
    include_lines = docutils.statemachine.string2lines(include_text, _TAB_WIDTH,
                                                       convert_whitespace=1)
    self.state_machine.insert_input(include_lines, self.__class__.__name__)

    return []


class RapiResourceDetails(s_compat.Directive):
  """Custom directive for RAPI resource details.

  See also <http://docutils.sourceforge.net/docs/howto/rst-directives.html>.

  """
  has_content = False
  required_arguments = 1
  optional_arguments = 0
  final_argument_whitespace = False

  def run(self):
    uri = self.arguments[0]

    try:
      handler = _RAPI_RESOURCES_FOR_DOCS[uri]
    except KeyError:
      raise self.error("Unknown resource URI '%s'" % uri)

    lines = [
      ".. list-table::",
      "   :widths: 1 4",
      "   :header-rows: 1",
      "",
      "   * - Method",
      "     - :ref:`Required permissions <rapi-users>`",
      ]

    for method in _GetHandlerMethods(handler):
      lines.extend([
        "   * - :ref:`%s <%s>`" % (method, _MakeRapiResourceLink(method, uri)),
        "     - %s" % _DescribeHandlerAccess(handler, method),
        ])

    # Inject into state machine
    include_lines = \
      docutils.statemachine.string2lines("\n".join(lines), _TAB_WIDTH,
                                         convert_whitespace=1)
    self.state_machine.insert_input(include_lines, self.__class__.__name__)

    return []


def setup(app):
  """Sphinx extension callback.

  """
  # TODO: Implement Sphinx directive for query fields
  app.add_directive("opcode_params", OpcodeParams)
  app.add_directive("opcode_result", OpcodeResult)
  app.add_directive("pyassert", PythonAssert)
  app.add_role("pyeval", PythonEvalRole)
  app.add_directive("rapi_access_table", RapiAccessTable)
  app.add_directive("rapi_resource_details", RapiResourceDetails)

  app.add_config_value("enable_manpages", False, True)
  app.add_role("manpage", _ManPageRole)