# This file is auto-generated by oagen. Do not edit.

import json

import pytest
from workos import WorkOSClient, AsyncWorkOSClient
from tests.generated_helpers import load_fixture

from workos.vault.models import (
    CreateDataKeyResponse,
    DecryptResponse,
    ObjectMetadata,
    ObjectSummary,
    ObjectWithoutValue,
    VaultObject,
    VersionListResponse,
    VaultOrder,
)
from workos._pagination import AsyncPage, SyncPage
from workos._errors import (
    AuthenticationError,
    BadRequestError,
    NotFoundError,
    RateLimitExceededError,
    ServerError,
    UnprocessableEntityError,
)


class TestVault:
    def test_create_data_key(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("create_data_key_response.json"),
        )
        result = workos.vault.create_data_key(context={})
        assert isinstance(result, CreateDataKeyResponse)
        assert result.data_key == "DR9idtey9MpMrA1VRFrz30HB1yNgL2PoHZyjAkFeWgg="
        assert (
            result.encrypted_keys
            == "V09TLkVLTS52MQBiZjUxY2NlYy03OGI0LTUyMDAtYjM4My0zNTczMGU3MWVmNjEBATEBJGJmNjVlMzI2LTQzYTAtNGIyMC04OGM0LTA3ZmYzZGU1NDM0YwF0YmY2NWUzMjYtNDNhMC00YjIwLTg4YzQtMDdmZjNkZTU0MzRj"
        )
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/vault/v1/keys/data-key")
        body = json.loads(request.content)
        assert "context" in body

    def test_create_decrypt(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("decrypt_response.json"),
        )
        result = workos.vault.create_decrypt(keys="test_keys")
        assert isinstance(result, DecryptResponse)
        assert result.data_key == "DR9idtey9MpMrA1VRFrz30HB1yNgL2PoHZyjAkFeWgg="
        assert result.id == "bf51ccec-78b4-5200-b383-35730e71ef61"
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/vault/v1/keys/decrypt")
        body = json.loads(request.content)
        assert body["keys"] == "test_keys"

    def test_create_rekey(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("create_data_key_response.json"),
        )
        result = workos.vault.create_rekey(
            context={}, encrypted_keys="test_encrypted_keys"
        )
        assert isinstance(result, CreateDataKeyResponse)
        assert result.data_key == "DR9idtey9MpMrA1VRFrz30HB1yNgL2PoHZyjAkFeWgg="
        assert (
            result.encrypted_keys
            == "V09TLkVLTS52MQBiZjUxY2NlYy03OGI0LTUyMDAtYjM4My0zNTczMGU3MWVmNjEBATEBJGJmNjVlMzI2LTQzYTAtNGIyMC04OGM0LTA3ZmYzZGU1NDM0YwF0YmY2NWUzMjYtNDNhMC00YjIwLTg4YzQtMDdmZjNkZTU0MzRj"
        )
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/vault/v1/keys/rekey")
        body = json.loads(request.content)
        assert "context" in body
        assert body["encrypted_keys"] == "test_encrypted_keys"

    def test_list_kv(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("list_object_summary.json"),
        )
        page = workos.vault.list_kv()
        assert isinstance(page, SyncPage)
        assert len(page.data) == 1
        assert isinstance(page.data[0], ObjectSummary)

    def test_list_kv_empty_page(self, workos, httpx_mock):
        httpx_mock.add_response(json={"data": [], "list_metadata": {}})
        page = workos.vault.list_kv()
        assert isinstance(page, SyncPage)
        assert page.data == []

    def test_list_kv_encodes_query_params(self, workos, httpx_mock):
        httpx_mock.add_response(json={"data": [], "list_metadata": {}})
        workos.vault.list_kv(
            limit=10,
            before="cursor before",
            after="cursor/after",
            order=VaultOrder("asc"),
            search="value search/test",
            updated_after="value updatedAfter/test",
        )
        request = httpx_mock.get_request()
        assert request.url.params["limit"] == "10"
        assert request.url.params["before"] == "cursor before"
        assert request.url.params["after"] == "cursor/after"
        assert request.url.params["order"] == "asc"
        assert request.url.params["search"] == "value search/test"
        assert request.url.params["updatedAfter"] == "value updatedAfter/test"

    def test_create_kv(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("object_metadata.json"),
        )
        result = workos.vault.create_kv(
            key_context={}, name="test_name", value="test_value"
        )
        assert isinstance(result, ObjectMetadata)
        assert result.environment_id == "environment_01K8ZYT4AWJ6XP0E0S8CTBHE3P"
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/vault/v1/kv")
        body = json.loads(request.content)
        assert "key_context" in body
        assert body["name"] == "test_name"
        assert body["value"] == "test_value"

    def test_get_name(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("vault_object.json"),
        )
        result = workos.vault.get_name("test_name")
        assert isinstance(result, VaultObject)
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        assert result.name == "my-secret"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/vault/v1/kv/name/test_name")

    def test_get_kv(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("vault_object.json"),
        )
        result = workos.vault.get_kv("test_id")
        assert isinstance(result, VaultObject)
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        assert result.name == "my-secret"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/vault/v1/kv/test_id")

    def test_update_kv(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("object_without_value.json"),
        )
        result = workos.vault.update_kv("test_id", value="test_value")
        assert isinstance(result, ObjectWithoutValue)
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        assert result.name == "my-secret"
        request = httpx_mock.get_request()
        assert request.method == "PUT"
        assert request.url.path.endswith("/vault/v1/kv/test_id")
        body = json.loads(request.content)
        assert body["value"] == "test_value"

    def test_delete_kv(self, workos, httpx_mock):
        httpx_mock.add_response(status_code=204)
        result = workos.vault.delete_kv("test_id")
        assert result is None
        request = httpx_mock.get_request()
        assert request.method == "DELETE"
        assert request.url.path.endswith("/vault/v1/kv/test_id")

    def test_delete_kv_encodes_query_params(self, workos, httpx_mock):
        httpx_mock.add_response(status_code=204)
        workos.vault.delete_kv("test_id", version_check="value version_check/test")
        request = httpx_mock.get_request()
        assert request.url.params["version_check"] == "value version_check/test"

    def test_list_kv_metadata(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("object_without_value.json"),
        )
        result = workos.vault.list_kv_metadata("test_id")
        assert isinstance(result, ObjectWithoutValue)
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        assert result.name == "my-secret"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/vault/v1/kv/test_id/metadata")

    def test_list_kv_versions(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("version_list_response.json"),
        )
        result = workos.vault.list_kv_versions("test_id")
        assert isinstance(result, VersionListResponse)
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/vault/v1/kv/test_id/versions")

    def test_create_data_key_with_request_options(self, workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("create_data_key_response.json"))
        workos.vault.create_data_key(
            context={}, request_options={"extra_headers": {"X-Custom": "value"}}
        )
        request = httpx_mock.get_request()
        assert request.headers["X-Custom"] == "value"

    def test_create_data_key_unauthorized(self, workos, httpx_mock):
        httpx_mock.add_response(
            status_code=401,
            json={"message": "Unauthorized"},
        )
        with pytest.raises(AuthenticationError):
            workos.vault.create_data_key(context={})

    def test_create_data_key_not_found(self, httpx_mock):
        workos = WorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(status_code=404, json={"message": "Not found"})
            with pytest.raises(NotFoundError):
                workos.vault.create_data_key(context={})
        finally:
            workos.close()

    def test_create_data_key_rate_limited(self, httpx_mock):
        workos = WorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(
                status_code=429,
                headers={"Retry-After": "0"},
                json={"message": "Slow down"},
            )
            with pytest.raises(RateLimitExceededError):
                workos.vault.create_data_key(context={})
        finally:
            workos.close()

    def test_create_data_key_server_error(self, httpx_mock):
        workos = WorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(status_code=500, json={"message": "Server error"})
            with pytest.raises(ServerError):
                workos.vault.create_data_key(context={})
        finally:
            workos.close()

    def test_create_data_key_bad_request(self, httpx_mock):
        workos = WorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(status_code=400, json={"message": "Bad request"})
            with pytest.raises(BadRequestError):
                workos.vault.create_data_key(context={})
        finally:
            workos.close()

    def test_create_data_key_unprocessable(self, httpx_mock):
        workos = WorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(status_code=422, json={"message": "Unprocessable"})
            with pytest.raises(UnprocessableEntityError):
                workos.vault.create_data_key(context={})
        finally:
            workos.close()

    # @oagen-ignore-start — client-side encrypt/decrypt tests (hand-maintained)

    def test_encrypt_decrypt_round_trip(self, workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("vault_data_key.json"))
        httpx_mock.add_response(json=load_fixture("vault_decrypt_key.json"))

        encrypted = workos.vault.encrypt(
            data="hello world", key_context={"tenant": "acme"}
        )
        assert isinstance(encrypted, str)
        assert encrypted != "hello world"

        decrypted = workos.vault.decrypt(encrypted_data=encrypted)
        assert decrypted == "hello world"

    def test_encrypt_decrypt_with_associated_data(self, workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("vault_data_key.json"))
        httpx_mock.add_response(json=load_fixture("vault_decrypt_key.json"))

        encrypted = workos.vault.encrypt(
            data="hello world",
            key_context={"tenant": "acme"},
            associated_data="context-info",
        )
        decrypted = workos.vault.decrypt(
            encrypted_data=encrypted, associated_data="context-info"
        )
        assert decrypted == "hello world"

    # @oagen-ignore-end


class TestAsyncVault:
    @pytest.mark.asyncio
    async def test_create_data_key(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("create_data_key_response.json"))
        result = await async_workos.vault.create_data_key(context={})
        assert isinstance(result, CreateDataKeyResponse)
        assert result.data_key == "DR9idtey9MpMrA1VRFrz30HB1yNgL2PoHZyjAkFeWgg="
        assert (
            result.encrypted_keys
            == "V09TLkVLTS52MQBiZjUxY2NlYy03OGI0LTUyMDAtYjM4My0zNTczMGU3MWVmNjEBATEBJGJmNjVlMzI2LTQzYTAtNGIyMC04OGM0LTA3ZmYzZGU1NDM0YwF0YmY2NWUzMjYtNDNhMC00YjIwLTg4YzQtMDdmZjNkZTU0MzRj"
        )
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/vault/v1/keys/data-key")

    @pytest.mark.asyncio
    async def test_create_decrypt(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("decrypt_response.json"))
        result = await async_workos.vault.create_decrypt(keys="test_keys")
        assert isinstance(result, DecryptResponse)
        assert result.data_key == "DR9idtey9MpMrA1VRFrz30HB1yNgL2PoHZyjAkFeWgg="
        assert result.id == "bf51ccec-78b4-5200-b383-35730e71ef61"
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/vault/v1/keys/decrypt")

    @pytest.mark.asyncio
    async def test_create_rekey(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("create_data_key_response.json"))
        result = await async_workos.vault.create_rekey(
            context={}, encrypted_keys="test_encrypted_keys"
        )
        assert isinstance(result, CreateDataKeyResponse)
        assert result.data_key == "DR9idtey9MpMrA1VRFrz30HB1yNgL2PoHZyjAkFeWgg="
        assert (
            result.encrypted_keys
            == "V09TLkVLTS52MQBiZjUxY2NlYy03OGI0LTUyMDAtYjM4My0zNTczMGU3MWVmNjEBATEBJGJmNjVlMzI2LTQzYTAtNGIyMC04OGM0LTA3ZmYzZGU1NDM0YwF0YmY2NWUzMjYtNDNhMC00YjIwLTg4YzQtMDdmZjNkZTU0MzRj"
        )
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/vault/v1/keys/rekey")

    @pytest.mark.asyncio
    async def test_list_kv(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("list_object_summary.json"))
        page = await async_workos.vault.list_kv()
        assert isinstance(page, AsyncPage)
        assert len(page.data) == 1
        assert isinstance(page.data[0], ObjectSummary)

    @pytest.mark.asyncio
    async def test_list_kv_empty_page(self, async_workos, httpx_mock):
        httpx_mock.add_response(json={"data": [], "list_metadata": {}})
        page = await async_workos.vault.list_kv()
        assert isinstance(page, AsyncPage)
        assert page.data == []

    @pytest.mark.asyncio
    async def test_list_kv_encodes_query_params(self, async_workos, httpx_mock):
        httpx_mock.add_response(json={"data": [], "list_metadata": {}})
        await async_workos.vault.list_kv(
            limit=10,
            before="cursor before",
            after="cursor/after",
            order=VaultOrder("asc"),
            search="value search/test",
            updated_after="value updatedAfter/test",
        )
        request = httpx_mock.get_request()
        assert request.url.params["limit"] == "10"
        assert request.url.params["before"] == "cursor before"
        assert request.url.params["after"] == "cursor/after"
        assert request.url.params["order"] == "asc"
        assert request.url.params["search"] == "value search/test"
        assert request.url.params["updatedAfter"] == "value updatedAfter/test"

    @pytest.mark.asyncio
    async def test_create_kv(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("object_metadata.json"))
        result = await async_workos.vault.create_kv(
            key_context={}, name="test_name", value="test_value"
        )
        assert isinstance(result, ObjectMetadata)
        assert result.environment_id == "environment_01K8ZYT4AWJ6XP0E0S8CTBHE3P"
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/vault/v1/kv")

    @pytest.mark.asyncio
    async def test_get_name(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("vault_object.json"))
        result = await async_workos.vault.get_name("test_name")
        assert isinstance(result, VaultObject)
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        assert result.name == "my-secret"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/vault/v1/kv/name/test_name")

    @pytest.mark.asyncio
    async def test_get_kv(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("vault_object.json"))
        result = await async_workos.vault.get_kv("test_id")
        assert isinstance(result, VaultObject)
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        assert result.name == "my-secret"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/vault/v1/kv/test_id")

    @pytest.mark.asyncio
    async def test_update_kv(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("object_without_value.json"))
        result = await async_workos.vault.update_kv("test_id", value="test_value")
        assert isinstance(result, ObjectWithoutValue)
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        assert result.name == "my-secret"
        request = httpx_mock.get_request()
        assert request.method == "PUT"
        assert request.url.path.endswith("/vault/v1/kv/test_id")

    @pytest.mark.asyncio
    async def test_delete_kv(self, async_workos, httpx_mock):
        httpx_mock.add_response(status_code=204)
        result = await async_workos.vault.delete_kv("test_id")
        assert result is None
        request = httpx_mock.get_request()
        assert request.method == "DELETE"
        assert request.url.path.endswith("/vault/v1/kv/test_id")

    @pytest.mark.asyncio
    async def test_delete_kv_encodes_query_params(self, async_workos, httpx_mock):
        httpx_mock.add_response(status_code=204)
        await async_workos.vault.delete_kv(
            "test_id", version_check="value version_check/test"
        )
        request = httpx_mock.get_request()
        assert request.url.params["version_check"] == "value version_check/test"

    @pytest.mark.asyncio
    async def test_list_kv_metadata(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("object_without_value.json"))
        result = await async_workos.vault.list_kv_metadata("test_id")
        assert isinstance(result, ObjectWithoutValue)
        assert result.id == "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
        assert result.name == "my-secret"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/vault/v1/kv/test_id/metadata")

    @pytest.mark.asyncio
    async def test_list_kv_versions(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("version_list_response.json"))
        result = await async_workos.vault.list_kv_versions("test_id")
        assert isinstance(result, VersionListResponse)
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/vault/v1/kv/test_id/versions")

    @pytest.mark.asyncio
    async def test_create_data_key_with_request_options(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("create_data_key_response.json"))
        await async_workos.vault.create_data_key(
            context={}, request_options={"extra_headers": {"X-Custom": "value"}}
        )
        request = httpx_mock.get_request()
        assert request.headers["X-Custom"] == "value"

    @pytest.mark.asyncio
    async def test_create_data_key_unauthorized(self, async_workos, httpx_mock):
        httpx_mock.add_response(status_code=401, json={"message": "Unauthorized"})
        with pytest.raises(AuthenticationError):
            await async_workos.vault.create_data_key(context={})

    @pytest.mark.asyncio
    async def test_create_data_key_not_found(self, httpx_mock):
        workos = AsyncWorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(status_code=404, json={"message": "Not found"})
            with pytest.raises(NotFoundError):
                await workos.vault.create_data_key(context={})
        finally:
            await workos.close()

    @pytest.mark.asyncio
    async def test_create_data_key_rate_limited(self, httpx_mock):
        workos = AsyncWorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(
                status_code=429,
                headers={"Retry-After": "0"},
                json={"message": "Slow down"},
            )
            with pytest.raises(RateLimitExceededError):
                await workos.vault.create_data_key(context={})
        finally:
            await workos.close()

    @pytest.mark.asyncio
    async def test_create_data_key_server_error(self, httpx_mock):
        workos = AsyncWorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(status_code=500, json={"message": "Server error"})
            with pytest.raises(ServerError):
                await workos.vault.create_data_key(context={})
        finally:
            await workos.close()

    @pytest.mark.asyncio
    async def test_create_data_key_bad_request(self, httpx_mock):
        workos = AsyncWorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(status_code=400, json={"message": "Bad request"})
            with pytest.raises(BadRequestError):
                await workos.vault.create_data_key(context={})
        finally:
            await workos.close()

    @pytest.mark.asyncio
    async def test_create_data_key_unprocessable(self, httpx_mock):
        workos = AsyncWorkOSClient(
            api_key="sk_test_123", client_id="client_test", max_retries=0
        )
        try:
            httpx_mock.add_response(status_code=422, json={"message": "Unprocessable"})
            with pytest.raises(UnprocessableEntityError):
                await workos.vault.create_data_key(context={})
        finally:
            await workos.close()

    # @oagen-ignore-start — client-side encrypt/decrypt tests (hand-maintained)

    @pytest.mark.asyncio
    async def test_encrypt_decrypt_round_trip(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("vault_data_key.json"))
        httpx_mock.add_response(json=load_fixture("vault_decrypt_key.json"))

        encrypted = await async_workos.vault.encrypt(
            data="hello world", key_context={"tenant": "acme"}
        )
        assert isinstance(encrypted, str)
        assert encrypted != "hello world"

        decrypted = await async_workos.vault.decrypt(encrypted_data=encrypted)
        assert decrypted == "hello world"

    @pytest.mark.asyncio
    async def test_encrypt_decrypt_with_associated_data(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("vault_data_key.json"))
        httpx_mock.add_response(json=load_fixture("vault_decrypt_key.json"))

        encrypted = await async_workos.vault.encrypt(
            data="hello world",
            key_context={"tenant": "acme"},
            associated_data="context-info",
        )
        decrypted = await async_workos.vault.decrypt(
            encrypted_data=encrypted, associated_data="context-info"
        )
        assert decrypted == "hello world"

    # @oagen-ignore-end
