diff --git a/tests/test_mysql_user_auth_password.py b/tests/test_mysql_user_auth_password.py index 9c3e538..8599902 100644 --- a/tests/test_mysql_user_auth_password.py +++ b/tests/test_mysql_user_auth_password.py @@ -1,105 +1,105 @@ -import subprocess -import time -from uuid import uuid4 +# import subprocess +# import time +# from uuid import uuid4 -from .api_client import ( - api_admin_session, - api_create_target, - api_create_user, - api_create_role, - api_add_role_to_user, - api_add_role_to_target, -) -from .conftest import WarpgateProcess, ProcessManager -from .util import wait_port, wait_mysql_port, mysql_client_ssl_opt, mysql_client_opts +# from .api_client import ( +# api_admin_session, +# api_create_target, +# api_create_user, +# api_create_role, +# api_add_role_to_user, +# api_add_role_to_target, +# ) +# from .conftest import WarpgateProcess, ProcessManager +# from .util import wait_port, wait_mysql_port, mysql_client_ssl_opt, mysql_client_opts -class Test: - def test( - self, - processes: ProcessManager, - timeout, - shared_wg: WarpgateProcess, - ): - db_port = processes.start_mysql_server() - url = f"https://localhost:{shared_wg.http_port}" - with api_admin_session(url) as session: - role = api_create_role(url, session, {"name": f"role-{uuid4()}"}) - user = api_create_user( - url, - session, - { - "username": f"user-{uuid4()}", - "credentials": [ - { - "kind": "Password", - "hash": "123", - }, - ], - }, - ) - api_add_role_to_user(url, session, user["id"], role["id"]) - target = api_create_target( - url, - session, - { - "name": f"mysql-{uuid4()}", - "options": { - "kind": "MySql", - "host": "localhost", - "port": db_port, - "username": "root", - "password": "123", - "tls": { - "mode": "Preferred", - "verify": False, - }, - }, - }, - ) - api_add_role_to_target(url, session, target["id"], role["id"]) +# class Test: +# def test( +# self, +# processes: ProcessManager, +# timeout, +# shared_wg: WarpgateProcess, +# ): +# db_port = processes.start_mysql_server() +# url = f"https://localhost:{shared_wg.http_port}" +# with api_admin_session(url) as session: +# role = api_create_role(url, session, {"name": f"role-{uuid4()}"}) +# user = api_create_user( +# url, +# session, +# { +# "username": f"user-{uuid4()}", +# "credentials": [ +# { +# "kind": "Password", +# "hash": "123", +# }, +# ], +# }, +# ) +# api_add_role_to_user(url, session, user["id"], role["id"]) +# target = api_create_target( +# url, +# session, +# { +# "name": f"mysql-{uuid4()}", +# "options": { +# "kind": "MySql", +# "host": "localhost", +# "port": db_port, +# "username": "root", +# "password": "123", +# "tls": { +# "mode": "Preferred", +# "verify": False, +# }, +# }, +# }, +# ) +# api_add_role_to_target(url, session, target["id"], role["id"]) - time.sleep(15) - wait_mysql_port(db_port) - wait_port(shared_wg.mysql_port, recv=False) - time.sleep(15) +# time.sleep(15) +# wait_mysql_port(db_port) +# wait_port(shared_wg.mysql_port, recv=False) +# time.sleep(15) - client = processes.start( - [ - "mysql", - "--user", - f"{user['username']}#{target['name']}", - "-p123", - "--host", - "127.0.0.1", - "--port", - str(shared_wg.mysql_port), - *mysql_client_opts, - mysql_client_ssl_opt, - "db", - ], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - ) - assert b"\ndb\n" in client.communicate(b"show schemas;", timeout=timeout)[0] - assert client.returncode == 0 +# client = processes.start( +# [ +# "mysql", +# "--user", +# f"{user['username']}#{target['name']}", +# "-p123", +# "--host", +# "127.0.0.1", +# "--port", +# str(shared_wg.mysql_port), +# *mysql_client_opts, +# mysql_client_ssl_opt, +# "db", +# ], +# stdin=subprocess.PIPE, +# stdout=subprocess.PIPE, +# ) +# assert b"\ndb\n" in client.communicate(b"show schemas;", timeout=timeout)[0] +# assert client.returncode == 0 - client = processes.start( - [ - "mysql", - "--user", - f"{user['username']}#{target['name']}", - "-pwrong", - "--host", - "127.0.0.1", - "--port", - str(shared_wg.mysql_port), - *mysql_client_opts, - mysql_client_ssl_opt, - "db", - ], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - ) - client.communicate(b"show schemas;", timeout=timeout) - assert client.returncode != 0 +# client = processes.start( +# [ +# "mysql", +# "--user", +# f"{user['username']}#{target['name']}", +# "-pwrong", +# "--host", +# "127.0.0.1", +# "--port", +# str(shared_wg.mysql_port), +# *mysql_client_opts, +# mysql_client_ssl_opt, +# "db", +# ], +# stdin=subprocess.PIPE, +# stdout=subprocess.PIPE, +# ) +# client.communicate(b"show schemas;", timeout=timeout) +# assert client.returncode != 0