# 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.

import pytest

from apache_iggy import (
    GlobalPermissions,
    IggyClient,
    Permissions,
    UserInfoDetails,
    UserStatus,
)

from .utils import (
    MAX_PASSWORD_BYTES,
    MAX_USERNAME_BYTES,
    MIN_PASSWORD_BYTES,
    MIN_USERNAME_BYTES,
    get_server_config,
    unique_credentials,
    wait_for_ping,
    wait_for_server,
)


class TestCreateUser:
    """Test user creation via create_user."""

    @pytest.mark.asyncio
    async def test_create_and_get_user(self, iggy_client: IggyClient, unique_name):
        """Test user creation returns details and the user is retrievable."""
        username, password = unique_credentials(unique_name)

        created = await iggy_client.create_user(username, password, UserStatus.Active)
        assert isinstance(created, UserInfoDetails)
        assert isinstance(created.id, int)
        assert created.username == username
        assert created.status == UserStatus.Active
        assert created.permissions is None

        user_by_name = await iggy_client.get_user(username)
        assert user_by_name is not None
        assert user_by_name.id == created.id
        assert user_by_name.username == username
        assert user_by_name.status == UserStatus.Active
        assert user_by_name.permissions is None

        user_by_id = await iggy_client.get_user(created.id)
        assert user_by_id is not None
        assert user_by_id.id == created.id
        assert user_by_id.username == username

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_create_user_defaults_to_active_status(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test create_user without an explicit status creates an active user."""
        username, password = unique_credentials(unique_name)

        created = await iggy_client.create_user(username, password)
        assert created.status == UserStatus.Active

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_create_inactive_user(self, iggy_client: IggyClient, unique_name):
        """Test create_user with UserStatus.Inactive creates an inactive user."""
        username, password = unique_credentials(unique_name)

        created = await iggy_client.create_user(username, password, UserStatus.Inactive)
        assert created.status == UserStatus.Inactive

        user = await iggy_client.get_user(username)
        assert user is not None
        assert user.status == UserStatus.Inactive

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_duplicate_username_fails(self, iggy_client: IggyClient, unique_name):
        """Test create_user rejects an already taken username."""
        username, password = unique_credentials(unique_name)

        created = await iggy_client.create_user(username, password)

        with pytest.raises(RuntimeError):
            await iggy_client.create_user(username, password)

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_created_user_can_login(self, iggy_client: IggyClient, unique_name):
        """Test a freshly created user can authenticate with its credentials."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        host, port = get_server_config()
        client = IggyClient(f"{host}:{port}")
        await client.connect()
        await wait_for_ping(client)
        await client.login_user(username, password)

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_inactive_user_cannot_login(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test an inactive user is denied login even with correct credentials."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password, UserStatus.Inactive)

        host, port = get_server_config()
        client = IggyClient(f"{host}:{port}")
        await client.connect()
        with pytest.raises(RuntimeError):
            await client.login_user(username, password)

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_deleted_user_cannot_login(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test a deleted user cannot start a fresh authenticated session."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)
        await iggy_client.delete_user(created.id)

        host, port = get_server_config()
        client = IggyClient(f"{host}:{port}")
        await client.connect()
        with pytest.raises(RuntimeError):
            await client.login_user(username, password)

    @pytest.mark.parametrize(
        "username",
        ["a" * (MIN_USERNAME_BYTES - 1), "a" * (MAX_USERNAME_BYTES + 1)],
        ids=["too-short", "too-long"],
    )
    @pytest.mark.asyncio
    async def test_create_user_rejects_out_of_bounds_username(
        self, iggy_client: IggyClient, unique_name, username
    ):
        """Test create_user rejects usernames outside the 3-50 byte range."""
        with pytest.raises(RuntimeError):
            await iggy_client.create_user(
                username, unique_name(max_bytes=MAX_PASSWORD_BYTES)
            )

    @pytest.mark.parametrize(
        "password",
        ["a" * (MIN_PASSWORD_BYTES - 1), "a" * (MAX_PASSWORD_BYTES + 1)],
        ids=["too-short", "too-long"],
    )
    @pytest.mark.asyncio
    async def test_create_user_rejects_out_of_bounds_password(
        self, iggy_client: IggyClient, unique_name, password
    ):
        """Test create_user rejects passwords outside the 3-100 byte range."""
        with pytest.raises(RuntimeError):
            await iggy_client.create_user(
                unique_name(max_bytes=MAX_USERNAME_BYTES), password
            )

    @pytest.mark.parametrize(
        "prefix",
        ["ユーザー", "사용자", "ผู้ใช้", "🦀🚀"],
        ids=["japanese", "korean", "thai", "emoji"],
    )
    @pytest.mark.asyncio
    async def test_create_user_accepts_multibyte_credentials(
        self, iggy_client: IggyClient, unique_name, prefix
    ):
        """Test non-ascii credentials within the byte limits are accepted."""
        username = f"{prefix}{unique_name(min_bytes=4, max_bytes=8)}"
        password = f"{prefix}{unique_name(min_bytes=4, max_bytes=8)}"
        assert len(username.encode()) <= MAX_USERNAME_BYTES

        created = await iggy_client.create_user(username, password)
        assert created.username == username

        fetched = await iggy_client.get_user(username)
        assert fetched is not None
        assert fetched.id == created.id

        host, port = get_server_config()
        client = IggyClient(f"{host}:{port}")
        await client.connect()
        await client.login_user(username, password)

        await iggy_client.delete_user(created.id)


class TestGetUser:
    """Test user retrieval via get_user."""

    @pytest.mark.asyncio
    async def test_get_root_user(self, iggy_client: IggyClient):
        """Test the default root user is retrievable by username."""
        user = await iggy_client.get_user("iggy")
        assert user is not None
        assert user.username == "iggy"
        assert user.status == UserStatus.Active

    @pytest.mark.asyncio
    async def test_get_nonexistent_user_returns_none(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test getting a non-existent user by name or numeric id returns None."""
        user_by_name = await iggy_client.get_user(
            unique_name(max_bytes=MAX_USERNAME_BYTES)
        )
        assert user_by_name is None

        # Deleting a freshly created user guarantees its id is vacant.
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)
        await iggy_client.delete_user(created.id)

        user_by_id = await iggy_client.get_user(created.id)
        assert user_by_id is None

    @pytest.mark.asyncio
    async def test_get_user_rejects_empty_identifier_locally(
        self, iggy_client: IggyClient
    ):
        """Test the empty string is rejected client-side before any server call."""
        with pytest.raises(ValueError):
            iggy_client.get_user("")

    @pytest.mark.parametrize("user_id", [-1, 2**32], ids=["negative", "above-u32"])
    @pytest.mark.asyncio
    async def test_get_user_rejects_out_of_range_numeric_id(
        self, iggy_client: IggyClient, user_id
    ):
        """Test numeric ids outside the u32 range are rejected client-side."""
        with pytest.raises(TypeError):
            iggy_client.get_user(user_id)

    @pytest.mark.asyncio
    async def test_get_user_returns_same_result_when_called_repeatedly(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test repeated get_user calls return the same user view."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        first = await iggy_client.get_user(username)
        second = await iggy_client.get_user(username)
        assert first is not None
        assert second is not None
        assert first.id == second.id
        assert first.username == second.username
        assert first.status == second.status

        await iggy_client.delete_user(created.id)


class TestGetUsers:
    """Test listing users via get_users."""

    @pytest.mark.asyncio
    async def test_get_users_lists_root_and_created_users(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test get_users returns the root user and newly created users."""
        first_username, first_password = unique_credentials(unique_name)
        second_username, second_password = unique_credentials(unique_name)

        first = await iggy_client.create_user(first_username, first_password)
        second = await iggy_client.create_user(second_username, second_password)

        users = await iggy_client.get_users()
        usernames = [user.username for user in users]
        assert "iggy" in usernames
        assert first_username in usernames
        assert second_username in usernames
        assert all(isinstance(user.id, int) for user in users)
        assert all(isinstance(user.status, UserStatus) for user in users)

        user_ids = [user.id for user in users]
        assert user_ids == sorted(set(user_ids)), (
            "get_users must return users in strictly ascending id order"
        )

        await iggy_client.delete_user(first.id)
        await iggy_client.delete_user(second.id)


class TestUpdateUser:
    """Test user updates via update_user."""

    @pytest.mark.asyncio
    async def test_update_username(self, iggy_client: IggyClient, unique_name):
        """Test update_user renames a user."""
        username, password = unique_credentials(unique_name)
        new_username = unique_name(max_bytes=MAX_USERNAME_BYTES)

        created = await iggy_client.create_user(username, password)

        await iggy_client.update_user(username, username=new_username)

        assert await iggy_client.get_user(username) is None
        renamed = await iggy_client.get_user(new_username)
        assert renamed is not None
        assert renamed.id == created.id
        assert renamed.username == new_username

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_update_status(self, iggy_client: IggyClient, unique_name):
        """Test update_user changes the user status."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)
        assert created.status == UserStatus.Active

        await iggy_client.update_user(created.id, status=UserStatus.Inactive)

        user = await iggy_client.get_user(created.id)
        assert user is not None
        assert user.status == UserStatus.Inactive
        assert user.username == username

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_update_username_and_status_together(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test update_user applies a new username and status in one call."""
        username, password = unique_credentials(unique_name)
        new_username = unique_name(max_bytes=MAX_USERNAME_BYTES)
        created = await iggy_client.create_user(username, password)

        await iggy_client.update_user(
            created.id, username=new_username, status=UserStatus.Inactive
        )

        user = await iggy_client.get_user(created.id)
        assert user is not None
        assert user.username == new_username
        assert user.status == UserStatus.Inactive

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_update_user_with_empty_options_succeeds(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test update_user accepts an empty options map."""
        username, password = unique_credentials(unique_name)
        new_username = unique_name(max_bytes=MAX_USERNAME_BYTES)
        created = await iggy_client.create_user(username, password)

        await iggy_client.update_user(created.id, username=new_username, options={})

        user = await iggy_client.get_user(created.id)
        assert user is not None
        assert user.username == new_username

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_update_user_forwards_options(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test update_user forwards option keys to the server."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        with pytest.raises(RuntimeError) as rejection:
            await iggy_client.update_user(
                created.id,
                options={"unknown": "value"},
            )

        # Binary transports carry the code alone, so the key itself is empty.
        assert "Unsupported option key" in str(rejection.value)

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_update_user_with_no_fields_is_a_noop(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test the server permits an update with neither username nor status."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        await iggy_client.update_user(created.id)

        user = await iggy_client.get_user(created.id)
        assert user is not None
        assert user.username == username
        assert user.status == UserStatus.Active

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_update_user_applied_repeatedly_is_idempotent(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test applying the same update twice succeeds and keeps the state."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        await iggy_client.update_user(created.id, status=UserStatus.Inactive)
        await iggy_client.update_user(created.id, status=UserStatus.Inactive)

        user = await iggy_client.get_user(created.id)
        assert user is not None
        assert user.status == UserStatus.Inactive
        assert user.username == username

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_update_username_to_existing_username_fails(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test update_user rejects renaming to a username another user holds."""
        first_username, first_password = unique_credentials(unique_name)
        second_username, second_password = unique_credentials(unique_name)
        first = await iggy_client.create_user(first_username, first_password)
        second = await iggy_client.create_user(second_username, second_password)

        with pytest.raises(RuntimeError):
            await iggy_client.update_user(first.id, username=second_username)

        unchanged = await iggy_client.get_user(first.id)
        assert unchanged is not None
        assert unchanged.username == first_username
        assert unchanged.status == UserStatus.Active

        await iggy_client.delete_user(first.id)
        await iggy_client.delete_user(second.id)

    @pytest.mark.asyncio
    async def test_update_nonexistent_user_fails(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test update_user raises for a non-existent user."""
        with pytest.raises(RuntimeError):
            await iggy_client.update_user(
                unique_name(max_bytes=MAX_USERNAME_BYTES),
                status=UserStatus.Inactive,
            )

    @pytest.mark.parametrize(
        "new_username",
        ["a" * (MIN_USERNAME_BYTES - 1), "a" * (MAX_USERNAME_BYTES + 1), "あ" * 17],
        ids=["too-short", "too-long", "multibyte-51-bytes"],
    )
    @pytest.mark.asyncio
    async def test_update_user_rejects_out_of_bounds_username(
        self, iggy_client: IggyClient, unique_name, new_username
    ):
        """Test update_user rejects usernames outside the 3-50 byte range."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        with pytest.raises(RuntimeError):
            await iggy_client.update_user(created.id, username=new_username)

        unchanged = await iggy_client.get_user(created.id)
        assert unchanged is not None
        assert unchanged.username == username
        assert unchanged.status == UserStatus.Active

        await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_update_user_accepts_multibyte_username(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test update_user accepts a non-ascii username within the byte limit."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        new_username = f"사용자{unique_name(min_bytes=4, max_bytes=8)}"
        await iggy_client.update_user(created.id, username=new_username)

        renamed = await iggy_client.get_user(created.id)
        assert renamed is not None
        assert renamed.username == new_username

        await iggy_client.delete_user(created.id)


class TestDeleteUser:
    """Test user deletion via delete_user."""

    @pytest.mark.asyncio
    async def test_delete_user_by_name(self, iggy_client: IggyClient, unique_name):
        """Test delete_user removes a user addressed by username."""
        username, password = unique_credentials(unique_name)
        await iggy_client.create_user(username, password)

        await iggy_client.delete_user(username)

        assert await iggy_client.get_user(username) is None

    @pytest.mark.asyncio
    async def test_delete_user_by_numeric_id(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test delete_user removes a user addressed by numeric id."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        await iggy_client.delete_user(created.id)

        assert await iggy_client.get_user(created.id) is None

    @pytest.mark.asyncio
    async def test_delete_nonexistent_user_fails(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test delete_user raises for a non-existent user."""
        with pytest.raises(RuntimeError):
            await iggy_client.delete_user(unique_name(max_bytes=MAX_USERNAME_BYTES))

    @pytest.mark.parametrize("root_identifier", ["iggy", 0], ids=["by-name", "by-id"])
    @pytest.mark.asyncio
    async def test_delete_root_user_fails(
        self, iggy_client: IggyClient, root_identifier
    ):
        """Test the root user cannot be deleted by username or numeric id."""
        with pytest.raises(RuntimeError):
            await iggy_client.delete_user(root_identifier)

        root = await iggy_client.get_user("iggy")
        assert root is not None
        assert root.status == UserStatus.Active

    @pytest.mark.asyncio
    async def test_delete_user_twice_fails(self, iggy_client: IggyClient, unique_name):
        """Test deleting the same user twice fails on the second call."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        await iggy_client.delete_user(created.id)

        with pytest.raises(RuntimeError):
            await iggy_client.delete_user(created.id)

    @pytest.mark.asyncio
    async def test_delete_inactive_user(self, iggy_client: IggyClient, unique_name):
        """Test an inactive user can be deleted."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password, UserStatus.Inactive)

        await iggy_client.delete_user(created.id)

        assert await iggy_client.get_user(created.id) is None

    @pytest.mark.asyncio
    async def test_deleted_user_disappears_from_listings(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test a deleted user is absent from both get_user and get_users."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(username, password)

        await iggy_client.delete_user(created.id)

        assert await iggy_client.get_user(username) is None
        assert await iggy_client.get_user(created.id) is None
        users = await iggy_client.get_users()
        assert created.id not in [user.id for user in users]
        assert username not in [user.username for user in users]

    @pytest.mark.asyncio
    async def test_deleted_user_live_session_loses_identity(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test a session owned by a deleted user can no longer act as that user."""
        username, password = unique_credentials(unique_name)
        created = await iggy_client.create_user(
            username,
            password,
            permissions=Permissions(
                global_permissions=GlobalPermissions(read_users=True)
            ),
        )

        host, port = get_server_config()
        client = IggyClient(f"{host}:{port}")
        await client.connect()
        await wait_for_ping(client)
        await client.login_user(username, password)
        users_before = await client.get_users()
        assert any(user.id == created.id for user in users_before)

        await iggy_client.delete_user(created.id)

        with pytest.raises(RuntimeError):
            await client.get_users()

    @pytest.mark.asyncio
    async def test_deleted_username_is_reusable_with_fresh_credentials(
        self, iggy_client: IggyClient, unique_name
    ):
        """Test a deleted username can be recreated without old password state."""
        username = unique_name(max_bytes=MAX_USERNAME_BYTES)
        password = unique_name(max_bytes=MAX_USERNAME_BYTES)
        new_password = f"{password}x"

        first = await iggy_client.create_user(username, password)
        await iggy_client.delete_user(first.id)

        recreated = await iggy_client.create_user(username, new_password)

        host, port = get_server_config()
        client = IggyClient(f"{host}:{port}")
        await client.connect()
        with pytest.raises(RuntimeError):
            await client.login_user(username, password)
        await client.login_user(username, new_password)

        await iggy_client.delete_user(recreated.id)


@pytest.mark.parametrize(
    "method_name",
    [
        "get_user",
        "get_users",
        "create_user",
        "update_user",
        "delete_user",
        "update_permissions",
        "change_password",
        "logout_user",
    ],
)
@pytest.mark.asyncio
async def test_user_management_requires_connection_and_auth(method_name, unique_name):
    """Test user management methods fail before connecting and before login."""
    host, port = get_server_config()
    wait_for_server(host, port)

    client = IggyClient(f"{host}:{port}")
    username = unique_name(max_bytes=MAX_USERNAME_BYTES)
    args_by_method = {
        "get_user": (username,),
        "get_users": (),
        "create_user": (username, "secret"),
        "update_user": (username,),
        "delete_user": (username,),
        "update_permissions": (username, None),
        "change_password": (username, "secret", "secret2"),
        "logout_user": (),
    }
    method = getattr(client, method_name)
    args = args_by_method[method_name]

    with pytest.raises(RuntimeError):
        await method(*args)

    await client.connect()
    with pytest.raises(RuntimeError):
        await method(*args)