Krikri::PythonTestRunner

WRAPPER = "import importlib.util\nimport json\nimport os\nimport sys\n\n\n_loaded = []\n\n\ndef _load_plugin(path):\n modname = \"krikri_test_%d\" % len(_loaded)\n spec = importlib.util.spec_from_file_location(modname, path)\n if spec is None or spec.loader is None:\n raise ImportError(\"cannot load test plugin: %s\" % path)\n module = importlib.util.module_from_spec(spec)\n spec.loader.exec_module(module)\n _loaded.append(module)\n return module\n\n\ndef _is_pass_context(func):\n # Same detection as the filter wrapper (jinja2 >= 3.0's\n # `jinja_pass_arg`, older `contextfilter`): a @pass_context\n # test gets a Context injected as its FIRST positional\n # argument by Jinja's own calling convention.\n arg = getattr(func, \"jinja_pass_arg\", None)\n if arg is not None and getattr(arg, \"name\", None) == \"context\":\n return True\n return bool(getattr(func, \"contextfilter\", False))\n\n\nclass _StubContext(object):\n # Same minimal stand-in as the filter wrapper's: covers the\n # variable-lookup patterns custom tests actually use over the\n # play vars snapshot krikri passes in.\n def __init__(self, variables):\n self.vars = variables\n self.parent = variables\n self.environment = None\n\n def resolve(self, name, default=None):\n return self.vars.get(name, default)\n\n def get(self, key, default=None):\n return self.vars.get(key, default)\n\n def get_all(self):\n return self.vars\n\n def __getitem__(self, key):\n return self.vars[key]\n\n def __contains__(self, key):\n return key in self.vars\n\n\ndef _plugin_tests(path):\n module = _load_plugin(path)\n plugin_dir = os.path.dirname(os.path.abspath(path))\n if plugin_dir not in sys.path:\n sys.path.insert(0, plugin_dir)\n plugin_class = getattr(module, \"TestModule\", None)\n if plugin_class is None:\n return {}\n # Ansible's own PluginLoader instantiates the plugin class\n # before calling tests() on the instance - a tests() defined\n # as a plain instance method (the common spelling) fails with\n # \"missing self\" when called on the class directly.\n tests = plugin_class().tests()\n if not isinstance(tests, dict):\n return {}\n return tests\n\n\ndef _load_all(paths):\n # Ansible's PluginLoader warns about a plugin file that fails\n # to import and keeps loading the rest, so one broken file\n # must not take down its siblings' tests.\n names = {}\n skipped = []\n for path in paths:\n try:\n names.update(_plugin_tests(path))\n except Exception as exc:\n skipped.append({\"path\": path, \"error\": str(exc)})\n return names, skipped\n\n\ndef _serialize(value):\n if isinstance(value, (set, frozenset)):\n return list(value)\n if isinstance(value, bytes):\n return value.decode(\"utf-8\", \"replace\")\n return str(value)\n\n\ndef main():\n payload = json.loads(sys.stdin.read())\n names, skipped = _load_all(payload.get(\"paths\", []))\n if payload.get(\"action\") == \"list\":\n sys.stdout.write(json.dumps(\n {\"ok\": True, \"names\": sorted(names), \"skipped\": skipped}))\n return\n name = payload.get(\"name\")\n func = names.get(name)\n if func is None:\n sys.stdout.write(json.dumps(\n {\"ok\": False, \"kind\": \"not_found\",\n \"error\": \"no test named %s in the given plugin files\" % name}))\n return\n if _is_pass_context(func):\n call_args = [_StubContext(payload.get(\"vars\") or {})]\n else:\n call_args = []\n call_args.append(payload.get(\"value\"))\n call_args.extend(payload.get(\"args\") or [])\n result = func(*call_args, **(payload.get(\"kwargs\") or {}))\n sys.stdout.write(json.dumps({\"ok\": True, \"result\": result}, default=_serialize))\n\n\ntry:\n main()\nexcept Exception as exc:\n sys.stdout.write(json.dumps(\n {\"ok\": False, \"kind\": \"error\",\n \"error\": \"%s: %s\" % (type(exc).__name__, exc)}))"

warning_sink=(warning_sink : IO)

call_test(name : String, sources : Array(String), value : JSON::Any, args : Array(JSON::Any), kwargs : Hash(String, JSON::Any), vars : Hash(String, JSON::Any) | Nil = nil) : JSON::Any

defines_test?(name : String, sources : Array(String)) : Bool

find_sources(role_path : String | Nil, playbook_dir : String | Nil) : Array(String)

test_names(sources : Array(String)) : Set(String)