scim2-client/tests/test_discovery.py at d7a0de45cfcec108cb8b7e3891c93e00375ca60f · python-scim/scim2-client · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import threading
import wsgiref.simple_server
from typing import Annotated
from typing import Union

import portpicker
import pytest
from httpx import Client
from scim2_models import EnterpriseUser
from scim2_models import Extension
from scim2_models import Group
from scim2_models import Meta
from scim2_models import Required
from scim2_models import ResourceType
from scim2_models import User

from scim2_client.engines.httpx import SyncSCIMClient

scim2_server = pytest.importorskip("scim2_server")
from scim2_server.backend import InMemoryBackend  # noqa: E402
from scim2_server.provider import SCIMProvider  # noqa: E402


class OtherExtension(Extension):
    schemas: Annotated[list[str], Required.true] = [
        "urn:ietf:params:scim:schemas:extension:Other:1.0:User"
    ]

    test: str | None = None
    test2: list[str] | None = None


def get_schemas():
    schemas = [
        User.to_schema(),
        Group.to_schema(),
        OtherExtension.to_schema(),
        EnterpriseUser.to_schema(),
    ]

    # SCIMProvider register_schema requires meta object to be set
    for schema in schemas:
        schema.meta = Meta(resource_type="Schema")

    return schemas


def get_resource_types():
    resource_types = [
        ResourceType.from_resource(User[Union[EnterpriseUser, OtherExtension]]),
        ResourceType.from_resource(Group),
    ]

    # SCIMProvider register_resource_type requires meta object to be set
    for resource_type in resource_types:
        resource_type.meta = Meta(resource_type="ResourceType")

    return resource_types


@pytest.fixture(scope="session")
def server():
    backend = InMemoryBackend()
    provider = SCIMProvider(backend)
    for schema in get_schemas():
        provider.register_schema(schema)
    for resource_type in get_resource_types():
        provider.register_resource_type(resource_type)

    host = "localhost"
    port = portpicker.pick_unused_port()
    httpd = wsgiref.simple_server.make_server(host, port, provider)

    server_thread = threading.Thread(target=httpd.serve_forever)
    server_thread.start()
    try:
        yield host, port
    finally:
        httpd.shutdown()
        server_thread.join()


def test_discovery_resource_types_multiple_extensions(server):
    host, port = server
    client = Client(base_url=f"http://{host}:{port}")
    scim_client = SyncSCIMClient(client)

    scim_client.discover()
    assert scim_client.get_resource_model("User")
    assert scim_client.get_resource_model("Group")

    # Try to create a user to see if discover filled everything correctly
    user_request = User[Union[EnterpriseUser, OtherExtension]](
        user_name="bjensen@example.com"
    )
    scim_client.create(user_request)