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

import json

import pytest

from tests.generated_helpers import load_fixture
from workos import AsyncWorkOSClient, WorkOSClient
from workos._errors import (
    AuthenticationError,
    BadRequestError,
    NotFoundError,
    RateLimitExceededError,
    ServerError,
    UnprocessableEntityError,
)
from workos._pagination import AsyncPage, SyncPage
from workos.common.models import PaginationOrder
from workos.sso.models import (
    Connection,
    ConnectionsConnectionType,
    Profile,
    SSOLogoutAuthorizeResponse,
    SSOTokenResponse,
)


class TestSSO:
    def test_list_connections(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("list_connection.json"),
        )
        page = workos.sso.list_connections()
        assert isinstance(page, SyncPage)
        assert len(page.data) == 1
        assert isinstance(page.data[0], Connection)

    def test_list_connections_empty_page(self, workos, httpx_mock):
        httpx_mock.add_response(json={"data": [], "list_metadata": {}})
        page = workos.sso.list_connections()
        assert isinstance(page, SyncPage)
        assert page.data == []

    def test_list_connections_encodes_query_params(self, workos, httpx_mock):
        httpx_mock.add_response(json={"data": [], "list_metadata": {}})
        workos.sso.list_connections(
            limit=10,
            before="cursor before",
            after="cursor/after",
            order=PaginationOrder("value_order"),
            connection_type=ConnectionsConnectionType("ADFSSAML"),
            domain="value domain/test",
            organization_id="value organization_id/test",
            search="value search/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"] == "value_order"
        assert request.url.params["connection_type"] == "ADFSSAML"
        assert request.url.params["domain"] == "value domain/test"
        assert request.url.params["organization_id"] == "value organization_id/test"
        assert request.url.params["search"] == "value search/test"

    def test_get_connection(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("connection.json"),
        )
        result = workos.sso.get_connection("test_id")
        assert isinstance(result, Connection)
        assert result.object == "connection"
        assert result.id == "conn_01E4ZCR3C56J083X43JQXF3JK5"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/connections/test_id")

    def test_delete_connection(self, workos, httpx_mock):
        httpx_mock.add_response(status_code=200, content=b"\n")
        result = workos.sso.delete_connection("test_id")
        assert result is None
        request = httpx_mock.get_request()
        assert request.method == "DELETE"
        assert request.url.path.endswith("/connections/test_id")

    def test_get_authorization_url(self, workos):
        result = workos.sso.get_authorization_url(redirect_uri="test_redirect_uri")
        assert isinstance(result, str)
        assert result.startswith("http")

    def test_get_logout_url(self, workos):
        result = workos.sso.get_logout_url(token="test_token")
        assert isinstance(result, str)
        assert result.startswith("http")

    def test_authorize_logout(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("sso_logout_authorize_response.json"),
        )
        result = workos.sso.authorize_logout(profile_id="test_profile_id")
        assert isinstance(result, SSOLogoutAuthorizeResponse)
        assert (
            result.logout_url
            == "https://auth.workos.com/sso/logout?token=eyJhbGciOiJSUzI1NiJ9"
        )
        assert (
            result.logout_token
            == "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJwcm9maWxlX2lkIjoicHJvZl8wMUdXUTFHMEgyRk02QVNFRjBIUzEzSENXOS0zMDRrZzAzZyIsImV4cCI6IjE1MTYyMzkwMjIifQ.Wru9Qlnf5DpohtGCKhZU4cVOd3zpiu7QQ-XEX--5A_4"
        )
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/sso/logout/authorize")
        body = json.loads(request.content)
        assert body["profile_id"] == "test_profile_id"

    def test_get_profile(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("profile.json"),
        )
        result = workos.sso.get_profile(access_token="test_access_token")
        assert isinstance(result, Profile)
        assert result.object == "profile"
        assert result.id == "prof_01DMC79VCBZ0NY2099737PSVF1"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/sso/profile")

    def test_get_profile_and_token(self, workos, httpx_mock):
        httpx_mock.add_response(
            json=load_fixture("sso_token_response.json"),
        )
        result = workos.sso.get_profile_and_token(code="test_code")
        assert isinstance(result, SSOTokenResponse)
        assert result.token_type == "Bearer"
        assert result.access_token == "eyJhbGciOiJSUzI1NiIsImtpZCI6InNzby..."
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/sso/token")
        body = json.loads(request.content)
        assert body["code"] == "test_code"

    def test_list_connections_with_request_options(self, workos, httpx_mock):
        httpx_mock.add_response(json={"data": [], "list_metadata": {}})
        workos.sso.list_connections(
            request_options={"extra_headers": {"X-Custom": "value"}}
        )
        request = httpx_mock.get_request()
        assert request.headers["X-Custom"] == "value"

    def test_list_connections_unauthorized(self, workos, httpx_mock):
        httpx_mock.add_response(
            status_code=401,
            json={"message": "Unauthorized"},
        )
        with pytest.raises(AuthenticationError):
            workos.sso.list_connections()

    def test_list_connections_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.sso.list_connections()
        finally:
            workos.close()

    def test_list_connections_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.sso.list_connections()
        finally:
            workos.close()

    def test_list_connections_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.sso.list_connections()
        finally:
            workos.close()

    def test_list_connections_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.sso.list_connections()
        finally:
            workos.close()

    def test_list_connections_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.sso.list_connections()
        finally:
            workos.close()


class TestAsyncSSO:
    @pytest.mark.asyncio
    async def test_list_connections(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("list_connection.json"))
        page = await async_workos.sso.list_connections()
        assert isinstance(page, AsyncPage)
        assert len(page.data) == 1
        assert isinstance(page.data[0], Connection)

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

    @pytest.mark.asyncio
    async def test_list_connections_encodes_query_params(
        self, async_workos, httpx_mock
    ):
        httpx_mock.add_response(json={"data": [], "list_metadata": {}})
        await async_workos.sso.list_connections(
            limit=10,
            before="cursor before",
            after="cursor/after",
            order=PaginationOrder("value_order"),
            connection_type=ConnectionsConnectionType("ADFSSAML"),
            domain="value domain/test",
            organization_id="value organization_id/test",
            search="value search/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"] == "value_order"
        assert request.url.params["connection_type"] == "ADFSSAML"
        assert request.url.params["domain"] == "value domain/test"
        assert request.url.params["organization_id"] == "value organization_id/test"
        assert request.url.params["search"] == "value search/test"

    @pytest.mark.asyncio
    async def test_get_connection(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("connection.json"))
        result = await async_workos.sso.get_connection("test_id")
        assert isinstance(result, Connection)
        assert result.object == "connection"
        assert result.id == "conn_01E4ZCR3C56J083X43JQXF3JK5"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/connections/test_id")

    @pytest.mark.asyncio
    async def test_delete_connection(self, async_workos, httpx_mock):
        httpx_mock.add_response(status_code=200, content=b"\n")
        result = await async_workos.sso.delete_connection("test_id")
        assert result is None
        request = httpx_mock.get_request()
        assert request.method == "DELETE"
        assert request.url.path.endswith("/connections/test_id")

    def test_get_authorization_url(self, async_workos):
        result = async_workos.sso.get_authorization_url(
            redirect_uri="test_redirect_uri"
        )
        assert isinstance(result, str)
        assert result.startswith("http")

    def test_get_logout_url(self, async_workos):
        result = async_workos.sso.get_logout_url(token="test_token")
        assert isinstance(result, str)
        assert result.startswith("http")

    @pytest.mark.asyncio
    async def test_authorize_logout(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("sso_logout_authorize_response.json"))
        result = await async_workos.sso.authorize_logout(profile_id="test_profile_id")
        assert isinstance(result, SSOLogoutAuthorizeResponse)
        assert (
            result.logout_url
            == "https://auth.workos.com/sso/logout?token=eyJhbGciOiJSUzI1NiJ9"
        )
        assert (
            result.logout_token
            == "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJwcm9maWxlX2lkIjoicHJvZl8wMUdXUTFHMEgyRk02QVNFRjBIUzEzSENXOS0zMDRrZzAzZyIsImV4cCI6IjE1MTYyMzkwMjIifQ.Wru9Qlnf5DpohtGCKhZU4cVOd3zpiu7QQ-XEX--5A_4"
        )
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/sso/logout/authorize")

    @pytest.mark.asyncio
    async def test_get_profile(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("profile.json"))
        result = await async_workos.sso.get_profile(access_token="test_access_token")
        assert isinstance(result, Profile)
        assert result.object == "profile"
        assert result.id == "prof_01DMC79VCBZ0NY2099737PSVF1"
        request = httpx_mock.get_request()
        assert request.method == "GET"
        assert request.url.path.endswith("/sso/profile")

    @pytest.mark.asyncio
    async def test_get_profile_and_token(self, async_workos, httpx_mock):
        httpx_mock.add_response(json=load_fixture("sso_token_response.json"))
        result = await async_workos.sso.get_profile_and_token(code="test_code")
        assert isinstance(result, SSOTokenResponse)
        assert result.token_type == "Bearer"
        assert result.access_token == "eyJhbGciOiJSUzI1NiIsImtpZCI6InNzby..."
        request = httpx_mock.get_request()
        assert request.method == "POST"
        assert request.url.path.endswith("/sso/token")

    @pytest.mark.asyncio
    async def test_list_connections_with_request_options(
        self, async_workos, httpx_mock
    ):
        httpx_mock.add_response(json={"data": [], "list_metadata": {}})
        await async_workos.sso.list_connections(
            request_options={"extra_headers": {"X-Custom": "value"}}
        )
        request = httpx_mock.get_request()
        assert request.headers["X-Custom"] == "value"

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

    @pytest.mark.asyncio
    async def test_list_connections_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.sso.list_connections()
        finally:
            await workos.close()

    @pytest.mark.asyncio
    async def test_list_connections_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.sso.list_connections()
        finally:
            await workos.close()

    @pytest.mark.asyncio
    async def test_list_connections_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.sso.list_connections()
        finally:
            await workos.close()

    @pytest.mark.asyncio
    async def test_list_connections_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.sso.list_connections()
        finally:
            await workos.close()

    @pytest.mark.asyncio
    async def test_list_connections_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.sso.list_connections()
        finally:
            await workos.close()
