From 7ea4bbdfb4a90f37f037efbda6f04a039a2e34d3 Mon Sep 17 00:00:00 2001 From: Daniel Hilst <392820+dhilst@users.noreply.github.com> Date: Fri, 11 Sep 2026 07:55:59 -0300 Subject: [PATCH] test(provtest): add provtest, a wire-level provision chain client provtest asks the management node the questions a booting node asks and asserts on the answers. DNS, TFTP and HTTP are driven by dig, curl and tftp, so the result is what a real client sees rather than what this tool's own protocol parser believed. The xcatd stages use sockets: they must choose their source address, because xcatd names a client by the reverse lookup of the address a connection arrived from. Scenarios are declarative .conf files sharing dhcptest's assertion grammar, exit codes and TAP output. Nothing here reads the xCAT database or runs an xCAT command; every expected value arrives via --set. validate and list run offline, in a checkout, with no root. --- xCAT-test/provtest/src/provtest | 18 + .../provtest/src/provtest_lib/__init__.py | 14 + .../provtest/src/provtest_lib/assertions.py | 229 +++++++++ xCAT-test/provtest/src/provtest_lib/cli.py | 160 ++++++ xCAT-test/provtest/src/provtest_lib/config.py | 242 +++++++++ xCAT-test/provtest/src/provtest_lib/dnsc.py | 120 +++++ xCAT-test/provtest/src/provtest_lib/errors.py | 46 ++ xCAT-test/provtest/src/provtest_lib/httpc.py | 165 ++++++ .../provtest/src/provtest_lib/machine.py | 229 +++++++++ xCAT-test/provtest/src/provtest_lib/model.py | 184 +++++++ .../provtest/src/provtest_lib/netutil.py | 91 ++++ xCAT-test/provtest/src/provtest_lib/proc.py | 92 ++++ xCAT-test/provtest/src/provtest_lib/report.py | 216 ++++++++ xCAT-test/provtest/src/provtest_lib/runner.py | 354 +++++++++++++ xCAT-test/provtest/src/provtest_lib/subst.py | 89 ++++ xCAT-test/provtest/src/provtest_lib/tftpc.py | 105 ++++ xCAT-test/provtest/src/provtest_lib/xcatc.py | 470 ++++++++++++++++++ 17 files changed, 2824 insertions(+) create mode 100755 xCAT-test/provtest/src/provtest create mode 100644 xCAT-test/provtest/src/provtest_lib/__init__.py create mode 100644 xCAT-test/provtest/src/provtest_lib/assertions.py create mode 100644 xCAT-test/provtest/src/provtest_lib/cli.py create mode 100644 xCAT-test/provtest/src/provtest_lib/config.py create mode 100644 xCAT-test/provtest/src/provtest_lib/dnsc.py create mode 100644 xCAT-test/provtest/src/provtest_lib/errors.py create mode 100644 xCAT-test/provtest/src/provtest_lib/httpc.py create mode 100644 xCAT-test/provtest/src/provtest_lib/machine.py create mode 100644 xCAT-test/provtest/src/provtest_lib/model.py create mode 100644 xCAT-test/provtest/src/provtest_lib/netutil.py create mode 100644 xCAT-test/provtest/src/provtest_lib/proc.py create mode 100644 xCAT-test/provtest/src/provtest_lib/report.py create mode 100644 xCAT-test/provtest/src/provtest_lib/runner.py create mode 100644 xCAT-test/provtest/src/provtest_lib/subst.py create mode 100644 xCAT-test/provtest/src/provtest_lib/tftpc.py create mode 100644 xCAT-test/provtest/src/provtest_lib/xcatc.py diff --git a/xCAT-test/provtest/src/provtest b/xCAT-test/provtest/src/provtest new file mode 100755 index 000000000..2846b30fd --- /dev/null +++ b/xCAT-test/provtest/src/provtest @@ -0,0 +1,18 @@ +#!/usr/bin/env python3 +"""provtest -- drive the xCAT provision chain the way a booting node does. + +Runs straight out of a checkout: this script puts its own directory on +sys.path, so no installation, no virtualenv and no pip are involved. Nothing +beyond the standard library is imported; the DNS, HTTP and TFTP stages are +driven by `dig`, `curl` and `tftp`, and the xCAT stages by a socket. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.realpath(__file__))) + +from provtest_lib.cli import main # noqa: E402 (path set up above) + +if __name__ == "__main__": + sys.exit(main()) diff --git a/xCAT-test/provtest/src/provtest_lib/__init__.py b/xCAT-test/provtest/src/provtest_lib/__init__.py new file mode 100644 index 000000000..753608dd0 --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/__init__.py @@ -0,0 +1,14 @@ +"""provtest -- drive the xCAT provision chain the way a booting node does. + +A node that is being provisioned speaks five protocols in turn: it resolves +names, fetches a loader and a configuration over TFTP, fetches a kernel and a +kickstart over HTTP, asks xcatd where it is going, and reports back while it +installs. This tool speaks all five, from a client's side, and asserts on what +comes back. + +It never reads the xCAT database and never runs an xCAT command. Every value it +expects is supplied to it, so a test cannot be satisfied by asking xCAT to +confirm its own output. +""" + +__version__ = "1.0" diff --git a/xCAT-test/provtest/src/provtest_lib/assertions.py b/xCAT-test/provtest/src/provtest_lib/assertions.py new file mode 100644 index 000000000..5a9aa6169 --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/assertions.py @@ -0,0 +1,229 @@ +"""The assertion mini-language: `target op value`. + +One assertion per line, under a single multi-line `assert` key. A single key +holding many lines sidesteps configparser's one-value-per-key rule and keeps +the grammar uniform, so `status == NXDOMAIN` and `text contains destiny=install` +parse through exactly the same path. + +The operator table is dhcptest's, so an operator reads the same in both suites. +""" + +import re + +from . import netutil, subst +from .errors import ConfigError +from .model import Assertion + +#: Targets that describe the step rather than the reply. +META_TARGETS = frozenset(["attempt"]) + +#: Operators that take no value. +NULLARY_OPS = frozenset(["present", "absent"]) + +#: Operators asserting that the reply is *not* something, which a reply that +#: says nothing on the subject satisfies. +NEGATIVE_OPS = frozenset(["!=", "not-in"]) + +OPS = frozenset([ + "==", "!=", "in", "not-in", "present", "absent", + "matches", "contains", "starts-with", "ends-with", + "<", "<=", ">", ">=", +]) + + +class Result(object): + """The outcome of one assertion, carrying enough to explain a failure.""" + + __slots__ = ("assertion", "ok", "expected", "actual", "detail") + + def __init__(self, assertion, ok, expected=None, actual=None, detail=""): + self.assertion = assertion + self.ok = ok + self.expected = expected + self.actual = actual + self.detail = detail + + def __repr__(self): + return "Result(%r, ok=%r, expected=%r, actual=%r)" % ( + self.assertion.render(), self.ok, self.expected, self.actual) + + +def parse(line, source=""): + """Parse one assertion line into an `Assertion`.""" + text = line.strip() + if not text: + raise ConfigError("empty assertion", source) + + parts = text.split(None, 2) + if len(parts) < 2: + raise ConfigError("assertion needs at least a target and an operator: %r" + % (text,), source) + + target, op = parts[0], parts[1] + value = parts[2] if len(parts) > 2 else None + + if op not in OPS: + raise ConfigError( + "unknown operator %r in %r (expected one of: %s)" + % (op, text, ", ".join(sorted(OPS))), source) + + if op in NULLARY_OPS: + if value: + raise ConfigError("operator %r takes no value: %r" % (op, text), source) + value = None + elif value is None: + # `error != ` with nothing after it is a legitimate "must be non-empty". + value = "" + + if op == "matches": + try: + re.compile(value) + except re.error as exc: + raise ConfigError("bad regex %r: %s" % (value, exc), source) + + return Assertion(target=target, op=op, value=value, source=source, text=text) + + +def parse_block(block, source=""): + """Parse a multi-line `assert = ...` value into a tuple of assertions.""" + result = [] + for number, raw in enumerate(str(block or "").splitlines(), start=1): + line = raw.strip() + if not line or line.startswith("#"): + continue + where = "%s line %d of assert block" % (source, number) if source else "" + result.append(parse(line, where)) + return tuple(result) + + +# --------------------------------------------------------------------------- +# evaluation + + +def evaluate(assertion, reply, context, extras=None): + """Check one assertion against one reply.""" + extras = extras or {} + expected = None + if assertion.value is not None: + expected = subst.resolve(assertion.value, context) + + present, actual = _actual(assertion.target, reply, context, extras) + + if assertion.op == "present": + return Result(assertion, present, expected="present", + actual="present" if present else "absent") + if assertion.op == "absent": + return Result(assertion, not present, expected="absent", + actual="present" if present else "absent") + + if not present: + # A negative comparison is satisfied by a target that is not there at + # all: a config that names no kernel has certainly not named the wrong + # one. `absent` remains the way to assert the absence itself. + if assertion.op in NEGATIVE_OPS: + return Result(assertion, True, expected=expected, actual=None, + detail="%s is not present in the reply" + % (assertion.target,)) + return Result(assertion, False, expected=expected, actual=None, + detail="%s is not present in the reply" % (assertion.target,)) + + ok, detail = _compare(assertion.op, actual, expected) + return Result(assertion, ok, expected=expected, + actual=subst.format_value(actual), detail=detail) + + +def _actual(target, reply, context, extras): + """Return `(present, value)` for one assertion target.""" + if target in META_TARGETS: + if target not in extras: + return False, None + return True, extras[target] + + if target.startswith("$"): + try: + return True, subst.resolve(target, context) + except ConfigError: + return False, None + + if reply is None: + return False, None + if not reply.has(target): + return False, None + return True, reply.field(target) + + +def _compare(op, actual, expected): + """Type-aware comparison. Returns `(ok, detail)`.""" + if op in ("==", "!="): + same = _equal(actual, expected) + return (same if op == "==" else not same), "" + + if op in ("in", "not-in"): + inside = _member(actual, expected) + return (inside if op == "in" else not inside), "" + + text = subst.format_value(actual) + if op == "matches": + return bool(re.search(expected, text, re.MULTILINE)), "" + if op == "contains": + return expected in text, "" + if op == "starts-with": + return text.startswith(expected), "" + if op == "ends-with": + return text.endswith(expected), "" + + left, right = _as_numbers(actual, expected) + if left is None: + return False, "%r and %r are not both numbers" % (actual, expected) + if op == "<": + return left < right, "" + if op == "<=": + return left <= right, "" + if op == ">": + return left > right, "" + if op == ">=": + return left >= right, "" + raise ConfigError("unhandled operator %r" % (op,)) + + +def _equal(actual, expected): + """Compare as addresses, then as numbers, then as text. + + A list on the left means "any of these", so a name with two A records + satisfies `data == `. Writing the whole set is `data.0` and + `data.1`, or `count == 2`. + """ + if isinstance(actual, (list, tuple)): + return any(_equal(item, expected) for item in actual) + + left_ip = netutil.parse_ip(subst.format_value(actual)) + right_ip = netutil.parse_ip(expected) + if left_ip is not None and right_ip is not None: + return left_ip == right_ip + + left, right = _as_numbers(actual, expected) + if left is not None: + return left == right + + return subst.format_value(actual) == str(expected) + + +def _member(actual, expected): + """`in` means membership: of a CIDR, of a `first-last` range, or of a list.""" + text = subst.format_value(actual) + if "/" in expected and netutil.parse_ip(text) is not None: + return netutil.ip_in(text, expected) + if netutil.parse_range(expected) is not None: + return netutil.ip_in_range(text, expected) + candidates = [item.strip() for item in expected.split(",") if item.strip()] + return any(_equal(actual, candidate) for candidate in candidates) + + +def _as_numbers(actual, expected): + """Both sides as ints, or `(None, None)` when either is not numeric.""" + try: + left = actual if isinstance(actual, int) else int(str(actual).strip(), 0) + right = expected if isinstance(expected, int) else int(str(expected).strip(), 0) + except (TypeError, ValueError): + return None, None + return left, right diff --git a/xCAT-test/provtest/src/provtest_lib/cli.py b/xCAT-test/provtest/src/provtest_lib/cli.py new file mode 100644 index 000000000..8b1f7dc29 --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/cli.py @@ -0,0 +1,160 @@ +"""Command line interface. + + provtest run [options] CONF [CONF...] # execute scenarios + provtest validate CONF [CONF...] # check offline + provtest list CONF [CONF...] # show what a file does + +`validate` and `list` need no root, no network and no client programs, so they +run in a checkout-only CI job -- which is where a typo in a field name or a +reference to a step that has not run yet should be caught, long before a +management node is involved. +""" + +import argparse +import sys + +from . import config, machine, report +from .errors import EXIT_CONFIG, EXIT_OK, ProvTestError + + +def build_parser(): + parser = argparse.ArgumentParser( + prog="provtest", + description="Drive the xCAT provision chain the way a booting node " + "does, one declarative .conf file at a time.") + subparsers = parser.add_subparsers(dest="command") + + run = subparsers.add_parser( + "run", help="execute scenarios against a real management node") + _add_conf_arguments(run) + run.add_argument("-b", "--bind", metavar="ADDR", + help="source address for every request; xcatd names a " + "client by the reverse lookup of this address") + run.add_argument("--timeout", type=float, metavar="SECS", + help="per-attempt timeout (default: 5)") + run.add_argument("--retries", type=int, metavar="N", + help="attempts per step (default: 2)") + run.add_argument("--format", default="tap", choices=sorted(report.REPORTERS), + help="output format (default: tap)") + run.add_argument("-s", "--scenario", action="append", default=[], + metavar="NAME", help="run only this scenario (repeatable)") + run.add_argument("-v", "--verbose", action="count", default=0, + help="print each reply as it arrives") + + validate = subparsers.add_parser( + "validate", help="parse and check .conf files without any network I/O") + _add_conf_arguments(validate) + + listing = subparsers.add_parser( + "list", help="list the scenarios and steps in .conf files") + _add_conf_arguments(listing) + + return parser + + +def _add_conf_arguments(parser): + parser.add_argument("conf", nargs="+", metavar="CONF", + help="a .conf file in INI syntax") + parser.add_argument("--set", action="append", default=[], metavar="KEY=VALUE", + dest="settings", + help="define %%(KEY)s for substitution (repeatable)") + + +def parse_settings(pairs): + settings = {} + for pair in pairs: + if "=" not in pair: + raise ProvTestError("--set needs KEY=VALUE, got %r" % (pair,)) + key, _, value = pair.partition("=") + settings[key.strip()] = value + return settings + + +def main(argv=None): + parser = build_parser() + args = parser.parse_args(argv) + if not args.command: + parser.print_help() + return EXIT_CONFIG + + try: + if args.command == "validate": + return command_validate(args) + if args.command == "list": + return command_list(args) + if args.command == "run": + return command_run(args) + except ProvTestError as exc: + sys.stderr.write("provtest: %s\n" % (exc,)) + return exc.exit_code + parser.print_help() + return EXIT_CONFIG + + +def command_validate(args): + # Raw: a file is checked in a checkout, where no value has been supplied + # for %(name)s yet, so interpolating it here would fail every time. + settings = parse_settings(args.settings) + scenarios, _ = config.load(args.conf, settings, raw=True) + problems = [] + for scenario in scenarios: + problems.extend(machine.validate_scenario(scenario)) + + for problem in problems: + sys.stderr.write("provtest: %s\n" % (problem,)) + if problems: + return EXIT_CONFIG + + needed = sorted(machine.required_variables(scenarios) - set(settings)) + sys.stdout.write("%d scenario(s) OK in %d file(s)\n" + % (len(scenarios), len(args.conf))) + if needed: + sys.stdout.write("required variables: %s\n" % (", ".join(needed),)) + return EXIT_OK + + +def command_list(args): + scenarios, _ = config.load(args.conf, parse_settings(args.settings), raw=True) + for scenario in scenarios: + sys.stdout.write("%s\n" % (scenario.name,)) + if scenario.description: + sys.stdout.write(" %s\n" % (scenario.description,)) + sys.stdout.write(" source: %s\n" % (scenario.source,)) + for step in scenario.steps: + sys.stdout.write(" step %-16s type=%-12s expect=%-4s asserts=%d\n" + % (step.name, step.type, step.expect or "ok", + len(step.assertions))) + needed = sorted(machine.required_variables([scenario])) + if needed: + sys.stdout.write(" requires: %s\n" % (", ".join(needed),)) + return EXIT_OK + + +def command_run(args): + from . import runner # spawns clients and opens sockets; not for list + + scenarios, _ = config.load(args.conf, parse_settings(args.settings)) + if args.scenario: + wanted = set(args.scenario) + missing = wanted - set(s.name for s in scenarios) + if missing: + raise ProvTestError( + "no such scenario: %s" % (", ".join(sorted(missing)),)) + scenarios = [s for s in scenarios if s.name in wanted] + + problems = [] + for scenario in scenarios: + problems.extend(machine.validate_scenario(scenario)) + if problems: + for problem in problems: + sys.stderr.write("provtest: %s\n" % (problem,)) + return EXIT_CONFIG + + reporter = report.make(args.format) + options = runner.RunOptions( + bind=args.bind, + timeout=args.timeout, + retries=args.retries, + verbose=args.verbose, + ) + return runner.run(scenarios, reporter, options) diff --git a/xCAT-test/provtest/src/provtest_lib/config.py b/xCAT-test/provtest/src/provtest_lib/config.py new file mode 100644 index 000000000..c0fce5eed --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/config.py @@ -0,0 +1,242 @@ +"""Reading declarative .conf files with configparser. + +Layout of a file:: + + [vars] optional; %(name)s values, overridable with --set + [defaults] optional; step keys applied to every step + [scenario ] starts a scenario + [step ] belongs to the scenario above it + +Variables use configparser's own `BasicInterpolation`: `%(name)s` is resolved +by the library, out of `[vars]` and of anything passed with `--set`. `$` has no +meaning to it, so `$grubcfg.text` reaches the runtime untouched -- and it has +to, because its value is a reply that has not arrived when the file is read. + +Unknown keys are a hard error rather than a silent skip, matching `xcattest`, +where an unrecognised `check:` fails the case instead of passing it. +""" + +import configparser +import os + +from . import assertions as assertions_mod +from . import machine +from .errors import ConfigError +from .model import Scenario, Step + +#: Keys accepted in a `[scenario ...]` section. +SCENARIO_KEYS = frozenset(["description"]) + +#: Keys `[defaults]` may carry. `type` and `assert` are per-step by nature: a +#: default type would make a mistyped `type=` silently become another protocol, +#: and a default assertion would be made about replies it was never written for. +DEFAULTS_KEYS = frozenset( + key + for step_type in machine.STEP_KEYS + for key in machine.allowed_keys(step_type) +) - frozenset(["type", "assert", "description"]) + + +def load(paths, overrides=None, raw=False): + """Read every path and return `(scenarios, variables)`. + + With `raw`, `%(name)s` is left in place instead of being interpolated, so + `validate` can check a file in a checkout where no value has been supplied + for it yet. + """ + scenarios = [] + variables = {} + for path in paths: + file_scenarios, file_vars = load_file(path, overrides, raw) + scenarios.extend(file_scenarios) + variables.update(file_vars) + variables.update(overrides or {}) + _reject_duplicate_scenarios(scenarios) + return scenarios, variables + + +def load_file(path, overrides=None, raw=False): + """Read one .conf file into `(scenarios, variables)`.""" + if not os.path.isfile(path): + raise ConfigError("no such file", path) + + parser = configparser.ConfigParser( + interpolation=None, + inline_comment_prefixes=(";",), + delimiters=("=",), + ) + parser.optionxform = str.lower + try: + with open(path) as handle: + parser.read_file(handle, source=path) + except configparser.Error as exc: + raise ConfigError(str(exc), path) + + variables = dict(parser["vars"]) if parser.has_section("vars") else {} + variables.update(overrides or {}) + + scenarios = [] + defaults = {} + current = None + steps = [] + + for section in parser.sections(): + items = _items(parser, section, path, raw, variables) + kind, _, name = section.partition(" ") + kind = kind.strip().lower() + name = name.strip() + where = "%s [%s]" % (path, section) + + if section.lower() == "vars": + continue # read above, into `variables` + + if section.lower() == "defaults": + _reject_unknown(items, DEFAULTS_KEYS, where) + defaults = items + continue + + if kind == "scenario": + if not name: + raise ConfigError("a [scenario] section needs a name", where) + if current is not None: + scenarios.append(_build_scenario(current, steps, path)) + _reject_unknown(items, SCENARIO_KEYS, where) + current = (name, items) + steps = [] + continue + + if kind == "step": + if current is None: + raise ConfigError( + "a [step] section must follow a [scenario] section", where) + if not name: + raise ConfigError("a [step] section needs a name", where) + steps.append(_build_step(name, items, defaults, where)) + continue + + raise ConfigError( + "unknown section; expected [vars], [defaults], [scenario ] " + "or [step ]", where) + + if current is not None: + scenarios.append(_build_scenario(current, steps, path)) + elif steps: + raise ConfigError("steps without a scenario", path) + + if not scenarios: + raise ConfigError("no [scenario ...] section found", path) + + return scenarios, variables + + +def _build_scenario(current, steps, path): + name, items = current + if not steps: + raise ConfigError("scenario %r has no steps" % (name,), path) + return Scenario( + name=name, + description=items.get("description", ""), + steps=steps, + source=path, + ) + + +def _build_step(name, items, defaults, where): + step_type = (items.get("type") or "").strip().lower() + if not step_type: + raise ConfigError("step %r has no type=" % (name,), where) + machine.check_type(step_type, where) + + allowed = machine.allowed_keys(step_type) + _reject_unknown(items, allowed, where) + + # A default only applies where the type in hand can use it, so one + # `[defaults] server =` serves dns, tftp, http and xcatreq steps alike + # without making `path =` illegal for the dns step in the same file. + merged = dict((key, value) for key, value in defaults.items() + if key in allowed) + merged.update(items) + + expect = (merged.get("expect") or "").strip().lower() or None + timeout = _float(merged.get("timeout"), 5.0, "timeout", where) + retries = _int(merged.get("retries"), 2, "retries", where) + parsed = assertions_mod.parse_block(merged.pop("assert", ""), where) + + for key in ("type", "expect", "timeout", "retries"): + merged.pop(key, None) + + return Step( + name=name, + type=step_type, + expect=expect, + params=merged, + assertions=parsed, + timeout=timeout, + retries=retries, + source=where, + ) + + +#: configparser's own variable syntax, `%(name)s`. The parser is built with +#: `interpolation=None` and this is applied by hand, per value, because both of +#: the built-in ways of supplying the values -- DEFAULTSECT and `get(vars=...)` +#: -- also make each variable an *option*, so `--set path=/install` would +#: overwrite the step key of the same name. Interpolating explicitly keeps the +#: two namespaces apart: variables are only ever read through `%(...)s`. +INTERPOLATION = configparser.BasicInterpolation() + + +def _items(parser, section, path, raw, variables=None): + """One section's keys, with `%(name)s` resolved out of `variables`.""" + items = {} + for key in parser.options(section): + value = parser.get(section, key, raw=True) + if raw: + items[key] = value + continue + try: + items[key] = INTERPOLATION.before_get( + parser, section, key, value, variables or {}) + except configparser.InterpolationMissingOptionError as exc: + raise ConfigError( + "%%(%s)s is not defined; pass --set %s=" + % (exc.reference, exc.reference), "%s [%s]" % (path, section)) + except configparser.Error as exc: + raise ConfigError(str(exc), "%s [%s]" % (path, section)) + return items + + +def _reject_unknown(items, allowed, where): + unknown = sorted(set(items) - set(allowed)) + if unknown: + raise ConfigError( + "unknown key(s): %s (allowed: %s)" + % (", ".join(unknown), ", ".join(sorted(allowed))), where) + + +def _reject_duplicate_scenarios(scenarios): + seen = {} + for scenario in scenarios: + if scenario.name in seen: + raise ConfigError( + "scenario %r is defined in both %s and %s" + % (scenario.name, seen[scenario.name], scenario.source)) + seen[scenario.name] = scenario.source + + +def _float(value, default, key, where): + if value is None or value == "": + return default + try: + return float(value) + except ValueError: + raise ConfigError("%s= must be a number, got %r" % (key, value), where) + + +def _int(value, default, key, where): + if value is None or value == "": + return default + try: + return int(value, 0) if isinstance(value, str) else int(value) + except ValueError: + raise ConfigError("%s= must be an integer, got %r" % (key, value), where) diff --git a/xCAT-test/provtest/src/provtest_lib/dnsc.py b/xCAT-test/provtest/src/provtest_lib/dnsc.py new file mode 100644 index 000000000..0c029db6d --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/dnsc.py @@ -0,0 +1,120 @@ +"""DNS, asked the way an operator asks: `dig`. + +A node resolves three things before it can be provisioned -- its own name, the +master's name, and whatever the kickstart refers to -- and xcatd then resolves +the node's address backwards to decide which node it is talking to. All four +are one resolver query, so one step type covers the whole stage. + +The query is always sent to a named server. A booting node uses the resolver +DHCP gave it, and the point of the test is that *that* server answers, not that +the host running the test has some resolver that does. +""" + +import re + +from . import proc +from .model import Reply + +HEADER_RE = re.compile(r"->>HEADER<<-\s+opcode:\s*(\w+),\s*status:\s*(\w+),\s*id:\s*(\d+)") +FLAGS_RE = re.compile(r"^;;\s*flags:\s*([^;]*);\s*(.*)$") +COUNT_RE = re.compile(r"(\w+):\s*(\d+)") + + +def build(server, name, rrtype="A", port=53, bind=None, recursion=True, + timeout=5.0, tries=1): + """The `dig` command line for one query.""" + dig = proc.require("dig", "DNS") + argv = [dig, "@%s" % (server,), name, rrtype.upper()] + if port and int(port) != 53: + argv += ["-p", str(int(port))] + if bind: + argv += ["-b", str(bind)] + argv += [ + "+noall", "+comments", "+answer", "+authority", + "+tries=%d" % (max(1, int(tries)),), + "+time=%d" % (max(1, int(round(float(timeout)))),), + "+norecurse" if not recursion else "+recurse", + ] + return argv + + +def query(server, name, rrtype="A", port=53, bind=None, recursion=True, + timeout=5.0): + """Run one query and decode what `dig` printed into a `Reply`.""" + argv = build(server, name, rrtype, port, bind, recursion, timeout) + done = proc.run(argv, timeout=float(timeout) + 5.0) + reply = decode(done.text, name=name, rrtype=rrtype, server=server) + reply.sent = done.command() + reply.raw = done.out + if done.timed_out: + reply.ok = False + reply.error = "dig did not return within the timeout" + elif done.rc != 0 and not reply.fields.get("status"): + reply.ok = False + reply.error = _why(done) + return reply + + +def decode(text, name="", rrtype="", server=""): + """Turn `dig +noall +comments +answer +authority` output into a `Reply`.""" + status = "" + flags = "" + counts = {} + names, ttls, types, data, records = [], [], [], [], [] + + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + continue + if stripped.startswith(";"): + match = HEADER_RE.search(stripped) + if match: + status = match.group(2) + continue + match = FLAGS_RE.match(stripped) + if match: + flags = match.group(1).strip() + counts = dict((key.upper(), int(value)) + for key, value in COUNT_RE.findall(match.group(2))) + continue + + parts = stripped.split(None, 4) + if len(parts) < 5 or parts[2].upper() != "IN": + continue + # An authority section shares the record syntax, and SOA records there + # are how a resolver says "this zone is mine and the name is not in + # it", so they are kept apart rather than counted as answers. + if parts[3].upper() == "SOA": + continue + names.append(parts[0].rstrip(".")) + ttls.append(int(parts[1]) if parts[1].isdigit() else 0) + types.append(parts[3].upper()) + data.append(parts[4].strip().rstrip(".") if parts[3].upper() != "TXT" + else parts[4].strip()) + records.append(stripped) + + fields = { + "status": status, + "rcode": status, + "flags": flags, + "count": counts.get("ANSWER", len(data)), + "authority": counts.get("AUTHORITY", 0), + "name": names, + "ttl": ttls, + "type": types, + "data": data, + "answers": records, + "question": name, + "server": server, + } + ok = status == "NOERROR" + return Reply(kind="dns", fields=fields, ok=ok, + error="" if status else "no answer was received") + + +def _why(done): + """The first line `dig` wrote to stderr, or a fallback.""" + for line in (done.errtext + done.text).splitlines(): + if "error" in line.lower() or "could be reached" in line.lower(): + return line.strip().lstrip(";").strip() + return "dig exited %d" % (done.rc,) diff --git a/xCAT-test/provtest/src/provtest_lib/errors.py b/xCAT-test/provtest/src/provtest_lib/errors.py new file mode 100644 index 000000000..552e22eaf --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/errors.py @@ -0,0 +1,46 @@ +"""Exceptions and exit codes. + +The codes are dhcptest's, so a caller driving both suites reads one table: + + 0 every assertion held + 1 at least one assertion failed + 2 the configuration or the command line is wrong + 3 this host cannot run the test at all + +3 is deliberately not 0. A host with no `dig` reports that it tested nothing +rather than reporting a green run, because a pass that proves nothing is worse +than a failure that says why. +""" + +EXIT_OK = 0 +EXIT_FAILED = 1 +EXIT_CONFIG = 2 +EXIT_UNSUPPORTED = 3 + + +class ProvTestError(Exception): + """Base class: every failure this tool raises on purpose.""" + + exit_code = EXIT_CONFIG + + +class ConfigError(ProvTestError): + """A .conf file says something that cannot be carried out.""" + + exit_code = EXIT_CONFIG + + def __init__(self, message, source=""): + self.message = message + self.source = source + Exception.__init__(self, self.__str__()) + + def __str__(self): + if self.source: + return "%s: %s" % (self.source, self.message) + return self.message + + +class UnsupportedError(ProvTestError): + """The host is missing something the run cannot go on without.""" + + exit_code = EXIT_UNSUPPORTED diff --git a/xCAT-test/provtest/src/provtest_lib/httpc.py b/xCAT-test/provtest/src/provtest_lib/httpc.py new file mode 100644 index 000000000..ea3530045 --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/httpc.py @@ -0,0 +1,165 @@ +"""HTTP, fetched the way an installer fetches: `curl`. + +Two different things are served over HTTP during a provision and both are +tested here. grub2-http and xnba fetch the kernel and initrd from +`/tftpboot/...`; the installer fetches the repository and its kickstart or +preseed from `/install/...`. They are two Apache aliases over the same port, +and a cluster that has moved off port 80 has to have moved the URL in the +kernel command line with it -- which is why the port comes out of the config +the node was handed, not out of a constant. + +The body is written to a file rather than captured through a pipe, so a digest +of a kernel can be compared against the same kernel fetched over TFTP without +either ever being decoded as text. +""" + +import hashlib +import os +import tempfile + +from . import proc +from .model import Reply + +#: Above this, a body is measured and digested but not carried as text. +TEXT_LIMIT = 1 << 20 + +#: What `-w` prints, in this order, on one line. +WRITE_OUT = "%{http_code}\\t%{size_download}\\t%{content_type}\\t%{url_effective}\\n" + + +def url_for(server, path, port=80, scheme="http"): + """Join a server, a port and a path into a URL.""" + text = str(path) + if text.startswith("http://") or text.startswith("https://"): + return text + if not text.startswith("/"): + text = "/" + text + port = int(port or 80) + if (scheme == "http" and port == 80) or (scheme == "https" and port == 443): + return "%s://%s%s" % (scheme, server, text) + return "%s://%s:%d%s" % (scheme, server, port, text) + + +def build(url, method="GET", bind=None, headers=(), timeout=10.0, + body_file=None, header_file=None, insecure=False): + """The `curl` command line for one request.""" + curl = proc.require("curl", "HTTP") + argv = [curl, "-sS", "--globoff", "--max-time", str(int(float(timeout))), + "-o", body_file, "-D", header_file, "-w", WRITE_OUT] + if bind: + argv += ["--interface", str(bind)] + if insecure: + argv += ["-k"] + method = (method or "GET").upper() + if method == "HEAD": + argv += ["-I"] + elif method != "GET": + argv += ["-X", method] + for header in headers: + argv += ["-H", header] + argv += [url] + return argv + + +def fetch(url, method="GET", bind=None, headers=(), timeout=10.0, + insecure=False): + """Make one request and return a `Reply` describing the response.""" + body_handle, body_file = tempfile.mkstemp(prefix="provtest-http-") + head_handle, header_file = tempfile.mkstemp(prefix="provtest-hdr-") + os.close(body_handle) + os.close(head_handle) + try: + argv = build(url, method, bind, headers, timeout, body_file, + header_file, insecure) + done = proc.run(argv, timeout=float(timeout) + 5.0) + return _decode(done, url, body_file, header_file) + finally: + for path in (body_file, header_file): + try: + os.unlink(path) + except OSError: + pass + + +def _decode(done, url, body_file, header_file): + status, size, content_type, effective = _write_out(done.text, url) + disk_size = os.path.getsize(body_file) if os.path.exists(body_file) else 0 + body = b"" + if 0 < disk_size <= TEXT_LIMIT: + with open(body_file, "rb") as handle: + body = handle.read() + digest = _digest(body_file) if disk_size else "" + + error = "" + if done.timed_out: + error = "the request did not finish within the timeout" + elif done.rc != 0: + error = (done.errtext.strip().replace("curl: ", "", 1) + or "curl exited %d" % (done.rc,)) + + fields = { + "status": status, + "size": size or disk_size, + "sha256": digest, + "text": body.decode("utf-8", "replace"), + "url": effective, + "header": _headers(header_file), + "content_type": content_type, + "error": error, + "ok": bool(status and 200 <= status < 400), + } + return Reply(kind="http", fields=fields, ok=fields["ok"], error=error, + sent=done.command(), raw=body) + + +def _write_out(text, url): + """The last `-w` line: status, size, content type, effective URL. + + The last line, not the first: a redirect that `curl` was asked to follow + would print one per hop, and the response under test is the final one. + """ + for line in reversed(text.strip().splitlines()): + parts = line.split("\t") + if len(parts) != 4: + continue + try: + status = int(parts[0]) + except ValueError: + status = 0 + try: + size = int(parts[1]) + except ValueError: + size = 0 + return status, size, parts[2].strip(), parts[3].strip() or url + return 0, 0, "", url + + +def _headers(path): + """The response headers, lowercased, as a mapping. + + Only the last response block is kept, for the same reason as above. + """ + headers = {} + if not os.path.exists(path): + return headers + with open(path, "rb") as handle: + text = handle.read().decode("utf-8", "replace") + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + continue + if stripped.upper().startswith("HTTP/"): + headers = {} + continue + key, _, value = stripped.partition(":") + if value: + headers[key.strip().lower()] = value.strip() + return headers + + +def _digest(path): + digest = hashlib.sha256() + with open(path, "rb") as handle: + for chunk in iter(lambda: handle.read(65536), b""): + digest.update(chunk) + return digest.hexdigest() diff --git a/xCAT-test/provtest/src/provtest_lib/machine.py b/xCAT-test/provtest/src/provtest_lib/machine.py new file mode 100644 index 000000000..aee839db5 --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/machine.py @@ -0,0 +1,229 @@ +"""What a step of each type may say, and what its reply will hold. + +This is the offline half of the tool. `validate` runs it in a checkout with no +network, no root and no server, so a typo in a field name or a step that names +no server is a configuration error reported before anything is sent. + +Every table here is declarative on purpose: adding a protocol means adding a +row, not a branch in the runner. +""" + +from . import subst +from .errors import ConfigError + +#: Keys every step may carry, whatever its type. `bind` is the client's own +#: address: xcatd names a client by the reverse lookup of the address it +#: connected from, so which end of the veth pair a request leaves by is itself +#: under test, and cannot be left to the kernel's source selection. +COMMON_KEYS = frozenset(["type", "expect", "timeout", "retries", "assert", + "bind"]) + +#: Per type: the keys it accepts, beyond COMMON_KEYS. +STEP_KEYS = { + "dns": frozenset(["server", "port", "name", "rrtype", "recursion"]), + "tftp": frozenset(["server", "port", "path", "mode"]), + "http": frozenset(["server", "port", "path", "url", "method", "header", + "insecure"]), + "xcatreq": frozenset(["server", "port", "command", "element", "raw", + "callback_port", "callback_listen", "callback_reply", + "callback_wait", "cert", "key", "source_port"]), + "monitor": frozenset(["server", "port", "send", "source_port"]), + "flowrequest": frozenset(["server", "port", "message", "source_port", + "replies"]), + "findme": frozenset(["server", "port", "payload", "encoding", "source_port", + "callback_listen", "callback_wait"]), + "extract": frozenset(["from", "pattern", "group"]), + "sleep": frozenset(["duration"]), + "noop": frozenset([]), +} + +#: Keys naming a value that must be present for the step to mean anything. +REQUIRED_KEYS = { + "dns": ("server", "name"), + "tftp": ("server", "path"), + "http": (), # url, or server+path; checked below + "xcatreq": ("server", "command"), + "monitor": ("server", "send"), + "flowrequest": ("server",), + "findme": ("server",), + "extract": ("from", "pattern"), + "sleep": ("duration",), + "noop": (), +} + +#: Per type: the reply fields a `.conf` may name, as an assertion target or +#: through `$step.field`. A name not in this set is a typo, and is reported as +#: one rather than silently never matching. +REPLY_FIELDS = { + "dns": frozenset(["status", "flags", "count", "data", "type", "ttl", + "name", "question", "answers", "authority", "server", + "rcode", "raw"]), + "tftp": frozenset(["ok", "size", "sha256", "text", "error", "path", + "server", "raw"]), + "http": frozenset(["status", "size", "sha256", "text", "url", "header", + "content_type", "error", "ok", "raw"]), + "xcatreq": frozenset(["destiny", "kernel", "initrd", "kcmdline", + "imgserver", "name", "error", "serverdone", + "elements", "data", "text", "handshake", "ok", + "callback_seen", "callback_data", "raw"]), + "monitor": frozenset(["greeting", "lines", "raw", "text", "ok", "closed", + "error"]), + "flowrequest": frozenset(["replies", "count", "ok", "error", "raw"]), + "findme": frozenset(["callbacks", "count", "ok", "error", "raw"]), + "extract": frozenset(["value", "groups", "matched", "count"]), + "sleep": frozenset([]), + "noop": frozenset([]), +} + +#: Reply fields that are a mapping rather than a scalar or a list, and so are +#: addressed as `header.content-type` or `elements.destiny`. +MAPPING_FIELDS = frozenset(["header", "elements"]) + +#: `expect=` values. `fail` is not an error: a scenario asserting that a path +#: outside the tftp root is refused wants the refusal to be the passing result. +EXPECTATIONS = frozenset(["ok", "fail", "any"]) + + +def known_step_types(): + return frozenset(STEP_KEYS) + + +def allowed_keys(step_type): + return COMMON_KEYS | STEP_KEYS.get(step_type, frozenset()) + + +def all_reply_fields(): + """Every field name any step type can produce.""" + names = set() + for fields in REPLY_FIELDS.values(): + names.update(fields) + return names + + +def check_field(field): + """True when `field` is a name some step type's reply could carry. + + Cross-type: a `$step.field` reference is resolved against whichever step + it names, and which step that is cannot be known from the text alone when + the reference is written in `[defaults]`. Validation is therefore that the + name exists *somewhere*; naming a field of the wrong type fails at run + time, where the reply is in hand and the message can say so. + """ + head, _, tail = field.rpartition(".") + if head: + if head in MAPPING_FIELDS: + return bool(tail) + if head not in all_reply_fields(): + return False + try: + int(tail) + except ValueError: + return False + return True + return field in all_reply_fields() + + +def validate_scenario(scenario): + """Every problem in one scenario, as a list of strings. + + Returns rather than raises, so one pass over a file reports everything + wrong with it instead of the first thing. + """ + problems = [] + seen = set() + for step in scenario.steps: + where = "%s scenario %r step %r" % (scenario.source, scenario.name, step.name) + if step.name in seen: + problems.append("%s: step %r is defined twice" % (where, step.name)) + seen.add(step.name) + + problems.extend(_validate_required(step, where)) + problems.extend(_validate_expect(step, where)) + problems.extend(_validate_targets(step, where)) + problems.extend(_validate_references(step, scenario, seen, where)) + return problems + + +def _validate_required(step, where): + problems = [] + for key in REQUIRED_KEYS.get(step.type, ()): + if not step.param(key): + problems.append("%s: a %s step needs %s=" % (where, step.type, key)) + if step.type == "http" and not step.param("url"): + if not (step.param("server") and step.param("path")): + problems.append( + "%s: an http step needs url=, or server= and path=" % (where,)) + if step.type == "extract": + group = step.param("group") + if group not in (None, ""): + try: + int(group) + except ValueError: + problems.append("%s: group= must be a number, got %r" + % (where, group)) + return problems + + +def _validate_expect(step, where): + if step.expect and step.expect not in EXPECTATIONS: + return ["%s: unknown expect=%r (known: %s)" + % (where, step.expect, ", ".join(sorted(EXPECTATIONS)))] + return [] + + +def _validate_targets(step, where): + problems = [] + fields = REPLY_FIELDS.get(step.type, frozenset()) + for assertion in step.assertions: + target = assertion.target + if target.startswith("$"): + continue + head, _, tail = target.rpartition(".") + if head in MAPPING_FIELDS: + if head not in fields: + problems.append("%s: a %s reply carries no %s" + % (where, step.type, head)) + continue + base = head or target + if base not in fields: + problems.append( + "%s: a %s reply has no field %r (it has: %s)" + % (where, step.type, target, ", ".join(sorted(fields)))) + return problems + + +def _validate_references(step, scenario, seen, where): + """Reject a `$step.field` naming a step that does not run before this one.""" + problems = [] + texts = [assertion.value for assertion in step.assertions] + texts.extend(str(value) for value in step.params.values()) + texts.extend(assertion.target for assertion in step.assertions) + for text in texts: + for name, field in subst.references(text): + if name not in seen: + problems.append( + "%s: $%s.%s refers to a step that has not run yet" + % (where, name, field)) + elif not check_field(field): + problems.append("%s: no reply field is called .%s" % (where, field)) + return problems + + +def required_variables(scenarios): + """Every `%(name)s` a run of these scenarios would have to be given.""" + names = set() + for scenario in scenarios: + for step in scenario.steps: + for value in step.params.values(): + names.update(subst.variables(value)) + for assertion in step.assertions: + names.update(subst.variables(assertion.value)) + names.update(subst.variables(assertion.target)) + return names + + +def check_type(step_type, where=""): + if step_type not in STEP_KEYS: + raise ConfigError( + "unknown step type %r (known: %s)" + % (step_type, ", ".join(sorted(STEP_KEYS))), where) diff --git a/xCAT-test/provtest/src/provtest_lib/model.py b/xCAT-test/provtest/src/provtest_lib/model.py new file mode 100644 index 000000000..7d53a6f1b --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/model.py @@ -0,0 +1,184 @@ +"""Plain data structures shared across the tool. + +No network and no subprocess: everything here can be built by hand in a unit +test, which is how the assertion language is tested without a server. + +Written for Python 3.6, so no dataclasses. +""" + +from .errors import ConfigError + + +class Assertion(object): + """One line of the assertion mini-language: `target op value`. + + For example:: + + status == NOERROR + data == 10.99.1.11 + text contains destiny=install + sha256 == $tftpfetch.sha256 + """ + + __slots__ = ("target", "op", "value", "source", "text") + + def __init__(self, target, op, value, source="", text=""): + self.target = target + self.op = op + self.value = value + self.source = source + self.text = text or self.render() + + def render(self): + if self.value is None: + return "%s %s" % (self.target, self.op) + return "%s %s %s" % (self.target, self.op, self.value) + + def __repr__(self): + return "Assertion(%r)" % (self.render(),) + + +class Step(object): + """One client action plus the assertions made about what came back.""" + + __slots__ = ("name", "type", "expect", "params", "assertions", "timeout", + "retries", "source") + + def __init__(self, name, type, expect, params, assertions, + timeout=5.0, retries=2, source=""): + self.name = name + self.type = type + self.expect = expect + self.params = dict(params) + self.assertions = tuple(assertions) + self.timeout = timeout + self.retries = retries + self.source = source + + def param(self, key, default=None): + value = self.params.get(key, default) + if isinstance(value, str): + value = value.strip() + return value + + def __repr__(self): + return "Step(%r, type=%r, expect=%r)" % (self.name, self.type, self.expect) + + +class Scenario(object): + """A named sequence of steps, run in order against one server.""" + + __slots__ = ("name", "description", "steps", "source") + + def __init__(self, name, description="", steps=(), source=""): + self.name = name + self.description = description + self.steps = tuple(steps) + self.source = source + + def step(self, name): + for step in self.steps: + if step.name == name: + return step + raise ConfigError("no such step %r in scenario %r" % (name, self.name)) + + def __repr__(self): + return "Scenario(%r, steps=%d)" % (self.name, len(self.steps)) + + +class Reply(object): + """What one step got back, as a flat namespace of fields. + + Unlike DHCP, the five protocols here share no header, so a reply is a bag + of named values rather than a fixed record. What each step type puts in it + is declared in `machine.REPLY_FIELDS`, so a `.conf` naming a field no step + of that type ever produces is rejected before any socket is opened. + + A field may hold a list -- DNS answers, monitor lines, findme callbacks. + Indexing is written `data.0`, and the bare name of a list resolves to its + first element, so `data == 10.99.1.11` reads the way an operator expects + for the ordinary one-answer case. + """ + + __slots__ = ("kind", "fields", "ok", "error", "sent", "attempt", "raw") + + def __init__(self, kind="", fields=None, ok=True, error="", sent="", + attempt=0, raw=b""): + self.kind = kind + self.fields = dict(fields or {}) + self.ok = ok + self.error = error + self.sent = sent + self.attempt = attempt + self.raw = raw + + def field(self, name): + """Resolve one `$step.` reference, or raise KeyError. + + `a.b` is looked up whole first, so a header written into the reply as + `header.content-type` wins over any attempt to index a list called + `header`. Only then is the name split into container and index. + """ + if name in self.fields: + return self._first(self.fields[name]) + head, _, tail = name.rpartition(".") + if head and head in self.fields: + container = self.fields[head] + if isinstance(container, (list, tuple)): + try: + index = int(tail) + except ValueError: + raise KeyError("%s is not an index into .%s" % (tail, head)) + if index >= len(container) or index < -len(container): + raise KeyError( + "%s reply has %d item(s) in .%s, so .%s does not exist" + % (self.kind, len(container), head, name)) + return container[index] + if isinstance(container, dict): + if tail not in container: + raise KeyError("%s reply has no .%s" % (self.kind, name)) + return container[tail] + raise KeyError("%s reply has no field .%s" % (self.kind or "this", name)) + + def has(self, name): + try: + value = self.field(name) + except KeyError: + return False + # An empty list is "nothing came back", which is absence, not a value. + if isinstance(value, (list, tuple)): + return bool(value) + return value not in (None, "") + + @staticmethod + def _first(value): + if isinstance(value, (list, tuple)): + return value[0] if value else [] + return value + + def summary(self): + parts = [] + for key in sorted(self.fields): + value = self.fields[key] + if key in ("text", "raw", "body"): + continue + parts.append("%s=%s" % (key, _short(value))) + head = "%s %s" % (self.kind or "reply", "ok" if self.ok else "failed") + if self.error: + head += " (%s)" % (self.error,) + return "%s %s" % (head, " ".join(parts)) + + def __repr__(self): + return "Reply(%s)" % (self.summary(),) + + +def _short(value, limit=60): + if isinstance(value, (list, tuple)): + text = "[" + ", ".join(str(item) for item in value) + "]" + elif isinstance(value, dict): + text = "{" + ", ".join("%s=%s" % (k, v) for k, v in sorted(value.items())) + "}" + else: + text = str(value) + if len(text) > limit: + text = text[:limit - 3] + "..." + return text diff --git a/xCAT-test/provtest/src/provtest_lib/netutil.py b/xCAT-test/provtest/src/provtest_lib/netutil.py new file mode 100644 index 000000000..3642e5227 --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/netutil.py @@ -0,0 +1,91 @@ +"""Address arithmetic, and nothing that touches a network. + +Kept separate from the clients so that the assertion language can compare +addresses in a unit test with no interface, no root and no server. +""" + +import ipaddress + + +def parse_ip(text): + """An `IPv4Address`, or None when `text` is not one.""" + try: + return ipaddress.IPv4Address(str(text).strip()) + except (ipaddress.AddressValueError, ValueError): + return None + + +def ip_in(address, network): + """True when `address` falls inside the CIDR block `network`.""" + ip = parse_ip(address) + if ip is None: + return False + try: + net = ipaddress.IPv4Network(str(network).strip(), strict=False) + except (ipaddress.AddressValueError, ipaddress.NetmaskValueError, ValueError): + return False + return ip in net + + +def parse_range(text): + """`first-last` as a pair of addresses, or None.""" + if text is None or "-" not in str(text): + return None + first, _, last = str(text).strip().partition("-") + low, high = parse_ip(first), parse_ip(last) + if low is None or high is None: + return None + return low, high + + +def ip_in_range(address, text): + """True when `address` is inside an inclusive `first-last` range.""" + bounds = parse_range(text) + ip = parse_ip(address) + if bounds is None or ip is None: + return False + low, high = bounds + return low <= ip <= high + + +def hex_ip(address, upper=True): + """The eight-digit hex a PXE or grub2 loader asks for its address by. + + This is the encoding the loader itself performs, not xCAT's: 10.99.1.11 + becomes 0A63010B. It is here so a scenario can say what the name should be + without the fixture having to compute it, and so the tool never has to ask + xCAT what name it wrote. + """ + ip = parse_ip(address) + if ip is None: + raise ValueError("not an IPv4 address: %r" % (address,)) + text = "%08X" % (int(ip),) + return text if upper else text.lower() + + +def reverse_name(address): + """The IN-ADDR.ARPA name a resolver is asked for a PTR by.""" + ip = parse_ip(address) + if ip is None: + raise ValueError("not an IPv4 address: %r" % (address,)) + return ip.reverse_pointer + + +def normalise_mac(text): + """A MAC in the lowercase colon form, or ValueError.""" + raw = str(text).strip().lower().replace("-", ":").replace(".", ":") + parts = [part for part in raw.split(":") if part != ""] + if len(parts) != 6: + raise ValueError("not a MAC address: %r" % (text,)) + octets = [] + for part in parts: + if len(part) > 2: + raise ValueError("not a MAC address: %r" % (text,)) + value = int(part, 16) + octets.append("%02x" % (value,)) + return ":".join(octets) + + +def dashed_mac(text): + """The `01-aa-bb-...` form a grub2 or pxelinux per-MAC file is named by.""" + return normalise_mac(text).replace(":", "-") diff --git a/xCAT-test/provtest/src/provtest_lib/proc.py b/xCAT-test/provtest/src/provtest_lib/proc.py new file mode 100644 index 000000000..3a4e352cc --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/proc.py @@ -0,0 +1,92 @@ +"""The one place a child process is started. + +DNS, HTTP and TFTP are driven by the clients an operator would reach for -- +`dig`, `curl`, `tftp` -- rather than by a hand-rolled implementation of each +protocol. What is under test is whether a real client gets a usable answer, +so using the real client removes a whole class of "the test's own parser was +wrong" from the result. + +Confining the spawning to this module is what makes that safe. Output is read +from a pipe with a timeout and a hard kill, never through a shell, and every +argument list is built as a list, so nothing a `.conf` supplies can be +interpreted as a command. +""" + +import os +import shutil +import subprocess + +from .errors import UnsupportedError + + +class Completed(object): + """What a client program did: exit status and the two streams, as bytes.""" + + __slots__ = ("argv", "rc", "out", "err", "timed_out") + + def __init__(self, argv, rc, out=b"", err=b"", timed_out=False): + self.argv = list(argv) + self.rc = rc + self.out = out + self.err = err + self.timed_out = timed_out + + @property + def text(self): + return self.out.decode("utf-8", "replace") + + @property + def errtext(self): + return self.err.decode("utf-8", "replace") + + def command(self): + return " ".join(self.argv) + + def __repr__(self): + return "Completed(%r, rc=%r)" % (self.command(), self.rc) + + +def which(program): + """The path to a client program, or None.""" + return shutil.which(program) + + +def require(program, why): + """The path to a client program, or a skip-the-host error saying why.""" + path = which(program) + if path is None: + raise UnsupportedError( + "%s is not installed, so %s cannot be tested" % (program, why)) + return path + + +def run(argv, timeout=10.0, stdin=None): + """Run a client program and collect what it said. + + A program that outlives its timeout is killed and reported as timed out + rather than raising, because "the server did not answer" is a result the + scenario may be asserting on. + """ + env = dict(os.environ) + # A localised client would report its errors in a language the parsers + # here do not read. + env["LC_ALL"] = "C" + env["LANG"] = "C" + try: + child = subprocess.Popen( + list(argv), + stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + ) + except OSError as exc: + return Completed(argv, 127, b"", str(exc).encode()) + + try: + out, err = child.communicate(input=stdin, timeout=timeout) + return Completed(argv, child.returncode, out, err) + except subprocess.TimeoutExpired: + child.kill() + out, err = child.communicate() + return Completed(argv, 124, out, err, timed_out=True) diff --git a/xCAT-test/provtest/src/provtest_lib/report.py b/xCAT-test/provtest/src/provtest_lib/report.py new file mode 100644 index 000000000..3df836196 --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/report.py @@ -0,0 +1,216 @@ +"""Result reporting: TAP 13 by default, plus human and JSON forms. + +TAP is the default because `prove` consumes it directly, and because it is what +dhcptest emits, so a driver that already collects one suite's output collects +this one unchanged. + +The failure diagnostic carries expected, received, the reply's fields and the +command or request that produced it, so a red test can be understood from the +log without re-running it -- which matters most in CI, where the network it ran +on no longer exists by the time anyone reads the result. +""" + +import json +import sys + +from .errors import EXIT_FAILED, EXIT_OK + + +class Record(object): + """One assertion outcome, with everything needed to explain it.""" + + __slots__ = ("scenario", "step", "description", "ok", "skip_reason", + "expected", "actual", "detail", "reply", "sent", "attempt", + "attempts") + + def __init__(self, scenario, step, description, ok, skip_reason=None, + expected=None, actual=None, detail="", reply=None, sent="", + attempt=0, attempts=0): + self.scenario = scenario + self.step = step + self.description = description + self.ok = ok + self.skip_reason = skip_reason + self.expected = expected + self.actual = actual + self.detail = detail + self.reply = reply + self.sent = sent + self.attempt = attempt + self.attempts = attempts + + @property + def name(self): + return "%s/%s: %s" % (self.scenario, self.step, self.description) + + +class Reporter(object): + """Base reporter: counts, and an exit code.""" + + def __init__(self, stream=None): + self.stream = stream or sys.stdout + self.records = [] + self.passed = 0 + self.failed = 0 + self.skipped = 0 + + def add(self, record): + self.records.append(record) + if record.skip_reason is not None: + self.skipped += 1 + elif record.ok: + self.passed += 1 + else: + self.failed += 1 + self.emit(record) + + def emit(self, record): + raise NotImplementedError + + def start(self): + pass + + def finish(self): + pass + + def exit_code(self): + return EXIT_FAILED if self.failed else EXIT_OK + + def write(self, text): + self.stream.write(text) + self.stream.flush() + + +class TapReporter(Reporter): + """TAP version 13, with a trailing plan and YAML failure diagnostics.""" + + def start(self): + self.write("TAP version 13\n") + + def emit(self, record): + number = len(self.records) + status = "ok" if (record.ok or record.skip_reason is not None) else "not ok" + line = "%s %d - %s" % (status, number, record.name) + if record.skip_reason is not None: + line += " # SKIP %s" % (record.skip_reason,) + self.write(line + "\n") + if not record.ok and record.skip_reason is None: + self.write(_yaml_block(record)) + + def finish(self): + self.write("1..%d\n" % (len(self.records),)) + self.write("# passed %d, failed %d, skipped %d\n" + % (self.passed, self.failed, self.skipped)) + + +class PrettyReporter(Reporter): + """Human-first output carrying the same information as the TAP form.""" + + def __init__(self, stream=None): + Reporter.__init__(self, stream) + self._scenario = None + + def emit(self, record): + if record.scenario != self._scenario: + self._scenario = record.scenario + self.write("\n%s\n" % (record.scenario,)) + if record.skip_reason is not None: + self.write(" SKIP %s/%s -- %s\n" + % (record.step, record.description, record.skip_reason)) + return + mark = "PASS" if record.ok else "FAIL" + self.write(" %s %s: %s\n" % (mark, record.step, record.description)) + if not record.ok: + for line in _diagnostic_lines(record): + self.write(" %s\n" % (line,)) + + def finish(self): + self.write("\nPROVTEST SUMMARY: total=%d passed=%d failed=%d skipped=%d\n" + % (len(self.records), self.passed, self.failed, self.skipped)) + + +class JsonReporter(Reporter): + """One JSON object per run, for callers that would rather not parse text.""" + + def emit(self, record): + pass + + def finish(self): + payload = { + "total": len(self.records), + "passed": self.passed, + "failed": self.failed, + "skipped": self.skipped, + "records": [_as_dict(record) for record in self.records], + } + self.write(json.dumps(payload, indent=2, sort_keys=True) + "\n") + + +REPORTERS = { + "tap": TapReporter, + "pretty": PrettyReporter, + "json": JsonReporter, +} + + +def make(name, stream=None): + try: + return REPORTERS[name](stream) + except KeyError: + raise ValueError("unknown report format %r" % (name,)) + + +# --------------------------------------------------------------------------- +# diagnostics + + +def _diagnostic_lines(record): + lines = [] + if record.expected is not None: + lines.append("expected: %r" % (record.expected,)) + lines.append("received: %r" % (record.actual,)) + if record.detail: + lines.append("detail: %s" % (_oneline(record.detail),)) + if record.reply is not None: + lines.append("reply: %s" % (_oneline(record.reply.summary()),)) + else: + lines.append("reply: none received") + if record.sent: + lines.append("sent: %s" % (_oneline(record.sent),)) + if record.attempts: + lines.append("attempt: %d of %d" % (record.attempt, record.attempts)) + return lines + + +def _yaml_block(record): + out = [" ---\n"] + for line in _diagnostic_lines(record): + key, _, value = line.partition(":") + out.append(" %s: %s\n" % (key.strip(), value.strip())) + out.append(" ...\n") + return "".join(out) + + +def _oneline(text, limit=400): + """Collapse a value to one line, because TAP's diagnostics are per line.""" + flat = " ".join(str(text).split()) + if len(flat) > limit: + flat = flat[:limit - 3] + "..." + return flat + + +def _as_dict(record): + return { + "scenario": record.scenario, + "step": record.step, + "assertion": record.description, + "ok": bool(record.ok), + "skip": record.skip_reason, + "expected": record.expected, + "actual": record.actual, + "detail": record.detail, + "reply": record.reply.summary() if record.reply is not None else None, + "sent": record.sent, + "attempt": record.attempt, + "attempts": record.attempts, + } diff --git a/xCAT-test/provtest/src/provtest_lib/runner.py b/xCAT-test/provtest/src/provtest_lib/runner.py new file mode 100644 index 000000000..5a8a54121 --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/runner.py @@ -0,0 +1,354 @@ +"""Executing scenarios: one step at a time, in the order they were written. + +A step is dispatched to the client for its protocol, its reply is bound under +the step's name so later steps can refer to it, and its assertions are then +evaluated against that reply. Binding before asserting is what makes the +cross-stage cases work: a TFTP step fetches a grub config, an extract step +pulls the kernel path out of it, and an HTTP step fetches exactly that path -- +none of which is possible if a step can only see values written in the file. +""" + +import re +import sys +import time + +from . import assertions as assertions_mod +from . import dnsc, httpc, machine, report, subst, tftpc, xcatc +from .errors import ConfigError, ProvTestError +from .model import Reply + +#: Step types whose "ok" is decided by the step itself rather than a server. +LOCAL_TYPES = frozenset(["extract", "sleep", "noop"]) + + +class RunOptions(object): + """Command-line overrides, applied on top of what the .conf says.""" + + __slots__ = ("bind", "timeout", "retries", "verbose") + + def __init__(self, bind=None, timeout=None, retries=None, verbose=0): + self.bind = bind + self.timeout = timeout + self.retries = retries + self.verbose = verbose + + +def run(scenarios, reporter, options): + """Run every scenario and return the process exit code.""" + reporter.start() + for scenario in scenarios: + _run_scenario(scenario, reporter, options) + reporter.finish() + return reporter.exit_code() + + +def _run_scenario(scenario, reporter, options): + context = subst.Context() + for step in scenario.steps: + try: + reply, attempt, attempts = _execute(step, context, options) + except ProvTestError as exc: + reporter.add(report.Record( + scenario.name, step.name, "step could not be run", ok=False, + detail=str(exc))) + return + context.bind(step.name, reply) + if options.verbose: + sys.stderr.write("# %s/%s: %s\n" + % (scenario.name, step.name, reply.summary())) + + extras = {"attempt": attempt} + _check_expectation(scenario, step, reply, reporter, attempt, attempts) + for assertion in step.assertions: + try: + result = assertions_mod.evaluate(assertion, reply, context, extras) + except ConfigError as exc: + reporter.add(report.Record( + scenario.name, step.name, assertion.render(), ok=False, + detail=str(exc), reply=reply, sent=reply.sent)) + continue + reporter.add(report.Record( + scenario.name, step.name, assertion.render(), ok=result.ok, + expected=result.expected, actual=result.actual, + detail=result.detail, reply=reply, sent=reply.sent, + attempt=attempt, attempts=attempts)) + + +def _check_expectation(scenario, step, reply, reporter, attempt, attempts): + """Report the transport outcome itself, unless the step waived it. + + The default is `ok`, so a step whose server never answered fails even when + the author wrote no assertion about it. Silence is not a pass. + """ + expect = step.expect or ("any" if step.type in LOCAL_TYPES else "ok") + if expect == "any": + return + wanted = (expect == "ok") + description = "expect %s" % (expect,) + reporter.add(report.Record( + scenario.name, step.name, description, ok=(reply.ok == wanted), + expected=expect, actual="ok" if reply.ok else "failed", + detail=reply.error, reply=reply, sent=reply.sent, + attempt=attempt, attempts=attempts)) + + +def _execute(step, context, options): + """Run one step, retrying only a transport that was expected to work.""" + retries = options.retries if options.retries is not None else step.retries + retries = max(1, int(retries)) + expect = step.expect or ("any" if step.type in LOCAL_TYPES else "ok") + + reply = None + for attempt in range(1, retries + 1): + reply = _dispatch(step, context, options) + reply.attempt = attempt + if expect != "ok" or reply.ok or attempt == retries: + return reply, attempt, retries + time.sleep(0.2) + return reply, retries, retries + + +def _dispatch(step, context, options): + handler = HANDLERS.get(step.type) + if handler is None: + raise ConfigError("no handler for step type %r" % (step.type,), step.source) + return handler(step, context, options) + + +# --------------------------------------------------------------------------- +# per-type handlers + + +def _dns(step, context, options): + param = _resolver(step, context) + return dnsc.query( + server=param("server"), + name=param("name"), + rrtype=param("rrtype") or "A", + port=_int(param("port"), 53), + bind=param("bind") or options.bind, + recursion=_bool(param("recursion"), True), + timeout=_timeout(step, options), + ) + + +def _tftp(step, context, options): + param = _resolver(step, context) + return tftpc.fetch( + server=param("server"), + path=param("path"), + port=_int(param("port"), 69), + mode=param("mode") or "octet", + timeout=_timeout(step, options), + ) + + +def _http(step, context, options): + param = _resolver(step, context) + url = param("url") + if not url: + url = httpc.url_for(param("server"), param("path"), + _int(param("port"), 80)) + headers = [line.strip() for line in (param("header") or "").splitlines() + if line.strip()] + return httpc.fetch( + url=url, + method=param("method") or "GET", + bind=param("bind") or options.bind, + headers=headers, + timeout=_timeout(step, options), + insecure=_bool(param("insecure"), False), + ) + + +def _xcatreq(step, context, options): + param = _resolver(step, context) + return xcatc.request( + server=param("server"), + command=param("command"), + port=_int(param("port"), 3001), + elements=_pairs(param("element")), + bind=param("bind") or options.bind, + timeout=_timeout(step, options), + cert=param("cert") or None, + key=param("key") or None, + callback_port=param("callback_port") or None, + callback_listen=param("callback_listen") or None, + callback_reply=param("callback_reply") or None, + callback_wait=_float(param("callback_wait"), 2.0), + raw=param("raw") or None, + source_port=_int(param("source_port"), 0), + ) + + +def _monitor(step, context, options): + param = _resolver(step, context) + send = [line.strip() for line in (param("send") or "").splitlines() + if line.strip()] + return xcatc.monitor( + server=param("server"), + send=send, + port=_int(param("port"), 3002), + bind=param("bind") or options.bind, + timeout=_timeout(step, options), + source_port=_int(param("source_port"), 0), + ) + + +def _flowrequest(step, context, options): + param = _resolver(step, context) + return xcatc.flowrequest( + server=param("server"), + port=_int(param("port"), 3001), + message=param("message") or "resourcerequest: xcatd", + bind=param("bind") or options.bind, + timeout=_timeout(step, options), + source_port=_int(param("source_port"), 0), + expected=_int(param("replies"), 1), + ) + + +def _findme(step, context, options): + param = _resolver(step, context) + payload = param("payload") + if not payload: + payload = xcatc.discovery_packet() + return xcatc.findme( + server=param("server"), + payload=payload, + port=_int(param("port"), 3001), + encoding=(param("encoding") or "gzip").lower(), + bind=param("bind") or options.bind, + source_port=_int(param("source_port"), 301), + timeout=_timeout(step, options), + callback_listen=_int(param("callback_listen"), 3001), + callback_wait=_float(param("callback_wait"), 5.0), + ) + + +def _extract(step, context, options): + """Pull a value out of an earlier reply, so a later step can fetch it. + + This is the join between stages. A grub2 config names a kernel; nothing in + the config file says what that name will be, and hard-coding it would make + the test assert xCAT's output against a copy of xCAT's output. Extracting + it and then fetching it asserts the only thing that matters: that what the + node was told to fetch can be fetched. + """ + param = _resolver(step, context) + source = param("from") or "" + pattern = step.param("pattern") or "" + try: + regex = re.compile(pattern, re.MULTILINE) + except re.error as exc: + raise ConfigError("bad regex %r: %s" % (pattern, exc), step.source) + + match = regex.search(source) + groups = list(match.groups()) if match else [] + index = _int(param("group"), 1 if groups else 0) + value = "" + if match: + if index > len(groups): + raise ConfigError( + "group=%d, but %r has %d capturing group(s)" + % (index, pattern, len(groups)), step.source) + value = match.group(index) or "" + + fields = { + "value": value, + "groups": [text or "" for text in groups], + "matched": bool(match), + "count": len(groups), + } + return Reply(kind="extract", fields=fields, ok=bool(match), + error="" if match else "%r matched nothing" % (pattern,), + sent="match %r" % (pattern,)) + + +def _sleep(step, context, options): + param = _resolver(step, context) + duration = _float(param("duration"), 1.0) + time.sleep(duration) + return Reply(kind="sleep", fields={}, ok=True, + sent="slept %.1fs" % (duration,)) + + +def _noop(step, context, options): + return Reply(kind="noop", fields={}, ok=True, sent="") + + +HANDLERS = { + "dns": _dns, + "tftp": _tftp, + "http": _http, + "xcatreq": _xcatreq, + "monitor": _monitor, + "flowrequest": _flowrequest, + "findme": _findme, + "extract": _extract, + "sleep": _sleep, + "noop": _noop, +} + +assert set(HANDLERS) == machine.known_step_types() + + +# --------------------------------------------------------------------------- +# parameter handling + + +def _resolver(step, context): + """A `param(key)` that resolves `$step.field` against what has run.""" + + def param(key, default=""): + value = step.param(key, default) + if value in (None, ""): + return value + return subst.resolve(value, context) + + return param + + +def _timeout(step, options): + return options.timeout if options.timeout is not None else step.timeout + + +def _int(value, default): + if value in (None, ""): + return default + try: + return int(str(value).strip(), 0) + except ValueError: + raise ConfigError("expected a number, got %r" % (value,)) + + +def _float(value, default): + if value in (None, ""): + return default + try: + return float(str(value).strip()) + except ValueError: + raise ConfigError("expected a number, got %r" % (value,)) + + +def _bool(value, default): + if value in (None, ""): + return default + text = str(value).strip().lower() + if text in ("1", "yes", "true", "on"): + return True + if text in ("0", "no", "false", "off"): + return False + raise ConfigError("expected a yes/no value, got %r" % (value,)) + + +def _pairs(block): + """`name = value` lines under an `element` key, as an ordered list.""" + pairs = [] + for line in (block or "").splitlines(): + text = line.strip() + if not text or text.startswith("#"): + continue + name, _, value = text.partition("=") + pairs.append((name.strip(), value.strip())) + return pairs diff --git a/xCAT-test/provtest/src/provtest_lib/subst.py b/xCAT-test/provtest/src/provtest_lib/subst.py new file mode 100644 index 000000000..28728229b --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/subst.py @@ -0,0 +1,89 @@ +"""Runtime references to earlier replies: `$step.field`. + +Variables are configparser's job. `%(name)s` is interpolated by +`BasicInterpolation` out of `[vars]` and `--set` before a value ever reaches +this module, and `BasicInterpolation` gives `$` no meaning of its own, so the +two schemes share a file with no escaping. + +What configparser cannot do is `$grubcfg.text`, because that value is a reply +that has not arrived when the file is read. Cross-stage assertions are the +whole point of this suite -- TFTP fetches a config, HTTP fetches what that +config named -- so this form is resolved at step-execution time instead, when +the earlier replies exist. +""" + +import re + +from .errors import ConfigError + +REF_RE = re.compile(r"\$([A-Za-z_][A-Za-z0-9_-]*)\.([A-Za-z_][A-Za-z0-9_.-]*)") + +#: configparser's variable syntax, matched only to report what a file needs. +VAR_RE = re.compile(r"%\(([^)]+)\)s") + + +class Context(object): + """Every reply a `$reference` can resolve against, at one point in a run.""" + + def __init__(self): + self.bindings = {} + self.order = [] + + def bind(self, name, reply): + if name not in self.bindings: + self.order.append(name) + self.bindings[name] = reply + + def get_binding(self, name): + if name not in self.bindings: + raise ConfigError( + "$%s is not available here; no step of that name has run yet" + % (name,)) + return self.bindings[name] + + def __repr__(self): + return "Context(bindings=%r)" % (self.order,) + + +def references(text): + """Every `$step.field` in a value, as `[(step, field), ...]`.""" + if text is None: + return [] + return [(name, field.rstrip(".")) for name, field in REF_RE.findall(str(text))] + + +def variables(text): + """Every `%(name)s` in a value.""" + if text is None: + return set() + return set(VAR_RE.findall(str(text))) + + +def resolve(text, context): + """Substitute every `$step.field` in `text`.""" + if text is None: + return None + + def replace(match): + name, field = match.group(1), match.group(2).rstrip(".") + reply = context.get_binding(name) + try: + return format_value(reply.field(field)) + except KeyError as exc: + raise ConfigError(str(exc)) + + return REF_RE.sub(replace, str(text)) + + +def format_value(value): + """Render a resolved value the way a .conf author would have typed it.""" + if isinstance(value, bool): + return "yes" if value else "no" + if isinstance(value, bytes): + return value.decode("utf-8", "replace") + if isinstance(value, (list, tuple)): + return ", ".join(format_value(item) for item in value) + if isinstance(value, dict): + return ", ".join("%s=%s" % (key, format_value(value[key])) + for key in sorted(value)) + return str(value) diff --git a/xCAT-test/provtest/src/provtest_lib/tftpc.py b/xCAT-test/provtest/src/provtest_lib/tftpc.py new file mode 100644 index 000000000..b94e4b3cc --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/tftpc.py @@ -0,0 +1,105 @@ +"""TFTP, fetched the way a loader fetches: the `tftp` client. + +The names are the thing under test at this stage. A loader asks for +`boot/grub2/grub.cfg-0A63010B` and nothing else; if `nodeset` wrote the hex a +digit out, the transfer fails and the node simply times out at the loader with +no error anywhere on the management node. So the fetch is by exact name, and +the reply carries the size and a digest, which is what lets one scenario assert +that the same bytes come back over TFTP and over HTTP. + +The `tftp` client has no way to choose a source address. It does not need one: +TFTP has no notion of who is asking, so nothing at this stage depends on which +end of the veth pair the request left by. +""" + +import hashlib +import os +import re +import tempfile + +from . import proc +from .model import Reply + +ERROR_RE = re.compile(r"Error code (\d+):\s*(.*)") + +#: The bytes above which a fetched file is summarised rather than carried. +#: A kernel is tens of megabytes and no assertion reads it as text; a config +#: file is a few hundred bytes and every assertion does. +TEXT_LIMIT = 1 << 20 + + +def build(server, path, port=69, mode="octet", local=None): + """The `tftp` command line for one fetch.""" + tftp = proc.require("tftp", "TFTP") + return [tftp, str(server), str(int(port)), "-m", mode, + "-c", "get", str(path), str(local)] + + +def fetch(server, path, port=69, mode="octet", timeout=10.0, keep=None): + """Fetch one file and return a `Reply` describing what arrived.""" + handle, local = tempfile.mkstemp(prefix="provtest-tftp-") + os.close(handle) + try: + argv = build(server, path, port, mode, local) + done = proc.run(argv, timeout=float(timeout) + 5.0) + reply = _decode(done, local, server, path) + if keep and reply.fields.get("ok"): + _copy(local, keep) + return reply + finally: + try: + os.unlink(local) + except OSError: + pass + + +def _decode(done, local, server, path): + size = os.path.getsize(local) if os.path.exists(local) else 0 + body = b"" + if size and size <= TEXT_LIMIT: + with open(local, "rb") as handle: + body = handle.read() + + digest = _digest(local) if size else "" + error = "" + match = ERROR_RE.search(done.text + done.errtext) + if match: + error = "%s (code %s)" % (match.group(2).strip(), match.group(1)) + elif done.timed_out: + error = "the transfer did not finish within the timeout" + elif done.rc != 0: + error = (done.errtext.strip() or done.text.strip() + or "tftp exited %d" % (done.rc,)) + + # tftp-hpa creates the local file before it knows whether the server will + # answer, so a zero-length result with no error is still a failure: no + # file this suite fetches is legitimately empty. + ok = not error and size > 0 + + fields = { + "ok": ok, + "size": size, + "sha256": digest, + "text": body.decode("utf-8", "replace"), + "error": error, + "path": path, + "server": server, + } + reply = Reply(kind="tftp", fields=fields, ok=ok, error=error, + sent=done.command(), raw=body) + return reply + + +def _digest(local): + digest = hashlib.sha256() + with open(local, "rb") as handle: + for chunk in iter(lambda: handle.read(65536), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _copy(local, target): + with open(local, "rb") as source: + with open(target, "wb") as sink: + for chunk in iter(lambda: source.read(65536), b""): + sink.write(chunk) diff --git a/xCAT-test/provtest/src/provtest_lib/xcatc.py b/xCAT-test/provtest/src/provtest_lib/xcatc.py new file mode 100644 index 000000000..68a61ab95 --- /dev/null +++ b/xCAT-test/provtest/src/provtest_lib/xcatc.py @@ -0,0 +1,470 @@ +"""A small xCAT client: the four ways a booting node talks to xcatd. + +A node being provisioned makes exactly four kinds of request, and none of them +needs an xCAT command or an xCAT library to make: + + * **TLS on 3001** -- `getdestiny`, `nextdestiny`, `getpostscript`, + `getcredentials`. The genesis scripts send a few lines of XML down + `openssl s_client` and read a few lines back; this does the same over a + socket, so a request can be sent from a chosen source address. + * **Plain TCP on 3002** -- the install monitor. A line protocol with no TLS + at all, framed by `ready` and `done`. + * **UDP on 3001** -- flow control (`resourcerequest: xcatd`) and `findme`. + * **A connection back to the node** -- xcatd answers a findme by connecting + out to the client's TCP 3001, and signs a certificate only after the + client's TCP 300 agrees. Both are listeners this module can stand up. + +Two things about xcatd make all of this reachable from a script, and both are +why this suite can exist at all. The TLS listener accepts a client with no +certificate, and identity comes from the reverse lookup of the address the +connection arrived from. So the address a request leaves by *is* the identity, +which is why every call here takes a `bind`. +""" + +import gzip +import re +import socket +import ssl +import threading +import time + +from .model import Reply + +#: The elements a node's own scripts read out of a response. Everything else +#: found in the XML is still available through `elements.`. +NAMED_ELEMENTS = ("destiny", "kernel", "initrd", "kcmdline", "imgserver", + "name", "data", "error") + +ELEMENT_RE = re.compile(r"<([A-Za-z_][\w.-]*)>([^<]*)") + +#: What the install monitor prints before it will read anything. +GREETING = "ready" + +#: The default callback port the genesis scripts declare. +CALLBACK_PORT = 300 + + +# --------------------------------------------------------------------------- +# listeners on the client's side + + +class Listener(object): + """A socket on the client end, for the connections xcatd makes outward. + + Started before the request that provokes the callback and stopped after + it, so a scenario can assert both that a callback arrived and -- the more + interesting case -- that it did not. + """ + + def __init__(self, address, port, reply=None, backlog=4): + self.address = address + self.port = int(port) + self.reply = reply + self.received = [] + self.peers = [] + self.error = "" + self._socket = None + self._thread = None + self._stop = threading.Event() + self._backlog = backlog + + def start(self): + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((self.address or "", self.port)) + sock.listen(self._backlog) + sock.settimeout(0.2) + except OSError as exc: + self.error = "cannot listen on %s:%d: %s" % ( + self.address or "*", self.port, exc) + return self + self._socket = sock + self._thread = threading.Thread(target=self._serve) + self._thread.daemon = True + self._thread.start() + return self + + def _serve(self): + while not self._stop.is_set(): + try: + conn, peer = self._socket.accept() + except socket.timeout: + continue + except OSError: + return + try: + conn.settimeout(2.0) + data = b"" + while True: + try: + chunk = conn.recv(4096) + except (socket.timeout, OSError): + break + if not chunk: + break + data += chunk + if self.reply is not None: + break + if self.reply is not None: + try: + conn.sendall(self.reply.encode()) + except OSError: + pass + self.received.append(data.decode("utf-8", "replace").strip()) + self.peers.append(peer[0]) + finally: + try: + conn.close() + except OSError: + pass + + def stop(self): + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=2.0) + if self._socket is not None: + try: + self._socket.close() + except OSError: + pass + return self.received + + def wait(self, seconds): + """Block until something arrives, or the time runs out.""" + deadline = time.time() + float(seconds) + while time.time() < deadline and not self.received: + time.sleep(0.05) + return self.received + + +# --------------------------------------------------------------------------- +# TLS on 3001 + + +def request(server, command, port=3001, elements=(), bind=None, timeout=10.0, + cert=None, key=None, callback_port=None, callback_listen=None, + callback_reply=None, callback_wait=2.0, raw=None, + source_port=None): + """Send one `` over TLS and decode what came back.""" + body = raw if raw is not None else _xml(command, elements, callback_port) + listener = None + if callback_listen: + listener = Listener(bind, int(callback_listen), callback_reply).start() + + # Always present, listener or not: "no callback arrived" is an assertion a + # scenario makes, and a field that is missing rather than false cannot be + # asserted against. + fields = {"callback_seen": False, "callback_data": []} + error = "" + handshake = False + text = "" + try: + sock = _tcp(server, int(port), bind, timeout, source_port) + context = _tls_context(cert, key) + try: + stream = context.wrap_socket(sock, server_hostname=None) + except ssl.SSLError as exc: + sock.close() + raise OSError("TLS handshake failed: %s" % (exc,)) + handshake = True + try: + stream.sendall(body.encode()) + text = _drain(stream, timeout) + finally: + try: + stream.close() + except OSError: + pass + except OSError as exc: + error = str(exc) + + if listener is not None: + listener.wait(callback_wait) + fields["callback_data"] = listener.stop() + fields["callback_seen"] = bool(fields["callback_data"]) + if listener.error and not error: + error = listener.error + + fields.update(decode_elements(text)) + fields["text"] = text + fields["handshake"] = handshake + fields["ok"] = handshake and bool(text) and not fields.get("error") + return Reply(kind="xcatreq", fields=fields, ok=fields["ok"], error=error, + sent=body.strip().replace("\n", " "), + raw=text.encode("utf-8", "replace")) + + +def decode_elements(text): + """Every `value` in a response, flattened. + + Flattened deliberately: the genesis scripts read the response with `sed`, + one element at a time and with no regard for nesting, so a test that + parsed it as a tree could pass where a node fails. + """ + elements = {} + for name, value in ELEMENT_RE.findall(text): + elements.setdefault(name, value.strip()) + fields = {"elements": elements, "serverdone": "", "%s" % (command,)] + if callback_port: + lines.append("%s" % (callback_port,)) + for name, value in elements: + lines.append("<%s>%s" % (name, value, name)) + lines.append("") + return "\n".join(lines) + "\n" + + +def _tls_context(cert=None, key=None): + """A client context that verifies nothing. + + That is not a shortcut: it is what the genesis scripts do. `openssl + s_client` with no `-CAfile` accepts whatever certificate it is shown, and + a node has no CA to check against until xcatd has signed one for it. + """ + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + if cert: + context.load_cert_chain(cert, key or None) + return context + + +def _drain(stream, timeout): + """Read until the server is done, or the timeout, whichever is first.""" + stream.settimeout(timeout) + deadline = time.time() + float(timeout) + text = "" + while time.time() < deadline: + try: + chunk = stream.recv(8192) + except (socket.timeout, ssl.SSLError): + break + except OSError: + break + if not chunk: + break + text += chunk.decode("utf-8", "replace") + if "", "findme"] + for name, value in elements: + lines.append("<%s>%s" % (name, value, name)) + lines.append("") + return "\n".join(lines) + "\n" + + +# --------------------------------------------------------------------------- +# sockets + + +def _tcp(server, port, bind, timeout, source_port=None): + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(timeout) + if bind or source_port: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((bind or "", int(source_port or 0))) + sock.connect((server, int(port))) + return sock + + +def _udp(bind, source_port=None): + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + if bind or source_port: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((bind or "", int(source_port or 0))) + return sock