|
| 1 | +import datetime |
| 2 | +import enum |
| 3 | +import uuid |
| 4 | + |
| 5 | +import pytest |
| 6 | +from scim2_models import InvalidValueException |
| 7 | +from scim2_models import SearchRequest |
| 8 | +from scim2_models import User |
| 9 | +from sqlalchemy import Enum |
| 10 | +from sqlalchemy import ForeignKey |
| 11 | +from sqlalchemy import create_engine |
| 12 | +from sqlalchemy.orm import DeclarativeBase |
| 13 | +from sqlalchemy.orm import Mapped |
| 14 | +from sqlalchemy.orm import mapped_column |
| 15 | +from sqlalchemy.orm import relationship |
| 16 | + |
| 17 | +from scim2_sqlalchemy import Many |
| 18 | +from scim2_sqlalchemy import ResourceMapping |
| 19 | + |
| 20 | + |
| 21 | +class Color(enum.Enum): |
| 22 | + Red = "Rouge" |
| 23 | + blue = "bleu" |
| 24 | + Green = "Vert" |
| 25 | + |
| 26 | + |
| 27 | +class Base(DeclarativeBase): |
| 28 | + pass |
| 29 | + |
| 30 | + |
| 31 | +class ColoredUser(Base): |
| 32 | + """A user whose title column stores the names of the members, and whose external identifier stores their values.""" |
| 33 | + |
| 34 | + __tablename__ = "colored_users" |
| 35 | + |
| 36 | + id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4) |
| 37 | + user_name: Mapped[str] = mapped_column(unique=True) |
| 38 | + title: Mapped[Color | None] = mapped_column(Enum(Color, name="title_color")) |
| 39 | + external_id: Mapped[Color | None] = mapped_column( |
| 40 | + Enum( |
| 41 | + Color, |
| 42 | + name="external_color", |
| 43 | + values_callable=lambda members: [member.value for member in members], |
| 44 | + ) |
| 45 | + ) |
| 46 | + created: Mapped[datetime.datetime] |
| 47 | + last_modified: Mapped[datetime.datetime] |
| 48 | + version: Mapped[int] = mapped_column() |
| 49 | + |
| 50 | + emails: Mapped[list["ColoredEmail"]] = relationship( |
| 51 | + cascade="all, delete-orphan", order_by="ColoredEmail.id" |
| 52 | + ) |
| 53 | + |
| 54 | + __mapper_args__ = {"version_id_col": version} |
| 55 | + |
| 56 | + |
| 57 | +class ColoredEmail(Base): |
| 58 | + __tablename__ = "colored_emails" |
| 59 | + |
| 60 | + id: Mapped[int] = mapped_column(primary_key=True) |
| 61 | + user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("colored_users.id")) |
| 62 | + value: Mapped[str] |
| 63 | + type: Mapped[Color | None] = mapped_column(Enum(Color, name="email_color")) |
| 64 | + |
| 65 | + |
| 66 | +COLORED_USERS = ResourceMapping( |
| 67 | + User, |
| 68 | + ColoredUser, |
| 69 | + { |
| 70 | + "id": ColoredUser.id, |
| 71 | + "userName": ColoredUser.user_name, |
| 72 | + "title": ColoredUser.title, |
| 73 | + "externalId": ColoredUser.external_id, |
| 74 | + "emails": Many( |
| 75 | + ColoredUser.emails, |
| 76 | + {"value": ColoredEmail.value, "type": ColoredEmail.type}, |
| 77 | + ), |
| 78 | + "meta.created": ColoredUser.created, |
| 79 | + "meta.lastModified": ColoredUser.last_modified, |
| 80 | + }, |
| 81 | + version=ColoredUser.version, |
| 82 | +) |
| 83 | + |
| 84 | + |
| 85 | +@pytest.fixture |
| 86 | +def storage(database_url, storage_factory, user_type): |
| 87 | + """Return a storage of users with enumerated columns: a with Red, b with blue, c with Green, d without.""" |
| 88 | + engine = create_engine(database_url) |
| 89 | + Base.metadata.drop_all(engine) |
| 90 | + Base.metadata.create_all(engine) |
| 91 | + engine.dispose() |
| 92 | + storage = storage_factory(mappings={"User": COLORED_USERS}) |
| 93 | + for user_name, title, external_id in [ |
| 94 | + ("a", "Red", "Rouge"), |
| 95 | + ("b", "blue", "bleu"), |
| 96 | + ("c", "Green", "Vert"), |
| 97 | + ("d", None, None), |
| 98 | + ]: |
| 99 | + emails = [{"value": f"{user_name}@example.org", "type": title}] |
| 100 | + storage.create( |
| 101 | + user_type, |
| 102 | + User( |
| 103 | + user_name=user_name, |
| 104 | + title=title, |
| 105 | + external_id=external_id, |
| 106 | + emails=emails, |
| 107 | + ), |
| 108 | + ) |
| 109 | + return storage |
| 110 | + |
| 111 | + |
| 112 | +def user_names(storage, user_type, **parameters): |
| 113 | + _, resources = storage.search([user_type], SearchRequest(**parameters)) |
| 114 | + return [resource.user_name for resource in resources] |
| 115 | + |
| 116 | + |
| 117 | +def test_an_enumerated_column_reads_back_the_string_it_stores(storage, user_type): |
| 118 | + """A resource holds the string the column stores: the name of the member, or its value with values_callable.""" |
| 119 | + _, (user,) = storage.search([user_type], SearchRequest(filter='userName eq "a"')) |
| 120 | + |
| 121 | + assert user.title == "Red" |
| 122 | + assert user.external_id == "Rouge" |
| 123 | + assert user.emails[0].type == "Red" |
| 124 | + |
| 125 | + |
| 126 | +@pytest.mark.parametrize( |
| 127 | + ("scim_filter", "expected"), |
| 128 | + [ |
| 129 | + ('title eq "blue"', ["b"]), |
| 130 | + ('title eq "BLUE"', ["b"]), |
| 131 | + ('title eq "bleu"', []), |
| 132 | + ('title gt "h"', ["a"]), |
| 133 | + ('title co "e"', ["a", "b", "c"]), |
| 134 | + ("title pr", ["a", "b", "c"]), |
| 135 | + ('externalId eq "bleu"', ["b"]), |
| 136 | + ('externalId eq "Bleu"', []), |
| 137 | + ('externalId gt "S"', ["b", "c"]), |
| 138 | + ('emails[type eq "green"]', ["c"]), |
| 139 | + ("emails.type pr", ["a", "b", "c"]), |
| 140 | + ], |
| 141 | +) |
| 142 | +def test_an_enumerated_column_is_filtered_on_the_string_it_stores( |
| 143 | + storage, user_type, scim_filter, expected |
| 144 | +): |
| 145 | + """A filter compares the string the resource holds, as for any string column.""" |
| 146 | + assert sorted(user_names(storage, user_type, filter=scim_filter)) == expected |
| 147 | + |
| 148 | + |
| 149 | +@pytest.mark.parametrize( |
| 150 | + ("sort_by", "sort_order", "expected"), |
| 151 | + [ |
| 152 | + ("title", "ascending", ["b", "c", "a", "d"]), |
| 153 | + ("title", "descending", ["d", "a", "c", "b"]), |
| 154 | + ("externalId", "ascending", ["a", "c", "b", "d"]), |
| 155 | + ], |
| 156 | +) |
| 157 | +def test_an_enumerated_column_sorts_on_the_string_it_stores( |
| 158 | + storage, user_type, sort_by, sort_order, expected |
| 159 | +): |
| 160 | + """The order is the one of the strings, and not the order the members are declared in.""" |
| 161 | + assert ( |
| 162 | + user_names(storage, user_type, sort_by=sort_by, sort_order=sort_order) |
| 163 | + == expected |
| 164 | + ) |
| 165 | + |
| 166 | + |
| 167 | +@pytest.mark.parametrize( |
| 168 | + "user", |
| 169 | + [ |
| 170 | + User(user_name="e", title="Purple"), |
| 171 | + User(user_name="e", title="Rouge"), |
| 172 | + User(user_name="e", external_id="Red"), |
| 173 | + User(user_name="e", emails=[{"value": "e@example.org", "type": "Purple"}]), |
| 174 | + ], |
| 175 | +) |
| 176 | +def test_an_enumerated_column_refuses_a_string_it_does_not_store( |
| 177 | + storage, user_type, user |
| 178 | +): |
| 179 | + """A string outside the enumeration is refused with a 400, rather than failing in the database.""" |
| 180 | + with pytest.raises(InvalidValueException): |
| 181 | + storage.create(user_type, user) |
0 commit comments