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

How to use _get_credentials method in tempest

Best Python code snippet using tempest_python

test_gbq.py

Source:test_gbq.py

...37    private_key_path = PRIVATE_KEY_JSON_PATH38    if not private_key_path:39        private_key_path = os.environ.get("GBQ_GOOGLE_APPLICATION_CREDENTIALS")40    return private_key_path41def _get_credentials():42    private_key_path = _get_private_key_path()43    if private_key_path:44        return service_account.Credentials.from_service_account_file(private_key_path)45def _get_client():46    project_id = _get_project_id()47    credentials = _get_credentials()48    return bigquery.Client(project=project_id, credentials=credentials)49def generate_rand_str(length: int = 10) -> str:50    return "".join(random.choices(string.ascii_lowercase, k=length))51def make_mixed_dataframe_v2(test_size):52    # create df to test for all BQ datatypes except RECORD53    bools = np.random.randint(2, size=(1, test_size)).astype(bool)54    flts = np.random.randn(1, test_size)55    ints = np.random.randint(1, 10, size=(1, test_size))56    strs = np.random.randint(1, 10, size=(1, test_size)).astype(str)57    times = [datetime.now(pytz.timezone("US/Arizona")) for t in range(test_size)]58    return DataFrame(59        {60            "bools": bools[0],61            "flts": flts[0],62            "ints": ints[0],63            "strs": strs[0],64            "times": times[0],65        },66        index=range(test_size),67    )68def test_read_gbq_without_deprecated_kwargs(monkeypatch):69    captured_kwargs = {}70    def mock_read_gbq(sql, **kwargs):71        captured_kwargs.update(kwargs)72        return DataFrame([[1.0]])73    monkeypatch.setattr("pandas_gbq.read_gbq", mock_read_gbq)74    pd.read_gbq("SELECT 1")75    assert "verbose" not in captured_kwargs76    assert "private_key" not in captured_kwargs77def test_read_gbq_with_new_kwargs(monkeypatch):78    captured_kwargs = {}79    def mock_read_gbq(sql, **kwargs):80        captured_kwargs.update(kwargs)81        return DataFrame([[1.0]])82    monkeypatch.setattr("pandas_gbq.read_gbq", mock_read_gbq)83    pd.read_gbq("SELECT 1", use_bqstorage_api=True)84    assert captured_kwargs["use_bqstorage_api"]85def test_read_gbq_without_new_kwargs(monkeypatch):86    captured_kwargs = {}87    def mock_read_gbq(sql, **kwargs):88        captured_kwargs.update(kwargs)89        return DataFrame([[1.0]])90    monkeypatch.setattr("pandas_gbq.read_gbq", mock_read_gbq)91    pd.read_gbq("SELECT 1")92    assert "use_bqstorage_api" not in captured_kwargs93@pytest.mark.parametrize("progress_bar", [None, "foo"])94def test_read_gbq_progress_bar_type_kwarg(monkeypatch, progress_bar):95    # GH 2985796    captured_kwargs = {}97    def mock_read_gbq(sql, **kwargs):98        captured_kwargs.update(kwargs)99        return DataFrame([[1.0]])100    monkeypatch.setattr("pandas_gbq.read_gbq", mock_read_gbq)101    pd.read_gbq("SELECT 1", progress_bar_type=progress_bar)102    if progress_bar:103        assert "progress_bar_type" in captured_kwargs104    else:105        assert "progress_bar_type" not in captured_kwargs106@pytest.mark.single107class TestToGBQIntegrationWithServiceAccountKeyPath:108    @pytest.fixture()109    def gbq_dataset(self):110        # Setup Dataset111        _skip_if_no_project_id()112        _skip_if_no_private_key_path()113        dataset_id = "pydata_pandas_bq_testing_" + generate_rand_str()114        self.client = _get_client()115        self.dataset = self.client.dataset(dataset_id)116        # Create the dataset117        self.client.create_dataset(bigquery.Dataset(self.dataset))118        table_name = generate_rand_str()119        destination_table = f"{dataset_id}.{table_name}"120        yield destination_table121        # Teardown Dataset122        self.client.delete_dataset(self.dataset, delete_contents=True)123    def test_roundtrip(self, gbq_dataset):124        destination_table = gbq_dataset125        test_size = 20001126        df = make_mixed_dataframe_v2(test_size)127        df.to_gbq(128            destination_table,129            _get_project_id(),130            chunksize=None,131            credentials=_get_credentials(),132        )133        result = pd.read_gbq(134            f"SELECT COUNT(*) AS num_rows FROM {destination_table}",135            project_id=_get_project_id(),136            credentials=_get_credentials(),137            dialect="standard",138        )139        assert result["num_rows"][0] == test_size140    @pytest.mark.parametrize(141        "if_exists, expected_num_rows, expectation",142        [143            ("append", 300, does_not_raise()),144            ("fail", 200, pytest.raises(pandas_gbq.gbq.TableCreationError)),145            ("replace", 100, does_not_raise()),146        ],147    )148    def test_gbq_if_exists(149        self, if_exists, expected_num_rows, expectation, gbq_dataset150    ):151        # GH 29598152        destination_table = gbq_dataset153        test_size = 200154        df = make_mixed_dataframe_v2(test_size)155        df.to_gbq(156            destination_table,157            _get_project_id(),158            chunksize=None,159            credentials=_get_credentials(),160        )161        with expectation:162            df.iloc[:100].to_gbq(163                destination_table,164                _get_project_id(),165                if_exists=if_exists,166                chunksize=None,167                credentials=_get_credentials(),168            )169        result = pd.read_gbq(170            f"SELECT COUNT(*) AS num_rows FROM {destination_table}",171            project_id=_get_project_id(),172            credentials=_get_credentials(),173            dialect="standard",174        )...

cli.py

Source:cli.py

...17        value = "Set from env"18    if not value.startswith("http"):19        raise ClickException(f"Store URL must start with: 'http'. You provided: {value}")20    return value21def _get_credentials(ctx, param, creds_file=None):22    if creds_file and not os.path.isfile(creds_file):23        raise ClickException(f"Credentials file does not exist: '{creds_file}'")24    return creds_file25@click.group()26def main():27    """Console script for jos."""28    click.echo("You are running the 'jos' command-line.")29    return 030@main.command()31@click.argument("bucket_id")32@click.option("-s", "--store-url", callback=_get_store_url)33@click.option("-c", "--creds-file", callback=_get_credentials)34def create_bucket(bucket_id, store_url, creds_file):35    #params = ctx.params...

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.