KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use set_env_var method in Robotframework

Best Python code snippet using robotframework

test_e2e_bq.py

Source:test_e2e_bq.py

...46    @staticmethod47    def get_env_var(key_name):48        return os.environ.get(key_name, 'Key name is not set')49    @staticmethod50    def set_env_var(key, value):51        """52        Sets environment variable.53        :param key:54        :param value:55        """56        os.environ[key] = value57    @staticmethod58    def get_output_results(test_case_no):59        """60        Calls the execute_transformation_query function from cud_correction_dag.py in source directory.61        :param test_case_no: It takes test case number so that it can be called for different test cases.62        :return: results of the "select" query63        """64        project_id = TestBQ.get_env_var('TEST_PROJECT_ID')65        TestBQ.set_env_var('project_id', project_id)66        billing_export_dataset_id = TestBQ.get_env_var('TEST_DATASET')67        TestBQ.set_env_var('billing_export_dataset_id',68                           billing_export_dataset_id)69        commitment_table_name = TestBQ.get_env_var(70            'TABLE_PREFIX') + test_case_no + "_commitments"71        TestBQ.set_env_var('commitment_table_name', commitment_table_name)72        billing_export_table_name = TestBQ.get_env_var(73            'TABLE_PREFIX') + test_case_no + "_export"74        TestBQ.set_env_var('billing_export_table_name',75                           billing_export_table_name)76        distribute_commitments_query = TestBQ.get_env_var(77            'DISTRIBUTE_COMMITMENTS_QUERY')78        TestBQ.set_env_var('distribute_commitments_query',79                           distribute_commitments_query)80        corrected_billing_query = TestBQ.get_env_var('CORRECTED_BILLING_QUERY')81        TestBQ.set_env_var('corrected_billing_query', corrected_billing_query)82        distribute_commitment_table = TestBQ.get_env_var(83            'DISTRIBUTE_COMMITMENT_TABLE')84        distribute_commitment_table = distribute_commitment_table + test_case_no85        TestBQ.set_env_var('distribute_commitment_table',86                           distribute_commitment_table)87        corrected_dataset_id = TestBQ.get_env_var('OUTPUT_TEST_DATASET')88        TestBQ.set_env_var('corrected_dataset_id', corrected_dataset_id)89        output_table_name = TestBQ.get_env_var('OUTPUT_TABLE_PREFIX')90        output_table_name = output_table_name + test_case_no91        TestBQ.set_env_var('corrected_table_name', output_table_name)92        main("", "")93        results = TestBQ.select_from_table(corrected_dataset_id,94                                           output_table_name)95        return results96    @staticmethod97    def select_from_table(dataset_id, table_name):98        """99        Method to run a "select" query from a table and returns results.100        :param dataset_id:101        :param table_name:102        :return:103        """104        client = bigquery.Client()105        select_sql = "select * from " + dataset_id + '.' + table_name...

test_config.py

Source:test_config.py

...5from panel import config, state6from panel.pane import HTML7from panel.tests.conftest import set_env_var8def test_env_var_console_output():9    with set_env_var('PANEL_CONSOLE_OUTPUT', 'disable'):10        assert config.console_output == 'disable'11    with set_env_var('PANEL_CONSOLE_OUTPUT', 'replace'):12        assert config.console_output == 'replace'13    with config.set(console_output='disable'):14        with set_env_var('PANEL_DOC_BUILD', 'accumulate'):15            assert config.console_output == 'disable'16def test_config_set_console_output():17    with config.set(console_output=False):18        assert config.console_output == 'disable'19    with config.set(console_output='disable'):20        assert config.console_output == 'disable'21    with config.set(console_output='replace'):22        assert config.console_output == 'replace'23    with config.set(console_output='disable'):24        with config.set(console_output='accumulate'):25            assert config.console_output == 'accumulate'26@pytest.mark.usefixtures("with_curdoc")27def test_session_override():28    config.sizing_mode = 'stretch_width'29    assert config.sizing_mode == 'stretch_width'30    assert state.curdoc in config._session_config31    assert config._session_config[state.curdoc] == {'sizing_mode': 'stretch_width'}32    state.curdoc = None33    assert config.sizing_mode is None34def test_console_output_replace_stdout(document, comm, get_display_handle):35    pane = HTML()36    with set_env_var('PANEL_CONSOLE_OUTPUT', 'replace'):37        model = pane.get_root(document, comm)38        handle = get_display_handle(model)39        pane._on_stdout(model.ref['id'], ['print output'])40        assert handle == {'text/html': 'print output</br>', 'raw': True}41        pane._on_stdout(model.ref['id'], ['new output'])42        assert handle == {'text/html': 'new output</br>', 'raw': True}43        pane._cleanup(model)44        assert model.ref['id'] not in state._handles45def test_console_output_accumulate_stdout(document, comm, get_display_handle):46    pane = HTML()47    model = pane.get_root(document, comm)48    handle = get_display_handle(model)49    pane._on_stdout(model.ref['id'], ['print output'])50    assert handle == {'text/html': 'print output</br>', 'raw': True}51    pane._on_stdout(model.ref['id'], ['new output'])52    assert handle == {'text/html': 'print output</br>\nnew output</br>', 'raw': True}53    pane._cleanup(model)54    assert model.ref['id'] not in state._handles55def test_console_output_disable_stdout(document, comm, get_display_handle):56    pane = HTML()57    with set_env_var('PANEL_CONSOLE_OUTPUT', 'disable'):58        model = pane.get_root(document, comm)59        handle = get_display_handle(model)60        pane._on_stdout(model.ref['id'], ['print output'])61        assert handle == {}62        pane._cleanup(model)63        assert model.ref['id'] not in state._handles64def test_console_output_replace_error(document, comm, get_display_handle):65    pane = HTML()66    with set_env_var('PANEL_CONSOLE_OUTPUT', 'replace'):67        model = pane.get_root(document, comm)68        handle = get_display_handle(model)69        try:70            1/071        except Exception as e:72            pane._on_error(model.ref['id'], e)73        assert 'text/html' in handle74        assert 'ZeroDivisionError' in handle['text/html']75        try:76            1 + '2'77        except Exception as e:78            pane._on_error(model.ref['id'], e)79        assert 'text/html' in handle80        assert 'ZeroDivisionError' not in handle['text/html']81        assert 'TypeError' in handle['text/html']82        pane._cleanup(model)83        assert model.ref['id'] not in state._handles84def test_console_output_accumulate_error(document, comm, get_display_handle):85    pane = HTML()86    model = pane.get_root(document, comm)87    handle = get_display_handle(model)88    try:89        1/090    except Exception as e:91        pane._on_error(model.ref['id'], e)92    assert 'text/html' in handle93    assert 'ZeroDivisionError' in handle['text/html']94    try:95        1 + '2'96    except Exception as e:97        pane._on_error(model.ref['id'], e)98    assert 'text/html' in handle99    assert 'ZeroDivisionError' in handle['text/html']100    assert 'TypeError' in handle['text/html']101    pane._cleanup(model)102    assert model.ref['id'] not in state._handles103def test_console_output_disable_error(document, comm, get_display_handle):104    pane = HTML()105    with set_env_var('PANEL_CONSOLE_OUTPUT', 'disable'):106        model = pane.get_root(document, comm)107        handle = get_display_handle(model)108        try:109            1/0110        except Exception as e:111            pane._on_error(model.ref['id'], e)112        assert handle == {}113        pane._cleanup(model)...

upload_to_pinata.py

Source:upload_to_pinata.py

...39    """40        Pin any file, or directory, to Pinata's IPFS nodes41        More: https://docs.pinata.cloud/api-pinning/pin-file42    """43    set_env_var()44    headers = {45        "pinata_api_key": os.getenv("PINATA_API_KEY"),46        "pinata_secret_api_key": os.getenv("PINATA_API_SECRET"),47    }48    current_working_dir = os.getcwd()4950    def get_all_files(dir: str):51        """get a list of absolute paths and relative paths to every file located in the directory"""52        abspaths, relativepaths, filenames = [], [], []53        dirname = _filepath.split("/")[-1:][0]54        for root, dirs, files_ in os.walk(os.path.abspath(dir)):55            for file in files_:56                abspaths.append(os.path.join(root, file))57                relativepaths.append(dirname + "/" + file)58                filenames.append(file)59        return abspaths, relativepaths, filenames6061    if os.path.isdir(_filepath):62        all_files_abs, all_files_rel, filenames = get_all_files(_filepath)63        os.chdir(Path(_filepath).parent)64        print("STARTING UPLOADING ALL FILES IN FOLDER . . .")65        files = [("file", (all_files_rel[i], open(all_files_abs[i], "rb")))66                 for i in range(len(all_files_abs))]67        response = requests.post(68            PINATA_BASE_URL + endpoint,69            files=files,70            headers=headers,71        )72    else:73        filename = _filepath.split("/")[-1:][0]74        files = {"file": (filename, open(_filepath, 'rb'))}7576    response = requests.post(77        PINATA_BASE_URL + endpoint,78        files=files,79        headers=headers,80    )81    os.chdir(current_working_dir)82    return(pinata_image_base_url.format(response.json()["IpfsHash"]))838485def unpin_all():86    set_env_var()87    headers = {88        "pinata_api_key": os.getenv("PINATA_API_KEY"),89        "pinata_secret_api_key": os.getenv("PINATA_API_SECRET"),90    }91    pinList_endpoint = "data/pinList?status=pinned&pageLimit=1000"92    unpin_endpoint = "pinning/unpin/{}"9394    # First, get the list of All pinned files95    print(f"Sending request to {PINATA_BASE_URL + pinList_endpoint}")96    response = requests.get(97        PINATA_BASE_URL + pinList_endpoint,98        headers=headers,99    )100    response = response.json()101    number_pinned = response["count"]102    list_hash_pinned = []103    print(f"START UNPINNING {number_pinned} FILES. . .")104    row_number = len(response["rows"])105    print(f"number of rows: {row_number}")106    for i in range(row_number):107        hash = response["rows"][i]["ipfs_pin_hash"]108        list_hash_pinned.append(hash)109        link_to_pinned_file = PINATA_BASE_URL + unpin_endpoint.format(hash)110        print(f"row {i} multi-hash: {hash}")111        unpin_response = requests.delete(112            link_to_pinned_file,113            headers=headers114        )115        print(f"File {i}: {unpin_response}")116117118def pin_dir_to_pinata(dirPath):119    set_env_var()120    headers = {121        "pinata_api_key": os.getenv("PINATA_API_KEY"),122        "pinata_secret_api_key": os.getenv("PINATA_API_SECRET"),123    }124125    dirname = dirPath.split("/")[-1:][0]126    os.chdir(Path(dirPath).parent)127    print(os.getcwd())128129    onlyfiles = [join(dirname, f)130                 for f in listdir(dirname) if isfile(join(dirname, f))]131    print(onlyfiles)132    data = []133    for file in onlyfiles:
...

environment.py

Source:environment.py

1import os2def set_env_var(name, value, overwrite=False):3    """4    Set an environment variable to a value.5    `Args:`6        name: str7            Name of the env var8        value: str9            New value for the env var10        overwrite: bool11            Whether to set the env var even if it already exists12    """13    # Do nothing if we don"t have a value14    if not value:15        return16    # Do nothing if env var already exists17    if os.environ.get(name) and not overwrite:18        return19    os.environ[name] = value20def setup_env_for_parsons(redshift_parameter="REDSHIFT",21                          aws_parameter="AWS",22                          creds=[]):23    """24    Sets up environment variables commonly used in parsons scripts.25    Call this at the beginning of your script.26    `Args:`27        redshift_parameter: str28            Name of the Civis script parameter holding Redshift credentials.29            This parameter should be of type "database (2 dropdown)" in Civis.30        aws_parameter: str31            Name of the Civis script parameter holding AWS credentials.32        creds: list33            Arbitrary arguments for which to remove the '_PASSWORD' suffix.34    """35    env = os.environ36    # Redshift setup37    set_env_var("REDSHIFT_PORT", "5432")38    set_env_var("REDSHIFT_DB", "dev")39    set_env_var("REDSHIFT_ID", env.get(f"{redshift_parameter}_ID"))40    set_env_var("REDSHIFT_NAME", env.get(f"{redshift_parameter}_NAME"))41    set_env_var("REDSHIFT_HOST", env.get(f"{redshift_parameter}_HOST"))42    set_env_var("REDSHIFT_USERNAME",43                env.get(f"{redshift_parameter}_CREDENTIAL_USERNAME"))44    set_env_var("REDSHIFT_PASSWORD",45                env.get(f"{redshift_parameter}_CREDENTIAL_PASSWORD"))46    # AWS setup47    set_env_var("S3_TEMP_BUCKET", "tmp-bucket")48    set_env_var("AWS_ACCESS_KEY_ID", env.get(f"{aws_parameter}_USERNAME"))49    set_env_var("AWS_SECRET_ACCESS_KEY", env.get(f"{aws_parameter}_PASSWORD"))50    for cred in creds:...

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.