|
1 | 1 | import ydb |
2 | 2 |
|
| 3 | +def load_ydb_ca_cert(path:str) -> str: |
| 4 | + """Load CA certification. |
| 5 | +
|
| 6 | + Args: |
| 7 | + path (str): path to CA certification. |
| 8 | + |
| 9 | + The function load_ydb_ca_cert accepts a path to a CA certificate |
| 10 | + and returns its content as a byte string for further passing to ydb.DriverConfig. |
| 11 | + If the specified path is incorrect or the certificate does not exist, |
| 12 | + it raises an exception with the message "CA not found". |
| 13 | +
|
| 14 | + """ |
| 15 | + if path is not None and os.path.exists(path): |
| 16 | + with open(path, "rb") as file: |
| 17 | + return file.read() |
| 18 | + else: |
| 19 | + raise FileNotFoundError("CA not found") |
3 | 20 |
|
4 | 21 | def test_driver_works(driver: ydb.Driver): |
| 22 | + """Tests the functionality of the YDB driver. |
| 23 | +
|
| 24 | + Waits for the driver to become ready and executes a simple SQL query to verify that the driver works as expected. |
| 25 | +
|
| 26 | + Args: |
| 27 | + driver (ydb.Driver): The YDB driver instance to test. |
| 28 | +
|
| 29 | + Raises: |
| 30 | + AssertionError: If the SQL query does not return the expected result. |
| 31 | + """ |
5 | 32 | driver.wait(5) |
6 | 33 | pool = ydb.QuerySessionPool(driver) |
7 | 34 | result = pool.execute_with_retries("SELECT 1 as cnt") |
8 | 35 | assert result[0].rows[0].cnt == 1 |
9 | 36 |
|
10 | | - |
11 | 37 | def auth_with_static_credentials(endpoint: str, database: str, user: str, password: str): |
| 38 | + """Authenticate using static credentials. |
| 39 | +
|
| 40 | + Args: |
| 41 | + endpoint (str): Accepts a string in the format `grpcs://<node-fqdn>:2136` or `grpcs://<node-ip>:2136`. |
| 42 | + database (str): Accepts a string, the database name in the format `/Root/<database-name>`. |
| 43 | + user (str): Username. |
| 44 | + password (str): User password. |
| 45 | +
|
| 46 | + Notes: |
| 47 | + The argument `root_certificates` of the function `ydb.DriverConfig` takes the content of the cluster's root certificate for connecting to cluster nodes via TLS. |
| 48 | + Note that the VM from which you are connecting must be in the cluster's domain for which the CA certificate is issued. |
| 49 | + """ |
| 50 | + |
12 | 51 | driver_config = ydb.DriverConfig( |
13 | | - endpoint=endpoint, |
14 | | - database=database, |
15 | | - credentials=ydb.StaticCredentials.from_user_password(user, password), |
| 52 | + endpoint = endpoint, |
| 53 | + database = database, |
| 54 | + credentials = ydb.StaticCredentials.from_user_password(user, password), |
| 55 | + root_certificates = load_ydb_ca_cert(path = <path_to_ca-cert.crt>) |
16 | 56 | ) |
17 | 57 |
|
18 | 58 | with ydb.Driver(driver_config=driver_config) as driver: |
|
0 commit comments