How to use running_on_ci method in Pytest

Best Python code snippet using pytest

test_ws.py

Source:test_ws.py

...15    assert any("Ports" in i for i in message)16    assert any("Network" in i for i in message)17# NOTE run the following tests with a board connected to the PC18@pytest.mark.skipif(19    running_on_ci(),20    reason="VMs have no serial ports",21)22def test_open_serial_default(socketio, serial_port, baudrate, message):23    general_open_serial(socketio, serial_port, baudrate, message, "default")24@pytest.mark.skipif(25    running_on_ci(),26    reason="VMs have no serial ports",27)28def test_open_serial_timed(socketio, serial_port, baudrate, message):29    general_open_serial(socketio, serial_port, baudrate, message, "timed")30@pytest.mark.skipif(31    running_on_ci(),32    reason="VMs have no serial ports",33)34def test_open_serial_timedraw(socketio, serial_port, baudrate, message):35    general_open_serial(socketio, serial_port, baudrate, message, "timedraw")36# NOTE run the following tests with a board connected to the PC and with the sketch found in tests/testdata/SerialEcho.ino on it be sure to change serial_address in conftest.py37@pytest.mark.skipif(38    running_on_ci(),39    reason="VMs have no serial ports",40)41def test_send_serial_default(socketio, close_port, serial_port, baudrate, message):42    general_send_serial(socketio, close_port, serial_port, baudrate, message, "default")43@pytest.mark.skipif(44    running_on_ci(),45    reason="VMs have no serial ports",46)47def test_send_serial_timed(socketio, close_port, serial_port, baudrate, message):48    general_send_serial(socketio, close_port, serial_port, baudrate, message, "timed")49@pytest.mark.skipif(50    running_on_ci(),51    reason="VMs have no serial ports",52)53def test_send_serial_timedraw(socketio, close_port, serial_port, baudrate, message):54    general_send_serial(socketio, close_port, serial_port, baudrate, message, "timedraw")55@pytest.mark.skipif(56    running_on_ci(),57    reason="VMs have no serial ports",58)59def test_send_emoji_serial_default(socketio, close_port, serial_port, baudrate, message):60    general_send_emoji_serial(socketio, close_port, serial_port, baudrate, message, "default")61@pytest.mark.skipif(62    running_on_ci(),63    reason="VMs have no serial ports",64)65def test_send_emoji_serial_timed(socketio, close_port, serial_port, baudrate, message):66    general_send_emoji_serial(socketio, close_port, serial_port, baudrate, message, "timed")67@pytest.mark.skipif(68    running_on_ci(),69    reason="VMs have no serial ports",70)71def test_send_emoji_serial_timedraw(socketio, close_port, serial_port, baudrate, message):72    general_send_emoji_serial(socketio, close_port, serial_port, baudrate, message, "timedraw")73def general_open_serial(socketio, serial_port, baudrate, message, buffertype):74    open_serial_port(socketio, serial_port, baudrate, message, buffertype)75    # test the closing of the serial port, we are gonna use close_port for the other tests76    socketio.emit('command', 'close ' + serial_port)77    time.sleep(.2)78    print (message)79    #check if port has been closed80    assert any("\"IsOpen\": false," in i for i in message)81def general_send_serial(socketio, close_port, serial_port, baudrate, message, buffertype):82    open_serial_port(socketio, serial_port, baudrate, message, buffertype)83    # send the string "ciao" using the serial connection84    socketio.emit('command', 'send ' + serial_port + ' ciao')85    time.sleep(1)86    print(message)87    # check if the send command has been registered88    assert any("send " + serial_port + " ciao" in i for i in message)89    #check if message has been sent back by the connected board90    if buffertype == "timedraw":91        output =  decode_output(extract_serial_data(message))92    elif buffertype in ("default", "timed"):93        output = extract_serial_data(message)94    assert "ciao" in output95    # the serial connection is closed by close_port() fixture: even if in case of test failure96def general_send_emoji_serial(socketio, close_port, serial_port, baudrate, message, buffertype):97    open_serial_port(socketio, serial_port, baudrate, message, buffertype)98    # send a lot of emoji: they can be messed up99    socketio.emit('command', 'send ' + serial_port + ' /"ð§ð§ð§ð§ð§ð§ð§ð§ð§ð§/"')100    time.sleep(1)101    print(message)102    # check if the send command has been registered103    assert any("send " + serial_port + " /\"ð§ð§ð§ð§ð§ð§ð§ð§ð§ð§/\"" in i for i in message)104    if buffertype == "timedraw":105        output =  decode_output(extract_serial_data(message))106    elif buffertype in ("default", "timed"):107        output = extract_serial_data(message)108    assert "/\"ð§ð§ð§ð§ð§ð§ð§ð§ð§ð§/\"" in output109    # the serial connection is closed by close_port() fixture: even if in case of test failure110def open_serial_port(socketio, serial_port, baudrate, message, buffertype):111    #open a new serial connection with the specified buffertype112    socketio.emit('command', 'open ' + serial_port + ' ' + baudrate + ' ' + buffertype)113    # give time to the message var to be filled114    time.sleep(.5)115    print(message)116    # the serial connection should be open now117    assert any("\"IsOpen\": true" in i for i in message)118@pytest.mark.skipif(119    running_on_ci(),120    reason="VMs have no serial ports",121)122def test_sendraw_serial(socketio, close_port, serial_port, baudrate, message):123    open_serial_port(socketio, serial_port, baudrate, message, "timedraw")124    #test with bytes125    integers = [1, 2, 3, 4, 5]126    bytes_array=bytearray(integers)127    encoded_integers = base64.b64encode(bytes_array).decode('ascii')128    socketio.emit('command', 'sendraw ' + serial_port + ' ' + encoded_integers)129    time.sleep(1)130    print(message)131    # check if the send command has been registered132    assert any(("sendraw " + serial_port + ' ' + encoded_integers) in i for i in message)133    #check if message has been sent back by the connected board...

conftest.py

Source:conftest.py

...16@pytest.fixture(scope="session")17def test_resources():18    return TEST_RESOURCES19@pytest.fixture(scope="session")20def running_on_ci() -> bool:21    return os.getenv("GITLAB_CI") is not None or os.getenv("CI") is not None22@pytest.fixture(scope="session")23def docker_services(24    docker_compose_file, docker_compose_project_name, docker_cleanup, running_on_ci25):26    """This overwrites pytest-docker's docker_services fixture to avoid starting containers on CI"""27    if running_on_ci:28        yield29    else:30        with get_docker_services(31            docker_compose_file, docker_compose_project_name, docker_cleanup32        ) as docker_service:33            yield docker_service34@pytest.fixture(scope="session")...

test_auth_via_cli.py

Source:test_auth_via_cli.py

1import subprocess2import requests3import os4import time5from foundations_spec import *6from foundations_contrib.utils import wait_for_condition7class TestAuthViaClient(Spec):8    max_time_out_in_sec = 609    ci_keycloak_host = "keycloak-headless.ci-pipeline.svc.cluster.local"10    running_on_ci = os.getenv("RUNNING_ON_CI")11    auth_server_host = ci_keycloak_host if running_on_ci else "localhost"12    @staticmethod13    def resolve_f9s_auth():14        import os.path as path15        return path.realpath("../foundations_authentication/src/")16    def keycloak_is_available(self) -> bool:17        try:18            requests.get(19                f"http://{self.auth_server_host}:8080/auth/"20            ).raise_for_status()21            return True22        except requests.ConnectionError:23            return False24        except requests.HTTPError as err:25            self.fail(err)26    def rest_api_is_available(self) -> bool:27        try:28            requests.get(29                f"http://localhost:37722/api/v2beta/projects"30            ).raise_for_status()31            return True32        except requests.ConnectionError:33            return False34        except requests.HTTPError as err:35            log_file = f"{os.getenv('FOUNDATIONS_HOME', '~/.foundations')}/logs/atlas_rest_api.log"36            with open(log_file) as logs:37                msg = "\n".join([str(err), "REST_API_LOGS:", logs.read()])38            self.fail(msg)39    def start_and_wait_for_keycloak(self) -> None:40        full_path = os.path.join(41            self.resolve_f9s_auth(), "foundations_authentication"42        )43        subprocess.run(["bash", "launch.sh"], cwd=full_path, stdout=subprocess.PIPE)44        def keycloak_is_ready() -> bool:45            try:46                res = requests.get(f"http://{self.auth_server_host}:8080/auth/")47            except requests.exceptions.ConnectionError:48                return False49            if res.status_code == 200:50                return True51            return False52        wait_for_condition(53            keycloak_is_ready,54            60,55            fail_hook=lambda: self.fail("keycloak failed to start"),56        )57    def start_and_wait_for_rest_api(self) -> None:58        import subprocess59        subprocess.run(60            "export REDIS_HOST=localhost && export FOUNDATIONS_SCHEDULER_URL=localhost && cd ../devops && python startup_atlas_api.py 37722 &",61            shell=True,62        )63        def rest_api_is_ready() -> bool:64            try:65                res = requests.get("http://localhost:37722/api/v2beta/projects")66            except requests.exceptions.ConnectionError:67                return False68            if res.status_code == 200:69                return True70            else:71                return False72        wait_for_condition(73            rest_api_is_ready,74            5,75            fail_hook=lambda: self.fail("Atlas REST API failed to start"),76        )77    def test_cli_login(self):78        if not self.keycloak_is_available():79            if self.running_on_ci:80                self.fail("Keycloak is unavailable in our cluster.")81            self.start_and_wait_for_keycloak()82        if not self.rest_api_is_available():83            if self.running_on_ci:84                self.fail("Atlas REST API is unavailable in our cluster.")85            self.start_and_wait_for_rest_api()86        with self.assert_does_not_raise():87            result = subprocess.run(88                "python -m foundations login http://localhost:5558 -u test -p test",89                stdout=subprocess.PIPE,90                shell=True,91                check=True,92            )...

test_board.py

Source:test_board.py

...11# a commercial license, send an email to license@arduino.cc.12import pytest13import simplejson as json14from .common import running_on_ci15@pytest.mark.skipif(running_on_ci(), reason="VMs have no serial ports")16def test_core_list(run_command):17    result = run_command("core update-index")18    assert result.ok19    result = run_command("board list --format json")20    assert result.ok21    # check is a valid json and contains a list of ports22    ports = json.loads(result.stdout).get("ports")23    assert isinstance(ports, list)24    for port in ports:25        assert "protocol" in port26        assert "protocol_label" in port27@pytest.mark.skipif(running_on_ci(), reason="VMs have no serial ports")28def test_core_listall(run_command):29    assert run_command("core update-index")30    result = run_command("board listall")31    print(result.stderr, result.stdout)32    assert result.ok33    assert ["Board", "Name", "FQBN"] == result.stdout.splitlines()[0].strip().split()34def test_core_search(run_command):35    url = "https://raw.githubusercontent.com/arduino/arduino-cli/master/test/testdata/test_index.json"36    assert run_command("core update-index --additional-urls={}".format(url))37    # default search38    result = run_command("core search avr")39    assert result.ok40    assert 2 < len(result.stdout.splitlines())41    result = run_command("core search avr --format json")...

Pytest Tutorial

Looking for an in-depth tutorial around pytest? LambdaTest covers the detailed pytest tutorial that has everything related to the pytest, from setting up the pytest framework to automation testing. Delve deeper into pytest testing by exploring advanced use cases like parallel testing, pytest fixtures, parameterization, executing multiple test cases from a single file, and more.

Chapters

What is pytest
Pytest installation: Want to start pytest from scratch? See how to install and configure pytest for Python automation testing.
Run first test with pytest framework: Follow this step-by-step tutorial to write and run your first pytest script.
Parallel testing with pytest: A hands-on guide to parallel testing with pytest to improve the scalability of your test automation.
Generate pytest reports: Reports make it easier to understand the results of pytest-based test runs. Learn how to generate pytest reports.
Pytest Parameterized tests: Create and run your pytest scripts while avoiding code duplication and increasing test coverage with parameterization.
Pytest Fixtures: Check out how to implement pytest fixtures for your end-to-end testing needs.
Execute Multiple Test Cases: Explore different scenarios for running multiple test cases in pytest from a single file.
Stop Test Suite after N Test Failures: See how to stop your test suite after n test failures in pytest using the @pytest.mark.incremental decorator and maxfail command-line option.