How to use test_exit_code method in gabbi

Best Python code snippet using gabbi_python

test_pipelines.py

Source:test_pipelines.py

...42        assert os.path.exists("pipeline")43run = MyRun.make_fixture("class", "{sys.executable} pipeline")44@pytest.mark.usefixtures("run")45class TestMyPipeline(unittest.TestCase):46    def test_exit_code(self):47        assert self.run_fixture.exit_code == 048"""49def test_pipeline_basic(mockpipe, testdir):50    """Test for basic run"""51    test = testdir.makepyfile(TEST_OK)52    result = testdir.inline_run(53        "-v",54        f"--base-pipeline-dir={test.dirname}",55        test56    )57    passed, skipped, failed = result.listoutcomes()58    assert len(passed) == 159    assert len(skipped) == 060    assert len(failed) == 061TEST_OK_CLASS_FIXTURE = f"""62import os, shutil, unittest63import pytest64from pytest_pipeline import PipelineRun, mark65class MyRun(PipelineRun):66    @mark.before_run67    def prep_executable(self):68        shutil.copy2("../pipeline", "pipeline")69        assert os.path.exists("pipeline")70run = MyRun.class_fixture("{sys.executable} pipeline")71@pytest.mark.usefixtures("run")72class TestMyPipelineAgain(unittest.TestCase):73    def test_exit_code(self):74        assert self.run_fixture.exit_code == 075"""76def test_pipeline_class_fixture(mockpipe, testdir):77    """Test for basic run"""78    test = testdir.makepyfile(TEST_OK_CLASS_FIXTURE)79    result = testdir.inline_run(80        "-v",81        f"--base-pipeline-dir={test.dirname}",82        test83    )84    passed, skipped, failed = result.listoutcomes()85    assert len(passed) == 186    assert len(skipped) == 087    assert len(failed) == 088TEST_REDIRECTION = f"""89import os, shutil, unittest90import pytest91from pytest_pipeline import PipelineRun, mark92class MyRun(PipelineRun):93    @mark.before_run94    def prep_executable(self):95        shutil.copy2("../pipeline", "pipeline")96        assert os.path.exists("pipeline")97run = MyRun.make_fixture(98    "class",99    cmd="{sys.executable} pipeline",100    stdout="stream.out",101    stderr="stream.err",102)103@pytest.mark.usefixtures("run")104class TestMyPipeline(unittest.TestCase):105    def test_exit_code(self):106        assert self.run_fixture.exit_code == 0107"""108def test_pipeline_redirection(mockpipe, testdir):109    test = testdir.makepyfile(TEST_REDIRECTION)110    result = testdir.inline_run(111        "-v",112        f"--base-pipeline-dir={test.dirname}",113        test114    )115    passed, skipped, failed = result.listoutcomes()116    assert len(passed) == 1117    assert len(skipped) == 0118    assert len(failed) == 0119    testdir_matches = glob.glob(os.path.join(test.dirname, "MyRun*"))120    assert len(testdir_matches) == 1121    testdir_pipeline = testdir_matches[0]122    stdout = os.path.join(testdir_pipeline, "stream.out")123    assert os.path.exists(stdout)124    assert open(stdout).read() == "stdout stream"125    stderr = os.path.join(testdir_pipeline, "stream.err")126    assert os.path.exists(stderr)127    assert open(stderr).read() == "stderr stream"128TEST_REDIRECTION_MEM = f"""129import os, shutil, unittest130import pytest131from pytest_pipeline import PipelineRun, mark132class MyRun(PipelineRun):133    @mark.before_run134    def prep_executable(self):135        shutil.copy2("../pipeline", "pipeline")136        assert os.path.exists("pipeline")137run = MyRun.make_fixture(138    "class",139    cmd="{sys.executable} pipeline",140    stdout=True,141    stderr=True,142)143@pytest.mark.usefixtures("run")144class TestMyPipeline(unittest.TestCase):145    def test_exit_code(self):146        assert self.run_fixture.exit_code == 0147    def test_stdout(self):148        assert self.run_fixture.stdout == b"stdout stream"149    def test_stderr(self):150        assert self.run_fixture.stderr == b"stderr stream"151"""152def test_pipeline_redirection_mem(mockpipe, testdir):153    test = testdir.makepyfile(TEST_REDIRECTION_MEM)154    result = testdir.inline_run(155        "-v",156        f"--base-pipeline-dir={test.dirname}",157        test158    )159    passed, skipped, failed = result.listoutcomes()160    assert len(passed) == 3161    assert len(skipped) == 0162    assert len(failed) == 0163    testdir_matches = glob.glob(os.path.join(test.dirname, "MyRun*"))164    assert len(testdir_matches) == 1165TEST_AS_NONCLASS_FIXTURE = f"""166import os, shutil, unittest167import pytest168from pytest_pipeline import PipelineRun, mark169class MyRun(PipelineRun):170    @mark.before_run171    def prep_executable(self):172        shutil.copy2("../pipeline", "pipeline")173        assert os.path.exists("pipeline")174run = MyRun.make_fixture("module", "{sys.executable} pipeline")175def test_exit_code(run):176    assert run.exit_code == 0177"""178def test_pipeline_as_nonclass_fixture(mockpipe, testdir):179    """Test for PipelineTest classes without run attribute"""180    test = testdir.makepyfile(TEST_AS_NONCLASS_FIXTURE)181    result = testdir.inline_run(182        "-v",183        f"--base-pipeline-dir={test.dirname}",184        test185    )186    passed, skipped, failed = result.listoutcomes()187    assert len(passed) == 1188    assert len(skipped) == 0189    assert len(failed) == 0190TEST_OK_GRANULAR = f"""191import os, shutil, unittest192import pytest193from pytest_pipeline import PipelineRun, mark194class MyRun(PipelineRun):195    @mark.before_run(order=2)196    def prep_executable(self):197        shutil.copy2("../pipeline", "pipeline")198        assert os.path.exists("pipeline")199    @mark.before_run(order=1)200    def check_init_condition(self):201        assert not os.path.exists("pipeline")202run = MyRun.make_fixture("class", cmd="{sys.executable} pipeline")203@pytest.mark.usefixtures("run")204class TestMyPipeline(unittest.TestCase):205    def test_exit_code(self):206        assert self.run_fixture.exit_code == 0207    def test_output_file(self):208        assert os.path.exists(os.path.join("output_dir", "results.txt"))209"""210def test_pipeline_granular(mockpipe, testdir):211    """Test for execution with 'order' specified in before_run and after_run"""212    test = testdir.makepyfile(TEST_OK_GRANULAR)213    result = testdir.inline_run(214        "-v",215        f"--base-pipeline-dir={test.dirname}",216        test217    )218    passed, skipped, failed = result.listoutcomes()219    assert len(passed) == 2220    assert len(skipped) == 0221    assert len(failed) == 0222MOCK_PIPELINE_TIMEOUT = """223#!/usr/bin/env python224if __name__ == "__main__":225    import time226    time.sleep(10)227"""228TEST_TIMEOUT = f"""229import os, shutil, unittest230import pytest231from pytest_pipeline import PipelineRun, mark232class MyRun(PipelineRun):233    @mark.before_run234    def test_and_prep_executable(self):235        shutil.copy2("../pipeline", "pipeline")236        assert os.path.exists("pipeline")237run = PipelineRun.make_fixture(238    "class",239    cmd="{sys.executable} pipeline",240    timeout=0.01,241)242@pytest.mark.usefixtures("run")243class TestMyPipeline(unittest.TestCase):244    def test_exit_code(self):245        assert self.run_fixture.exit_code != 0246"""247@pytest.fixture(scope="function")248def mockpipe_timeout(request, testdir):249    """Mock pipeline script with timeout"""250    mp = testdir.makefile("", pipeline=MOCK_PIPELINE_TIMEOUT)251    return mp252def test_pipeline_timeout(mockpipe_timeout, testdir):253    """Test for execution with timeout"""254    test = testdir.makepyfile(TEST_TIMEOUT)255    result = testdir.inline_run(256        "-v",257        f"--base-pipeline-dir={test.dirname}",258        test259    )260    passed, skipped, failed = result.listoutcomes()261    assert len(passed) == 0262    assert len(skipped) == 0263    assert len(failed) == 1264MOCK_PIPELINE_FMT = """265#!/usr/bin/env python266import sys267if __name__ == "__main__":268    print(sys.argv[1])269"""270TEST_FMT = f"""271import os, shutil, unittest272import pytest273from pytest_pipeline import PipelineRun, mark274class MyRun(PipelineRun):275    @mark.before_run276    def prep_executable(self):277        shutil.copy2("../pipeline", "pipeline")278        assert os.path.exists("pipeline")279run = MyRun.make_fixture(280    "class",281    "{sys.executable} pipeline {{run_dir}}",282    stdout=True,283)284@pytest.mark.usefixtures("run")285class TestMyPipeline(unittest.TestCase):286    def test_exit_code(self):287        assert self.run_fixture.exit_code == 0288    def test_stdout(self):289        stdout = self.run_fixture.stdout.decode("utf-8").strip()290        assert self.run_fixture.run_dir == stdout291"""292@pytest.fixture(scope="function")293def mockpipe_fmt(request, testdir):294    """Mock pipeline script with timeout"""295    mp = testdir.makefile("", pipeline=MOCK_PIPELINE_FMT)296    return mp297def test_pipeline_fmt(mockpipe_fmt, testdir):298    """Test for run with templated command"""299    test = testdir.makepyfile(TEST_FMT)300    result = testdir.inline_run(...

test_runners.py

Source:test_runners.py

1import threading2import time3from dataclasses import dataclass4from datetime import timedelta5from django.utils import timezone6from lct.languages.executors import CompilationError7from lct.languages.models import get_language_by_name8from .models import TestStatuses9@dataclass10class TestResult:11    data_in: str12    data_out: str13    exec_time: timedelta = 014    status: TestStatuses = TestStatuses.OK15    @property16    def passed(self):17        return self.status == TestStatuses.OK18class TestError(Exception):19    def __init__(self, data_in, data_out):20        self.data_in = data_in21        self.data_out = data_out22class IncorrectResult(TestError):23    pass24class RuntimeTestError(TestError):25    pass26class TimeoutTestError(TestError):27    pass28class BadChecker(TestError):29    pass30class BaseTestRun(threading.Thread):31    def __init__(self, *args, test, executor, **kwargs):32        super().__init__(*args, **kwargs)33        self.exc = None34        self.result = None35        self.test = test36        self.executor = executor37        self.is_finished = False38    def perform_test(self, test):39        raise NotImplementedError40    def run(self):41        ex = self.executor.exec()42        try:43            self.result = self.perform_test(ex)44        except TestError as exc:45            self.exc = exc46        finally:47            ex.stop()48    def stop(self):49        self.is_finished = True50class InOutTestRun(BaseTestRun):51    def perform_test(self, ex):52        ex.write(self.test.data_in)53        reads = []54        while ex.get_exit_code() is None:55            if self.is_finished:56                raise TimeoutTestError(data_in="", data_out="")57            reads.append(ex.read())58        exit_code = ex.get_exit_code()59        reads.append(ex.read())60        read = "".join(reads)61        if exit_code == 0:62            normalized_expected = self.test.data_out.rstrip().replace("\r", "")63            normalized_output = read.rstrip().replace("\r", "")64            if normalized_output == normalized_expected:65                result = TestResult(66                    data_in=self.test.data_in, data_out=self.test.data_out67                )68            else:69                raise IncorrectResult(self.test.data_in, read)70        else:71            raise RuntimeTestError(self.test.data_in, read)72        return result73class BaseTestRunner:74    test_run_class = None75    def __init__(self, tests, code, executor_class, timeout=5, memory=128):76        self.executor_class = executor_class77        self.code = code78        self.tests = tests79        self.executor = None80        self.memory = memory81        self.timeout = timeout82    def test(self, test):83        run = self.test_run_class(test=test, executor=self.executor)84        start_dt = timezone.now()85        run.start()86        run.join(timeout=self.timeout)87        if run.is_alive():88            run.stop()89            run.join()90        end_dt = timezone.now()91        exec_time = (end_dt - start_dt).total_seconds() * 100092        if run.result is not None:93            run.result.exec_time = exec_time94            return run.result95        if isinstance(run.exc, TimeoutTestError):96            return TestResult(97                data_in=run.exc.data_in,98                data_out=run.exc.data_out,99                status=TestStatuses.TO,100                exec_time=exec_time,101            )102        elif isinstance(run.exc, IncorrectResult):103            return TestResult(104                data_in=run.exc.data_in,105                data_out=run.exc.data_out,106                status=TestStatuses.WA,107                exec_time=exec_time,108            )109        elif isinstance(run.exc, RuntimeTestError):110            return TestResult(111                data_in=run.exc.data_in,112                data_out=run.exc.data_out,113                status=TestStatuses.RE,114                exec_time=exec_time,115            )116    def test_all(self):117        results = []118        try:119            self.executor = self.executor_class(self.code, memory=self.memory)120        except CompilationError:121            for _ in self.tests:122                results.append(123                    TestResult(data_in="", data_out="", status=TestStatuses.CE)124                )125            return results126        for test in self.tests:127            results.append(self.test(test))128        self.executor.stop()129        return results130class InOutTestRunner(BaseTestRunner):131    test_run_class = InOutTestRun132class CheckerTestRun(BaseTestRun):133    def perform_test(self, ex):134        lang = get_language_by_name(self.test.lang)135        executor = lang.executor_class(self.test.code)136        test_ex = executor.exec()137        outs = []138        test_outs = []139        while ex.get_exit_code() is None and test_ex.get_exit_code() is None:140            if self.is_finished:141                raise TimeoutTestError(data_in="", data_out="")142            test_out = test_ex.read()143            if test_out:144                test_outs.append(test_out)145                ex.write(test_out)146            out = ex.read()147            if out:148                outs.append(out)149                test_ex.write(out)150        time.sleep(0.2)151        out = ex.read()152        if out:153            outs.append(out)154            if test_ex.get_exit_code() is None:155                test_ex.write(out)156                time.sleep(0.3)157        outs.append(ex.read())158        test_outs.append(test_ex.read())159        out = "".join(outs)160        test_out = "".join(test_outs)161        exit_code = ex.get_exit_code()162        test_exit_code = ex.get_exit_code()163        test_ex.stop()164        executor.stop()165        if exit_code != 0 and exit_code is not None:166            raise RuntimeTestError(data_in=test_out, data_out=out)167        if test_exit_code == 0:168            if "ok" in test_out:169                return TestResult(data_in=test_out, data_out=out)170            elif "error" in test_out:171                raise IncorrectResult(data_in=test_out, data_out=out)172            else:173                raise BadChecker(data_in=test_out, data_out=out)174        if test_exit_code != 0 and test_exit_code is not None:175            raise BadChecker(data_in=test_out, data_out=out)176        if test_exit_code is None:177            if "ok" in test_out:178                return TestResult(data_in=test_out, data_out=out)179            elif "error" in test_out:180                raise IncorrectResult(data_in=test_out, data_out=out)181            else:182                raise IncorrectResult(data_in=test_out, data_out=out)183        if exit_code == 0 and test_exit_code == 0:184            if "ok" in test_out:185                return TestResult(data_in=test_out, data_out=out)186            elif "error" in test_out:187                raise IncorrectResult(data_in=test_out, data_out=out)188            else:189                raise BadChecker(data_in=test_out, data_out=out)190        elif exit_code != 0:191            raise RuntimeTestError(data_in=test_out, data_out=out)192        else:193            raise BadChecker(data_in=test_out, data_out=out)194class CheckerTest:195    def __init__(self, code, executor_class):196        self.code = code197        self.executor_class = executor_class198    def get_executor(self):199        return self.executor_class(self.code)200class CheckerTestRunner(BaseTestRunner):...

test_exit_code.py

Source:test_exit_code.py

1r"""2Tests python exit code in various circumstances3Script Commands:4    python ~/misc/tests/python/test_exit_code.py --exit-mode=sys  --exit-val=0 ; \5    echo $? ; \6    python ~/misc/tests/python/test_exit_code.py --exit-mode=sys  --exit-val=1 ; \7    echo $? ; \8    python ~/misc/tests/python/test_exit_code.py --exit-mode=sys  --exit-val=None ; \9    echo $? ; \10    python ~/misc/tests/python/test_exit_code.py --exit-mode=sys  --exit-val=err ; \11    echo $?12    python ~/misc/tests/python/test_exit_code.py --exit-mode=None --exit-val=0 ; \13    echo $? ; \14    python ~/misc/tests/python/test_exit_code.py --exit-mode=None --exit-val=1 ; \15    echo $? ; \16    python ~/misc/tests/python/test_exit_code.py --exit-mode=None --exit-val=None ; \17    echo $? ; \18    python ~/misc/tests/python/test_exit_code.py --exit-mode=None --exit-val=err ; \19    echo $?20Results:21    (py38) joncrall@Ooo:~/misc/tests/python$     python ~/misc/tests/python/test_exit_code.py --exit-mode=sys  --exit-val=0 ; \22    >     echo $? ; \23    >     python ~/misc/tests/python/test_exit_code.py --exit-mode=sys  --exit-val=1 ; \24    >     echo $? ; \25    >     python ~/misc/tests/python/test_exit_code.py --exit-mode=sys  --exit-val=None ; \26    >     echo $? ; \27    >     python ~/misc/tests/python/test_exit_code.py --exit-mode=sys  --exit-val=err ; \28    >     echo $?29    030    131    032    Traceback (most recent call last):33      File "/home/joncrall/misc/tests/python/test_exit_code.py", line 46, in <module>34        sys.exit(main())35      File "/home/joncrall/misc/tests/python/test_exit_code.py", line 36, in main36        raise Exception37    Exception38    139    (py38) joncrall@Ooo:~/misc/tests/python$40    (py38) joncrall@Ooo:~/misc/tests/python$     python ~/misc/tests/python/test_exit_code.py --exit-mode=None --exit-val=0 ; \41    >     echo $? ; \42    >     python ~/misc/tests/python/test_exit_code.py --exit-mode=None --exit-val=1 ; \43    >     echo $? ; \44    >     python ~/misc/tests/python/test_exit_code.py --exit-mode=None --exit-val=None ; \45    >     echo $? ; \46    >     python ~/misc/tests/python/test_exit_code.py --exit-mode=None --exit-val=err ; \47    >     echo $?48    049    050    051"""52import sys53import ubelt as ub54def main():55    val = ub.argval('--exit-val', default=None)56    if val == 'None':57        val = None58    if val == 'err':59        raise Exception60    if val is not None:61        val = int(val)62    return val63if __name__ == '__main__':64    mode = ub.argval('--exit-mode', default='sys')65    if mode == 'sys':66        sys.exit(main())67    else:...

test-interactive.py

Source:test-interactive.py

1#!/usr/bin/env python32import argparse3import pexpect4import time5import sys6import io7import os8import shlex9import importlib.util10script_dir = os.path.dirname(os.path.realpath(__file__))11def import_test_func(path, func_name='test_interactive'):12    module_name = os.path.basename(path)13    spec = importlib.util.spec_from_file_location(module_name, path)14    foo = importlib.util.module_from_spec(spec)15    spec.loader.exec_module(foo)16    return getattr(foo, func_name)17def get_earthly_binary(earthly):18    if os.path.isfile(earthly):19        return earthly20    if os.path.sep not in earthly:21        for path in os.environ['PATH'].split(':'):22            fullpath = os.path.join(path, earthly)23            if os.path.isfile(fullpath):24                return fullpath25    raise RuntimeError(f'failed to find earthly binary: {earthly}')26if __name__ == '__main__':27    parser = argparse.ArgumentParser()28    parser.add_argument('-e', '--earthly', help="earthly binary to run test against", default='earthly')29    parser.add_argument('-t', '--timeout', help="fail test if it takes longer than this many seconds", type=float, default=30.0)30    args = parser.parse_args()31    earthly_path = os.path.realpath(get_earthly_binary(args.earthly))32    print(f'Running interactive tests against "{earthly_path}"')33    exit_code = 034    for test_name, test in (35            ('test-simple', import_test_func(os.path.join(script_dir, 'simple', 'test-simple.py'))),36            ('test-interactive-run', import_test_func(os.path.join(script_dir, 'interactive-run', 'test-interactive-run.py'))),37            ('test-docker-compose', import_test_func(os.path.join(script_dir, 'docker-compose', 'test-docker-compose.py'))),38            ):39        print(f'Running {test_name}')40        test_exit_code = test(earthly_path, args.timeout)41        if test_exit_code == 2:42            print(f'{test_name} timedout')43            exit_code = test_exit_code44        elif test_exit_code:45            print(f'{test_name} failed with exit code={test_exit_code}')46            exit_code = test_exit_code47        else:48            print(f'{test_name} passed')...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.