selftests/functional/nrunner.py
import os
import sys
import time
import unittest
from avocado.core.job import Job
from avocado.utils import process
from selftests.utils import AVOCADO, BASEDIR, TestCaseTmpDir, skipUnlessPathExists
RUNNER = "avocado-runner-noop"
class NRunnerFeatures(TestCaseTmpDir):
@skipUnlessPathExists("/bin/false")
def test_custom_exit_codes(self):
config = {
"run.results_dir": self.tmpdir.name,
"resolver.references": ["/bin/false"],
"runner.exectest.exitcodes.skip": [1],
}
with Job.from_config(job_config=config) as job:
self.assertEqual(job.run(), 0)
@skipUnlessPathExists("/bin/false")
@skipUnlessPathExists("/bin/true")
def test_failfast(self):
config = {
"run.results_dir": self.tmpdir.name,
"resolver.references": [
"/bin/true",
"/bin/false",
"/bin/true",
"/bin/true",
],
"run.failfast": True,
"run.shuffle": False,
"run.max_parallel_tasks": 1,
}
with Job.from_config(job_config=config) as job:
self.assertEqual(job.run(), 9)
self.assertEqual(job.result.passed, 1)
self.assertEqual(job.result.errors, 0)
self.assertEqual(job.result.failed, 1)
self.assertEqual(job.result.skipped, 2)
class RunnableRun(unittest.TestCase):
def test_noop(self):
res = process.run(f"{RUNNER} runnable-run -k noop", ignore_status=True)
self.assertIn(b"'status': 'started'", res.stdout)
self.assertIn(b"'status': 'finished'", res.stdout)
self.assertIn(b"'time': ", res.stdout)
self.assertEqual(res.exit_status, 0)
def test_instrumented(self):
test_uri = os.path.join(
BASEDIR, "examples", "tests", "passtest.py:PassTest.test"
)
res = process.run(
f"{RUNNER} runnable-run -k avocado-instrumented -u " f"{test_uri}",
ignore_status=True,
)
self.assertIn(b"'status': 'started'", res.stdout)
self.assertIn(b"'status': 'finished'", res.stdout)
self.assertIn(b"'result': 'pass'", res.stdout)
self.assertIn(b"'time': ", res.stdout)
self.assertEqual(res.exit_status, 0)
def test_exec_test(self):
# 'base64:LWM=' becomes '-c' and makes Python execute the
# commands on the subsequent argument
cmd = (
f"{RUNNER} runnable-run -k exec-test -u {sys.executable} "
f"-a 'base64:LWM=' -a 'import sys; sys.exit(99)'"
)
res = process.run(cmd, ignore_status=True)
self.assertIn(b"'status': 'finished'", res.stdout)
self.assertIn(b"'returncode': 99", res.stdout)
self.assertEqual(res.exit_status, 0)
@skipUnlessPathExists("/bin/sh")
def test_exec_test_echo(self):
# 'base64:LW4=' becomes '-n' and prevents echo from printing a newline
cmd = (
f"{RUNNER} runnable-run -k exec-test -u /bin/echo "
f"-a 'base64:LW4=' -a _Avocado_Runner_"
)
res = process.run(cmd, ignore_status=True)
self.assertIn(b"'status': 'finished'", res.stdout)
self.assertIn(b"'log': b'_Avocado_Runner_'", res.stdout)
self.assertIn(b"'returncode': 0", res.stdout)
self.assertEqual(res.exit_status, 0)
@skipUnlessPathExists("/bin/sh")
@skipUnlessPathExists("/bin/echo")
def test_exec_recipe(self):
recipe = os.path.join(
BASEDIR,
"examples",
"nrunner",
"recipes",
"runnables",
"exec_test_sh_echo_env_var.json",
)
cmd = f"{RUNNER} runnable-run-recipe {recipe}"
res = process.run(cmd, ignore_status=True)
lines = res.stdout_text.splitlines()
if len(lines) == 1:
first_status = final_status = lines[0]
else:
first_status = lines[0]
stdout_status = lines[-3]
final_status = lines[-1]
self.assertIn("'status': 'started'", first_status)
self.assertIn("'time': ", first_status)
self.assertIn("'status': 'finished'", final_status)
self.assertIn("'log': b'Hello world!\\n'", stdout_status)
self.assertIn("'time': ", final_status)
self.assertEqual(res.exit_status, 0)
def test_noop_valid_kwargs(self):
res = process.run(f"{RUNNER} runnable-run -k noop foo=bar", ignore_status=True)
self.assertEqual(res.exit_status, 0)
def test_noop_invalid_kwargs(self):
res = process.run(f"{RUNNER} runnable-run -k noop foo", ignore_status=True)
self.assertIn(b'Invalid keyword parameter: "foo"', res.stderr)
self.assertEqual(res.exit_status, 2)
@skipUnlessPathExists("/bin/env")
def test_exec_test_kwargs(self):
cmd = f"{RUNNER} runnable-run -k exec-test -u /bin/env X=Y"
res = process.run(cmd, ignore_status=True)
self.assertIn(b"'status': 'finished'", res.stdout)
self.assertIn(b"X=Y\\n", res.stdout)
self.assertEqual(res.exit_status, 0)
class ExecTestStdOutErr(unittest.TestCase):
def test_64kib(self):
path = os.path.join(
BASEDIR, "selftests", ".data", "exec_test_std", "exec_test_64kib.py"
)
res = process.run(
f"avocado-runner-exec-test runnable-run -u {path}", ignore_status=True
)
self.assertIn(b"'type': 'stdout'", res.stdout)
self.assertIn(b"'type': 'stderr'", res.stdout)
self.assertIn(b"'result': 'pass'", res.stdout)
self.assertIn(b"'returncode': 0", res.stdout)
self.assertEqual(res.exit_status, 0)
class ExecTestStdOutErrOutputDir(TestCaseTmpDir):
def test_1mib(self):
path = os.path.join(
BASEDIR, "selftests", ".data", "exec_test_std", "exec_test_1mib.py"
)
res = process.run(
f"avocado-runner-exec-test runnable-run -u {path} output_dir={self.tmpdir.name}",
ignore_status=True,
)
self.assertIn(b"'type': 'stdout'", res.stdout)
self.assertIn(b"'type': 'stderr'", res.stdout)
self.assertIn(b"'result': 'pass'", res.stdout)
self.assertIn(b"'returncode': 0", res.stdout)
self.assertEqual(res.exit_status, 0)
def test_error_existing_stdout_stderr(self):
self.test_1mib()
path = os.path.join(
BASEDIR, "selftests", ".data", "exec_test_std", "exec_test_1mib.py"
)
res = process.run(
f"avocado-runner-exec-test runnable-run -u {path} output_dir={self.tmpdir.name}",
ignore_status=True,
)
self.assertIn(b"'result': 'error'", res.stdout)
self.assertEqual(res.exit_status, 0)
class TaskRun(unittest.TestCase):
def test_noop(self):
res = process.run(
f"{RUNNER} task-run -i XXXno-opXXX -k noop", ignore_status=True
)
self.assertIn(b"'status': 'finished'", res.stdout)
self.assertIn(b"'id': 'XXXno-opXXX'", res.stdout)
self.assertEqual(res.exit_status, 0)
@skipUnlessPathExists("/bin/uname")
def test_recipe_exec_test_1(self):
recipe = os.path.join(
BASEDIR,
"examples",
"nrunner",
"recipes",
"tasks",
"exec-test",
"1-uname.json",
)
cmd = f"{RUNNER} task-run-recipe {recipe}"
res = process.run(cmd, ignore_status=True)
lines = res.stdout_text.splitlines()
if len(lines) == 1:
first_status = final_status = lines[0]
else:
first_status = lines[0]
final_status = lines[-1]
self.assertIn("'status': 'started'", first_status)
self.assertIn("'id': 1", first_status)
self.assertIn("'id': 1", first_status)
self.assertIn("'status': 'finished'", final_status)
self.assertEqual(res.exit_status, 0)
@skipUnlessPathExists("/bin/echo")
def test_recipe_exec_test_2(self):
recipe = os.path.join(
BASEDIR,
"examples",
"nrunner",
"recipes",
"tasks",
"exec-test",
"2-echo.json",
)
cmd = f"{RUNNER} task-run-recipe {recipe}"
res = process.run(cmd, ignore_status=True)
lines = res.stdout_text.splitlines()
if len(lines) == 1:
first_status = final_status = lines[0]
else:
first_status = lines[0]
stdout_status = lines[-3]
final_status = lines[-1]
self.assertIn("'status': 'started'", first_status)
self.assertIn("'id': 2", first_status)
self.assertIn("'id': 2", first_status)
self.assertIn("'status': 'finished'", final_status)
self.assertIn("'log': b'avocado'", stdout_status)
self.assertEqual(res.exit_status, 0)
@skipUnlessPathExists("/bin/sleep")
def test_recipe_exec_test_3(self):
recipe = os.path.join(
BASEDIR,
"examples",
"nrunner",
"recipes",
"tasks",
"exec-test",
"3-sleep.json",
)
cmd = f"{RUNNER} task-run-recipe {recipe}"
res = process.run(cmd, ignore_status=True)
lines = res.stdout_text.splitlines()
# based on the :data:`avocado.core.nrunner.RUNNER_RUN_STATUS_INTERVAL`
# this runnable should produce multiple status lines
self.assertGreater(len(lines), 1)
first_status = lines[0]
self.assertIn("'status': 'started'", first_status)
self.assertIn("'id': 3", first_status)
final_status = lines[-1]
self.assertIn("'id': 3", first_status)
self.assertIn("'status': 'finished'", final_status)
self.assertEqual(res.exit_status, 0)
class TaskRunStatusService(TestCaseTmpDir):
@skipUnlessPathExists("/bin/sleep")
@skipUnlessPathExists("/bin/nc")
def test_task_status_service_lost(self):
nc_path = os.path.join(self.tmpdir.name, "socket")
nc_proc = process.SubProcess(f"nc -lU {nc_path}")
nc_proc.start()
task_proc = process.SubProcess(
f"avocado-runner-exec-test task-run -i 1 -u /bin/sleep -a 3 -s {nc_path}"
)
task_proc.start()
time.sleep(1)
nc_proc.kill()
time.sleep(1)
self.assertIn(
f"Connection with {nc_path} has been lost.".encode(), task_proc.get_stderr()
)
class ResolveSerializeRun(TestCaseTmpDir):
@skipUnlessPathExists("/bin/true")
def test(self):
cmd = "%s list --write-recipes-to-directory=%s -- /bin/true"
cmd %= (AVOCADO, self.tmpdir.name)
res = process.run(cmd)
self.assertEqual(b"exec-test /bin/true\n", res.stdout)
cmd = "%s runnable-run-recipe %s"
cmd %= (RUNNER, os.path.join(self.tmpdir.name, "1.json"))
res = process.run(cmd)
self.assertIn(b"'status': 'finished'", res.stdout)