| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| from typing import Any |
| |
| import pytest |
| |
| from apache_iggy import ( |
| GlobalPermissions, |
| IggyClient, |
| Permissions, |
| StreamPermissions, |
| TopicPermissions, |
| ) |
| |
| from .utils import ( |
| MAX_PASSWORD_BYTES, |
| MAX_USERNAME_BYTES, |
| MIN_PASSWORD_BYTES, |
| get_server_config, |
| login_fresh_client, |
| unique_credentials, |
| wait_for_ping, |
| ) |
| |
| GLOBAL_PERMISSION_FLAGS = [ |
| "manage_servers", |
| "read_servers", |
| "manage_users", |
| "read_users", |
| "manage_streams", |
| "read_streams", |
| "manage_topics", |
| "read_topics", |
| "poll_messages", |
| "send_messages", |
| ] |
| |
| STREAM_PERMISSION_FLAGS = [ |
| "manage_stream", |
| "read_stream", |
| "manage_topics", |
| "read_topics", |
| "poll_messages", |
| "send_messages", |
| ] |
| |
| TOPIC_PERMISSION_FLAGS = [ |
| "manage_topic", |
| "read_topic", |
| "poll_messages", |
| "send_messages", |
| ] |
| |
| |
| class TestPermissionsModel: |
| """Test constructing the Permissions classes without a server.""" |
| |
| def test_default_permissions_deny_everything(self): |
| """Test Permissions() has every global flag off and no streams.""" |
| permissions = Permissions() |
| |
| assert permissions.streams is None |
| global_permissions = permissions.global_permissions |
| assert global_permissions.manage_servers is False |
| assert global_permissions.read_servers is False |
| assert global_permissions.manage_users is False |
| assert global_permissions.read_users is False |
| assert global_permissions.manage_streams is False |
| assert global_permissions.read_streams is False |
| assert global_permissions.manage_topics is False |
| assert global_permissions.read_topics is False |
| assert global_permissions.poll_messages is False |
| assert global_permissions.send_messages is False |
| |
| @pytest.mark.parametrize("flag", GLOBAL_PERMISSION_FLAGS) |
| def test_each_global_permission_flag_round_trips_alone(self, flag): |
| """Test setting one flag flips only that flag.""" |
| global_permissions = GlobalPermissions(**{flag: True}) |
| |
| for name in GLOBAL_PERMISSION_FLAGS: |
| assert getattr(global_permissions, name) is (name == flag) |
| |
| @pytest.mark.parametrize("flag", STREAM_PERMISSION_FLAGS) |
| def test_each_stream_permission_flag_round_trips_alone(self, flag): |
| """Test setting one flag flips only that flag.""" |
| flags: dict[str, Any] = {flag: True} |
| stream_permissions = StreamPermissions(**flags) |
| |
| for name in STREAM_PERMISSION_FLAGS: |
| assert getattr(stream_permissions, name) is (name == flag) |
| |
| @pytest.mark.parametrize("flag", TOPIC_PERMISSION_FLAGS) |
| def test_each_topic_permission_flag_round_trips_alone(self, flag): |
| """Test setting one flag flips only that flag.""" |
| topic_permissions = TopicPermissions(**{flag: True}) |
| |
| for name in TOPIC_PERMISSION_FLAGS: |
| assert getattr(topic_permissions, name) is (name == flag) |
| |
| def test_nested_stream_and_topic_permissions_are_preserved(self): |
| """Test stream and topic permission dicts survive construction.""" |
| permissions = Permissions( |
| global_permissions=GlobalPermissions(read_servers=True), |
| streams={ |
| 1: StreamPermissions( |
| read_stream=True, |
| topics={7: TopicPermissions(poll_messages=True)}, |
| ), |
| 42: StreamPermissions(manage_stream=True), |
| }, |
| ) |
| |
| streams = permissions.streams |
| assert streams is not None |
| assert set(streams) == {1, 42} |
| assert streams[1].read_stream is True |
| assert streams[1].manage_stream is False |
| topics = streams[1].topics |
| assert topics is not None |
| assert set(topics) == {7} |
| assert topics[7].poll_messages is True |
| assert topics[7].manage_topic is False |
| assert streams[42].manage_stream is True |
| assert streams[42].topics is None |
| |
| def test_empty_permission_dicts_normalize_to_none(self): |
| """Test empty stream and topic dicts are equivalent to None.""" |
| assert Permissions(streams={}).streams is None |
| assert Permissions(streams={}) == Permissions() |
| assert StreamPermissions(topics={}).topics is None |
| assert StreamPermissions(topics={}) == StreamPermissions() |
| |
| def test_stream_id_above_u32_is_rejected(self): |
| """Test dict keys outside the u32 wire range fail at construction.""" |
| with pytest.raises(OverflowError): |
| Permissions(streams={2**32: StreamPermissions(read_stream=True)}) |
| with pytest.raises(OverflowError): |
| StreamPermissions(topics={2**32: TopicPermissions(read_topic=True)}) |
| |
| def test_permissions_equality(self): |
| """Test structurally identical Permissions compare equal.""" |
| first = Permissions( |
| global_permissions=GlobalPermissions(read_streams=True), |
| streams={3: StreamPermissions(poll_messages=True)}, |
| ) |
| second = Permissions( |
| global_permissions=GlobalPermissions(read_streams=True), |
| streams={3: StreamPermissions(poll_messages=True)}, |
| ) |
| different = Permissions( |
| global_permissions=GlobalPermissions(manage_streams=True) |
| ) |
| |
| assert first == second |
| assert first != different |
| |
| |
| class TestCreateUserWithPermissions: |
| """Test create_user with the permissions argument.""" |
| |
| @pytest.mark.asyncio |
| async def test_create_user_with_global_permissions_round_trips( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test global permissions set at creation come back from get_user.""" |
| username, password = unique_credentials(unique_name) |
| permissions = Permissions( |
| global_permissions=GlobalPermissions(read_users=True, read_streams=True), |
| streams={}, |
| ) |
| |
| created = await iggy_client.create_user( |
| username, password, permissions=permissions |
| ) |
| assert created.permissions == permissions |
| |
| fetched = await iggy_client.get_user(created.id) |
| assert fetched is not None |
| fetched_permissions = fetched.permissions |
| assert fetched_permissions is not None |
| assert fetched_permissions == permissions |
| assert fetched_permissions.streams is None |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_create_user_with_nested_permissions_round_trips( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test per-stream and per-topic permissions survive the round trip.""" |
| username, password = unique_credentials(unique_name) |
| stream_name = unique_name(prefix="perm-stream-") |
| await iggy_client.create_stream(stream_name) |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| topic_name = unique_name(prefix="perm-topic-") |
| await iggy_client.create_topic(stream_name, topic_name, partitions_count=1) |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| |
| permissions = Permissions( |
| streams={ |
| stream.id: StreamPermissions( |
| read_stream=True, |
| poll_messages=True, |
| topics={topic.id: TopicPermissions(send_messages=True)}, |
| ) |
| } |
| ) |
| created = await iggy_client.create_user( |
| username, password, permissions=permissions |
| ) |
| assert created.permissions == permissions |
| |
| fetched = await iggy_client.get_user(created.id) |
| assert fetched is not None |
| fetched_permissions = fetched.permissions |
| assert fetched_permissions is not None |
| assert fetched_permissions == permissions |
| streams = fetched_permissions.streams |
| assert streams is not None |
| assert streams[stream.id].read_stream is True |
| topics = streams[stream.id].topics |
| assert topics is not None |
| assert topics[topic.id].send_messages is True |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_user_with_manage_streams_can_create_stream( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a granted global permission is enforced for the new user.""" |
| username, password = unique_credentials(unique_name) |
| permissions = Permissions( |
| global_permissions=GlobalPermissions(manage_streams=True) |
| ) |
| created = await iggy_client.create_user( |
| username, password, permissions=permissions |
| ) |
| |
| client = await login_fresh_client(username, password) |
| stream_name = unique_name(prefix="granted-") |
| await client.create_stream(stream_name) |
| |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_user_without_permissions_cannot_create_stream( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a user created without permissions is denied stream creation.""" |
| username, password = unique_credentials(unique_name) |
| created = await iggy_client.create_user(username, password) |
| |
| client = await login_fresh_client(username, password) |
| with pytest.raises(RuntimeError): |
| await client.create_stream(unique_name(prefix="denied-")) |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_per_stream_permission_is_scoped_to_that_stream( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test read access to one stream does not leak to another stream.""" |
| readable_name = unique_name(prefix="readable-") |
| hidden_name = unique_name(prefix="hidden-") |
| await iggy_client.create_stream(readable_name) |
| await iggy_client.create_stream(hidden_name) |
| readable = await iggy_client.get_stream(readable_name) |
| assert readable is not None |
| |
| username, password = unique_credentials(unique_name) |
| permissions = Permissions( |
| streams={readable.id: StreamPermissions(read_stream=True)} |
| ) |
| created = await iggy_client.create_user( |
| username, password, permissions=permissions |
| ) |
| |
| client = await login_fresh_client(username, password) |
| visible = await client.get_stream(readable_name) |
| assert visible is not None |
| assert visible.id == readable.id |
| with pytest.raises(RuntimeError): |
| await client.get_stream(hidden_name) |
| |
| await iggy_client.delete_user(created.id) |
| |
| |
| class TestUpdatePermissions: |
| """Test permission updates via update_permissions.""" |
| |
| @pytest.mark.asyncio |
| async def test_update_permissions_grants_and_round_trips( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test permissions granted after creation come back from get_user.""" |
| username, password = unique_credentials(unique_name) |
| created = await iggy_client.create_user(username, password) |
| assert created.permissions is None |
| |
| permissions = Permissions( |
| global_permissions=GlobalPermissions(read_streams=True) |
| ) |
| await iggy_client.update_permissions(created.id, permissions) |
| |
| fetched = await iggy_client.get_user(created.id) |
| assert fetched is not None |
| assert fetched.permissions == permissions |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_update_permissions_replaces_previous_permissions( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_permissions overwrites instead of merging.""" |
| username, password = unique_credentials(unique_name) |
| initial = Permissions( |
| global_permissions=GlobalPermissions(read_streams=True, read_users=True) |
| ) |
| created = await iggy_client.create_user(username, password, permissions=initial) |
| |
| replacement = Permissions( |
| global_permissions=GlobalPermissions(read_servers=True) |
| ) |
| await iggy_client.update_permissions(created.id, replacement) |
| |
| fetched = await iggy_client.get_user(created.id) |
| assert fetched is not None |
| fetched_permissions = fetched.permissions |
| assert fetched_permissions is not None |
| assert fetched_permissions == replacement |
| assert fetched_permissions.global_permissions.read_streams is False |
| assert fetched_permissions.global_permissions.read_users is False |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_update_permissions_none_clears_permissions( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_permissions with None removes existing permissions.""" |
| username, password = unique_credentials(unique_name) |
| permissions = Permissions( |
| global_permissions=GlobalPermissions(read_streams=True) |
| ) |
| created = await iggy_client.create_user( |
| username, password, permissions=permissions |
| ) |
| |
| await iggy_client.update_permissions(created.id, None) |
| |
| fetched = await iggy_client.get_user(created.id) |
| assert fetched is not None |
| assert fetched.permissions is None |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_update_permissions_takes_effect_for_new_session( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a permission granted after creation is enforced on next login.""" |
| username, password = unique_credentials(unique_name) |
| created = await iggy_client.create_user(username, password) |
| |
| denied_client = await login_fresh_client(username, password) |
| with pytest.raises(RuntimeError): |
| await denied_client.create_stream(unique_name(prefix="before-grant-")) |
| |
| await iggy_client.update_permissions( |
| created.id, |
| Permissions(global_permissions=GlobalPermissions(manage_streams=True)), |
| ) |
| |
| granted_client = await login_fresh_client(username, password) |
| stream_name = unique_name(prefix="after-grant-") |
| await granted_client.create_stream(stream_name) |
| assert await iggy_client.get_stream(stream_name) is not None |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_user_without_manage_users_cannot_update_permissions( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a user lacking manage_users is denied updating another user.""" |
| actor_username, actor_password = unique_credentials(unique_name) |
| actor = await iggy_client.create_user(actor_username, actor_password) |
| target_username, target_password = unique_credentials(unique_name) |
| target = await iggy_client.create_user(target_username, target_password) |
| |
| actor_client = await login_fresh_client(actor_username, actor_password) |
| with pytest.raises(RuntimeError): |
| await actor_client.update_permissions( |
| target.id, |
| Permissions(global_permissions=GlobalPermissions(read_streams=True)), |
| ) |
| |
| await iggy_client.delete_user(actor.id) |
| await iggy_client.delete_user(target.id) |
| |
| @pytest.mark.asyncio |
| async def test_user_with_manage_users_can_update_permissions( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a user granted manage_users can update another user.""" |
| actor_username, actor_password = unique_credentials(unique_name) |
| actor = await iggy_client.create_user( |
| actor_username, |
| actor_password, |
| permissions=Permissions( |
| global_permissions=GlobalPermissions(manage_users=True) |
| ), |
| ) |
| target_username, target_password = unique_credentials(unique_name) |
| target = await iggy_client.create_user(target_username, target_password) |
| |
| granted = Permissions(global_permissions=GlobalPermissions(read_streams=True)) |
| actor_client = await login_fresh_client(actor_username, actor_password) |
| await actor_client.update_permissions(target.id, granted) |
| |
| fetched = await iggy_client.get_user(target.id) |
| assert fetched is not None |
| assert fetched.permissions == granted |
| |
| await iggy_client.delete_user(actor.id) |
| await iggy_client.delete_user(target.id) |
| |
| @pytest.mark.asyncio |
| async def test_failed_update_leaves_target_permissions_unchanged( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a denied update does not alter the target's permissions.""" |
| actor_username, actor_password = unique_credentials(unique_name) |
| actor = await iggy_client.create_user(actor_username, actor_password) |
| target_username, target_password = unique_credentials(unique_name) |
| initial = Permissions(global_permissions=GlobalPermissions(read_users=True)) |
| target = await iggy_client.create_user( |
| target_username, target_password, permissions=initial |
| ) |
| |
| actor_client = await login_fresh_client(actor_username, actor_password) |
| with pytest.raises(RuntimeError): |
| await actor_client.update_permissions( |
| target.id, |
| Permissions(global_permissions=GlobalPermissions(manage_servers=True)), |
| ) |
| |
| fetched = await iggy_client.get_user(target.id) |
| assert fetched is not None |
| assert fetched.permissions == initial |
| |
| await iggy_client.delete_user(actor.id) |
| await iggy_client.delete_user(target.id) |
| |
| @pytest.mark.asyncio |
| async def test_update_permissions_nonexistent_user_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_permissions raises for a non-existent user.""" |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_permissions( |
| unique_name(max_bytes=MAX_USERNAME_BYTES), |
| Permissions(global_permissions=GlobalPermissions(read_streams=True)), |
| ) |
| |
| |
| class TestChangePassword: |
| """Test password changes via change_password.""" |
| |
| @pytest.mark.asyncio |
| async def test_change_password_swaps_credentials( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test the new password works for login and the old one is rejected.""" |
| username, password = unique_credentials(unique_name) |
| new_password = unique_name(max_bytes=MAX_PASSWORD_BYTES) |
| created = await iggy_client.create_user(username, password) |
| |
| await iggy_client.change_password(created.id, password, new_password) |
| |
| host, port = get_server_config() |
| client = IggyClient(f"{host}:{port}") |
| await client.connect() |
| await wait_for_ping(client) |
| with pytest.raises(RuntimeError): |
| await client.login_user(username, password) |
| await client.login_user(username, new_password) |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_change_password_rejects_wrong_current_password( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test change_password fails when the current password is wrong.""" |
| username = unique_name(max_bytes=MAX_USERNAME_BYTES) |
| # Keep one byte of headroom so the wrong password stays within the |
| # server limit and the failure is about the mismatch, not the length. |
| password = unique_name(max_bytes=MAX_PASSWORD_BYTES - 1) |
| created = await iggy_client.create_user(username, password) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.change_password( |
| created.id, f"{password}x", unique_name(max_bytes=MAX_PASSWORD_BYTES) |
| ) |
| |
| client = await login_fresh_client(username, password) |
| await client.ping() |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_user_can_change_own_password( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a user without permissions can change its own password.""" |
| username, password = unique_credentials(unique_name) |
| new_password = unique_name(max_bytes=MAX_PASSWORD_BYTES) |
| created = await iggy_client.create_user(username, password) |
| |
| client = await login_fresh_client(username, password) |
| await client.change_password(created.id, password, new_password) |
| |
| relogin = await login_fresh_client(username, new_password) |
| await relogin.ping() |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_user_with_manage_users_can_change_other_password( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a user granted manage_users can change another user's password.""" |
| actor_username, actor_password = unique_credentials(unique_name) |
| actor = await iggy_client.create_user( |
| actor_username, |
| actor_password, |
| permissions=Permissions( |
| global_permissions=GlobalPermissions(manage_users=True) |
| ), |
| ) |
| target_username, target_password = unique_credentials(unique_name) |
| target = await iggy_client.create_user(target_username, target_password) |
| new_password = unique_name(max_bytes=MAX_PASSWORD_BYTES) |
| |
| actor_client = await login_fresh_client(actor_username, actor_password) |
| await actor_client.change_password(target.id, target_password, new_password) |
| |
| relogin = await login_fresh_client(target_username, new_password) |
| await relogin.ping() |
| |
| await iggy_client.delete_user(actor.id) |
| await iggy_client.delete_user(target.id) |
| |
| @pytest.mark.asyncio |
| async def test_failed_password_change_leaves_target_credentials_valid( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a user lacking manage_users is denied and old credentials stay valid.""" |
| actor_username, actor_password = unique_credentials(unique_name) |
| actor = await iggy_client.create_user(actor_username, actor_password) |
| target_username, target_password = unique_credentials(unique_name) |
| target = await iggy_client.create_user(target_username, target_password) |
| |
| actor_client = await login_fresh_client(actor_username, actor_password) |
| with pytest.raises(RuntimeError): |
| await actor_client.change_password( |
| target.id, target_password, unique_name(max_bytes=MAX_PASSWORD_BYTES) |
| ) |
| |
| target_client = await login_fresh_client(target_username, target_password) |
| await target_client.ping() |
| |
| await iggy_client.delete_user(actor.id) |
| await iggy_client.delete_user(target.id) |
| |
| @pytest.mark.parametrize( |
| "new_password", |
| ["a" * (MIN_PASSWORD_BYTES - 1), "a" * (MAX_PASSWORD_BYTES + 1)], |
| ids=["too-short", "too-long"], |
| ) |
| @pytest.mark.asyncio |
| async def test_change_password_rejects_out_of_bounds_new_password( |
| self, iggy_client: IggyClient, unique_name, new_password |
| ): |
| """Test change_password rejects new passwords outside the 3-100 byte range.""" |
| username, password = unique_credentials(unique_name) |
| created = await iggy_client.create_user(username, password) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.change_password(created.id, password, new_password) |
| |
| client = await login_fresh_client(username, password) |
| await client.ping() |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_change_password_nonexistent_user_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test change_password raises for a non-existent user.""" |
| with pytest.raises(RuntimeError): |
| await iggy_client.change_password( |
| unique_name(max_bytes=MAX_USERNAME_BYTES), |
| unique_name(max_bytes=MAX_PASSWORD_BYTES), |
| unique_name(max_bytes=MAX_PASSWORD_BYTES), |
| ) |
| |
| |
| class TestLogoutUser: |
| """Test session termination via logout_user.""" |
| |
| @pytest.mark.asyncio |
| async def test_logout_user_ends_the_session( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test authenticated calls fail after logout.""" |
| username, password = unique_credentials(unique_name) |
| created = await iggy_client.create_user( |
| username, |
| password, |
| permissions=Permissions( |
| global_permissions=GlobalPermissions(read_users=True) |
| ), |
| ) |
| |
| client = await login_fresh_client(username, password) |
| assert await client.get_user(username) is not None |
| |
| await client.logout_user() |
| |
| with pytest.raises(RuntimeError): |
| await client.get_user(username) |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_logout_twice_fails(self, iggy_client: IggyClient, unique_name): |
| """Test a second logout on the same session raises.""" |
| username, password = unique_credentials(unique_name) |
| created = await iggy_client.create_user(username, password) |
| |
| client = await login_fresh_client(username, password) |
| await client.logout_user() |
| |
| with pytest.raises(RuntimeError): |
| await client.logout_user() |
| |
| await iggy_client.delete_user(created.id) |
| |
| @pytest.mark.asyncio |
| async def test_login_after_logout_restores_the_session( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test the same client can log in again after logging out.""" |
| username, password = unique_credentials(unique_name) |
| created = await iggy_client.create_user( |
| username, |
| password, |
| permissions=Permissions( |
| global_permissions=GlobalPermissions(read_users=True) |
| ), |
| ) |
| |
| client = await login_fresh_client(username, password) |
| await client.logout_user() |
| |
| await client.login_user(username, password) |
| assert await client.get_user(username) is not None |
| |
| await iggy_client.delete_user(created.id) |
