diff --git a/integration/test_users.py b/integration/test_users.py index 223bedfd7..01a960506 100644 --- a/integration/test_users.py +++ b/integration/test_users.py @@ -1,9 +1,8 @@ -import random - import pytest +from _pytest.fixtures import SubRequest import weaviate -from integration.conftest import ClientFactory +from integration.conftest import ClientFactory, _sanitize_collection_name from weaviate.auth import Auth from weaviate.rbac.models import Role, RoleBase, UserTypes @@ -11,6 +10,10 @@ RBAC_AUTH_CREDS = Auth.api_key("admin-key") +def _unique_user_id(name: str) -> str: + return _sanitize_collection_name(name).lower() + + def test_own_user(client_factory: ClientFactory) -> None: with client_factory(ports=RBAC_PORTS, auth_credentials=RBAC_AUTH_CREDS) as client: if client._connection._weaviate_version.is_lower_than(1, 28, 0): @@ -75,129 +78,129 @@ def test_get_static_db_user(client_factory: ClientFactory) -> None: assert user.user_type == UserTypes.DB_STATIC -def test_create_user_and_get(client_factory: ClientFactory) -> None: +def test_create_user_and_get(client_factory: ClientFactory, request: SubRequest) -> None: with client_factory(ports=(RBAC_PORTS), auth_credentials=Auth.api_key("admin-key")) as client: if client._connection._weaviate_version.is_lower_than(1, 30, 0): pytest.skip("This test requires Weaviate 1.30.0 or higher") - randomUserName = "new-user" + str(random.randint(1, 1000)) - apiKey = client.users.db.create(user_id=randomUserName) + user_id = _unique_user_id(request.node.name) + api_key = client.users.db.create(user_id=user_id) with weaviate.connect_to_local( - port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], auth_credentials=Auth.api_key(apiKey) + port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], auth_credentials=Auth.api_key(api_key) ) as client2: user = client2.users.get_my_user() - assert user.user_id == randomUserName - dynamicUser = client.users.db.get(user_id=randomUserName) - assert dynamicUser is not None - assert dynamicUser.user_id == randomUserName - assert dynamicUser.user_type == UserTypes.DB_DYNAMIC - assert client.users.db.delete(user_id=randomUserName) + assert user.user_id == user_id + dynamic_user = client.users.db.get(user_id=user_id) + assert dynamic_user is not None + assert dynamic_user.user_id == user_id + assert dynamic_user.user_type == UserTypes.DB_DYNAMIC + assert client.users.db.delete(user_id=user_id) -def test_delete_user(client_factory: ClientFactory) -> None: +def test_delete_user(client_factory: ClientFactory, request: SubRequest) -> None: with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client: if client._connection._weaviate_version.is_lower_than(1, 30, 0): pytest.skip("This test requires Weaviate 1.30.0 or higher") - randomUserName = "new-user" + str(random.randint(1, 1000)) - client.users.db.create(user_id=randomUserName) - assert client.users.db.delete(user_id=randomUserName) + user_id = _unique_user_id(request.node.name) + client.users.db.create(user_id=user_id) + assert client.users.db.delete(user_id=user_id) assert not client.users.db.delete(user_id="I-do-not-exist") -def test_rotate_user_key(client_factory: ClientFactory) -> None: +def test_rotate_user_key(client_factory: ClientFactory, request: SubRequest) -> None: with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client: if client._connection._weaviate_version.is_lower_than(1, 30, 0): pytest.skip("This test requires Weaviate 1.30.0 or higher") - randomUserName = "new-user" + str(random.randint(1, 1000)) - apiKey = client.users.db.create(user_id=randomUserName) + user_id = _unique_user_id(request.node.name) + api_key = client.users.db.create(user_id=user_id) with weaviate.connect_to_local( - port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], auth_credentials=Auth.api_key(apiKey) + port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], auth_credentials=Auth.api_key(api_key) ) as client2: user = client2.users.get_my_user() - assert user.user_id == randomUserName + assert user.user_id == user_id - apiKeyNew = client.users.db.rotate_key(user_id=randomUserName) + api_key_new = client.users.db.rotate_key(user_id=user_id) with weaviate.connect_to_local( - port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], auth_credentials=Auth.api_key(apiKeyNew) + port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], auth_credentials=Auth.api_key(api_key_new) ) as client2: user = client2.users.get_my_user() - assert user.user_id == randomUserName + assert user.user_id == user_id - assert client.users.db.delete(user_id=randomUserName) + assert client.users.db.delete(user_id=user_id) -def test_de_activate(client_factory: ClientFactory) -> None: +def test_de_activate(client_factory: ClientFactory, request: SubRequest) -> None: with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client: if client._connection._weaviate_version.is_lower_than(1, 30, 0): pytest.skip("This test requires Weaviate 1.30.0 or higher") - randomUserName = "new-user" + str(random.randint(1, 1000)) - client.users.db.create(user_id=randomUserName) + user_id = _unique_user_id(request.node.name) + client.users.db.create(user_id=user_id) - assert client.users.db.deactivate(user_id=randomUserName) + assert client.users.db.deactivate(user_id=user_id) assert not client.users.db.deactivate( - user_id=randomUserName + user_id=user_id ) # second deactivation returns a conflict => false - user = client.users.db.get(user_id=randomUserName) + user = client.users.db.get(user_id=user_id) assert not user.active - assert client.users.db.activate(user_id=randomUserName) + assert client.users.db.activate(user_id=user_id) assert not client.users.db.activate( - user_id=randomUserName + user_id=user_id ) # second activation returns a conflict => false - user = client.users.db.get(user_id=randomUserName) + user = client.users.db.get(user_id=user_id) assert user.active - client.users.db.delete(user_id=randomUserName) + client.users.db.delete(user_id=user_id) -def test_deactivate_and_revoke(client_factory: ClientFactory) -> None: +def test_deactivate_and_revoke(client_factory: ClientFactory, request: SubRequest) -> None: with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client: if client._connection._weaviate_version.is_lower_than(1, 30, 0): pytest.skip("This test requires Weaviate 1.30.0 or higher") - randomUserName = "new-user" + str(random.randint(1, 1000)) - apiKeyOld = client.users.db.create(user_id=randomUserName) - assert client.users.db.deactivate(user_id=randomUserName, revoke_key=True) + user_id = _unique_user_id(request.node.name) + api_key_old = client.users.db.create(user_id=user_id) + assert client.users.db.deactivate(user_id=user_id, revoke_key=True) with pytest.raises(weaviate.exceptions.UnexpectedStatusCodeError): weaviate.connect_to_local( port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], - auth_credentials=Auth.api_key(apiKeyOld), + auth_credentials=Auth.api_key(api_key_old), ) # re-activating is not enough - assert client.users.db.activate(user_id=randomUserName) + assert client.users.db.activate(user_id=user_id) with pytest.raises(weaviate.exceptions.UnexpectedStatusCodeError): weaviate.connect_to_local( port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], - auth_credentials=Auth.api_key(apiKeyOld), + auth_credentials=Auth.api_key(api_key_old), ) - apiKeyNew = client.users.db.rotate_key(user_id=randomUserName) + api_key_new = client.users.db.rotate_key(user_id=user_id) with weaviate.connect_to_local( - port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], auth_credentials=Auth.api_key(apiKeyNew) + port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], auth_credentials=Auth.api_key(api_key_new) ) as client2: user = client2.users.get_my_user() - assert user.user_id == randomUserName + assert user.user_id == user_id - client.users.db.delete(user_id=randomUserName) + client.users.db.delete(user_id=user_id) -def test_deprecated_syntax(client_factory: ClientFactory) -> None: +def test_deprecated_syntax(client_factory: ClientFactory, request: SubRequest) -> None: with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client: if client._connection._weaviate_version.is_lower_than(1, 30, 0): pytest.skip("This test requires Weaviate 1.30.0 or higher") - randomUserName = "new-user" + str(random.randint(1, 1000)) - client.users.db.create(user_id=randomUserName) - roles = client.users.db.get_assigned_roles(user_id=randomUserName) + user_id = _unique_user_id(request.node.name) + client.users.db.create(user_id=user_id) + roles = client.users.db.get_assigned_roles(user_id=user_id) assert len(roles) == 0 - client.users.db.delete(user_id=randomUserName) + client.users.db.delete(user_id=user_id) def test_list_all_users(client_factory: ClientFactory) -> None: @@ -216,71 +219,77 @@ def test_list_all_users(client_factory: ClientFactory) -> None: client.users.db.delete(user_id=f"list-all-{i}") -def test_get_user_created_at_and_api_key_first_letters(client_factory: ClientFactory) -> None: +def test_get_user_created_at_and_api_key_first_letters( + client_factory: ClientFactory, request: SubRequest +) -> None: with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client: if client._connection._weaviate_version.is_lower_than(1, 30, 0): pytest.skip("This test requires Weaviate 1.30.0 or higher") - randomUserName = "new-user" + str(random.randint(1, 1000)) - client.users.db.create(user_id=randomUserName) + user_id = _unique_user_id(request.node.name) + client.users.db.create(user_id=user_id) try: - user = client.users.db.get(user_id=randomUserName) + user = client.users.db.get(user_id=user_id) assert user is not None assert user.created_at is not None assert user.api_key_first_letters is not None assert len(user.api_key_first_letters) > 0 finally: - client.users.db.delete(user_id=randomUserName) + client.users.db.delete(user_id=user_id) -def test_get_user_include_last_used_time(client_factory: ClientFactory) -> None: +def test_get_user_include_last_used_time( + client_factory: ClientFactory, request: SubRequest +) -> None: with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client: if client._connection._weaviate_version.is_lower_than(1, 30, 0): pytest.skip("This test requires Weaviate 1.30.0 or higher") - randomUserName = "new-user" + str(random.randint(1, 1000)) - apiKey = client.users.db.create(user_id=randomUserName) + user_id = _unique_user_id(request.node.name) + api_key = client.users.db.create(user_id=user_id) try: # log in with the new user to generate a lastUsedAt timestamp with weaviate.connect_to_local( port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], - auth_credentials=Auth.api_key(apiKey), + auth_credentials=Auth.api_key(api_key), ) as client2: - assert client2.users.get_my_user().user_id == randomUserName + assert client2.users.get_my_user().user_id == user_id # without include_last_used_time, last_used_time should be None - user = client.users.db.get(user_id=randomUserName) + user = client.users.db.get(user_id=user_id) assert user is not None assert user.last_used_time is None # with include_last_used_time=True, last_used_time should be populated - user = client.users.db.get(user_id=randomUserName, include_last_used_time=True) + user = client.users.db.get(user_id=user_id, include_last_used_time=True) assert user is not None assert user.last_used_time is not None finally: - client.users.db.delete(user_id=randomUserName) + client.users.db.delete(user_id=user_id) -def test_list_all_include_last_used_time(client_factory: ClientFactory) -> None: +def test_list_all_include_last_used_time( + client_factory: ClientFactory, request: SubRequest +) -> None: with client_factory(ports=RBAC_PORTS, auth_credentials=Auth.api_key("admin-key")) as client: if client._connection._weaviate_version.is_lower_than(1, 30, 0): pytest.skip("This test requires Weaviate 1.30.0 or higher") - randomUserName = "new-user" + str(random.randint(1, 1000)) - apiKey = client.users.db.create(user_id=randomUserName) + user_id = _unique_user_id(request.node.name) + api_key = client.users.db.create(user_id=user_id) try: # log in with the new user to generate a lastUsedAt timestamp with weaviate.connect_to_local( port=RBAC_PORTS[0], grpc_port=RBAC_PORTS[1], - auth_credentials=Auth.api_key(apiKey), + auth_credentials=Auth.api_key(api_key), ) as client2: - assert client2.users.get_my_user().user_id == randomUserName + assert client2.users.get_my_user().user_id == user_id # without include_last_used_time, last_used_time should be None users = client.users.db.list_all() - target = next((u for u in users if u.user_id == randomUserName), None) + target = next((u for u in users if u.user_id == user_id), None) assert target is not None assert target.created_at is not None assert target.api_key_first_letters is not None @@ -288,8 +297,8 @@ def test_list_all_include_last_used_time(client_factory: ClientFactory) -> None: # with include_last_used_time=True, last_used_time should be populated users = client.users.db.list_all(include_last_used_time=True) - target = next((u for u in users if u.user_id == randomUserName), None) + target = next((u for u in users if u.user_id == user_id), None) assert target is not None assert target.last_used_time is not None finally: - client.users.db.delete(user_id=randomUserName) + client.users.db.delete(user_id=user_id)