govoplan-idm/tests/test_scim.py

from __future__ import annotations

from datetime import UTC, datetime
import json

import pytest

from govoplan_core.security.http_fetch import HttpFetchResponse
from govoplan_idm.backend.scim import (
    SCIM_GROUP_SCHEMA,
    SCIM_LIST_SCHEMA,
    SCIM_USER_SCHEMA,
    ScimClient,
    ScimError,
    ScimLocalProjection,
    ScimProfile,
    ScimProvisioningPlanner,
    ScimSnapshot,
    parse_scim_list_response,
)


def _profile(*, absent_user_action="review") -> ScimProfile:
    return ScimProfile(
        provider_id="institutional-idp",
        base_url="https://idp.example.test/scim/v2",
        credential_ref="core-credential:scim",
        immutable_match_attribute="urn:example:params:scim:schemas:extension:staff:2.0:User:personnelNumber",
        absent_user_action=absent_user_action,
        page_size=2,
    )


def _user(resource_id: str, number: str, *, active: bool = True) -> dict[str, object]:
    return {
        "schemas": [
            SCIM_USER_SCHEMA,
            "urn:example:params:scim:schemas:extension:staff:2.0:User",
        ],
        "id": resource_id,
        "userName": f"user-{number}",
        "active": active,
        "urn:example:params:scim:schemas:extension:staff:2.0:User:personnelNumber": number,
        "meta": {"version": f'W/"{resource_id}"'},
    }


def _list(resources: list[dict[str, object]], *, total: int, start: int) -> bytes:
    return json.dumps(
        {
            "schemas": [SCIM_LIST_SCHEMA],
            "totalResults": total,
            "startIndex": start,
            "itemsPerPage": len(resources),
            "Resources": resources,
        }
    ).encode()


def test_scim_client_reads_complete_one_based_paginated_snapshot() -> None:
    calls: list[str] = []

    def transport(url, *, method, headers, body):
        calls.append(url)
        assert method == "GET" and body is None
        assert "Authorization" not in headers
        if "/Users?" in url and "startIndex=1" in url:
            payload = _list([_user("u-1", "100"), _user("u-2", "200")], total=3, start=1)
        elif "/Users?" in url:
            payload = _list([_user("u-3", "300")], total=3, start=3)
        else:
            payload = _list([], total=0, start=1)
        return HttpFetchResponse(200, {"Content-Type": "application/scim+json"}, payload)

    snapshot = ScimClient(_profile(), transport=transport).fetch_snapshot()

    assert snapshot.complete is True
    assert [item.resource_id for item in snapshot.users] == ["u-1", "u-2", "u-3"]
    assert snapshot.groups == ()
    assert snapshot.page_count == 3
    assert len(calls) == 3


def test_planner_links_by_explicit_immutable_attribute_and_quarantines_collision() -> None:
    page = parse_scim_list_response(
        json.loads(_list([_user("u-1", "100"), _user("u-2", "200")], total=2, start=1)),
        resource_type="User",
    )
    snapshot = ScimSnapshot(
        provider_id="institutional-idp",
        observed_at=datetime(2026, 8, 23, tzinfo=UTC),
        users=page.resources,
        complete=True,
        page_count=1,
    )
    local = (
        ScimLocalProjection("local-100", "User", "100", revision=2),
        ScimLocalProjection("local-200-a", "User", "200", revision=1),
        ScimLocalProjection("local-200-b", "User", "200", revision=1),
    )

    plan = ScimProvisioningPlanner(_profile()).plan(snapshot, local)

    assert [(item.action, item.local_id) for item in plan.operations] == [
        ("link", "local-100"),
        ("quarantine", None),
    ]
    assert len(plan.plan_sha256) == 64


def test_absence_never_deactivates_from_incomplete_snapshot() -> None:
    local = (
        ScimLocalProjection(
            "local-100",
            "User",
            "100",
            revision=3,
            provider_resource_id="u-1",
        ),
    )
    incomplete = ScimSnapshot(
        provider_id="institutional-idp",
        observed_at=datetime(2026, 8, 23, tzinfo=UTC),
        complete=False,
    )
    complete = ScimSnapshot(
        provider_id="institutional-idp",
        observed_at=datetime(2026, 8, 23, tzinfo=UTC),
        complete=True,
    )

    first = ScimProvisioningPlanner(_profile(absent_user_action="deactivate")).plan(incomplete, local)
    second = ScimProvisioningPlanner(_profile(absent_user_action="deactivate")).plan(complete, local)

    assert first.operations == ()
    assert first.warnings
    assert second.operations[0].action == "deactivate"
    assert second.operations[0].expected_local_revision == 3


def test_groups_are_business_projections_and_not_access_grants() -> None:
    group = {
        "schemas": [SCIM_GROUP_SCHEMA],
        "id": "g-1",
        "displayName": "Payroll reviewers",
        "externalId": "group-100",
        "members": [{"value": "u-1"}],
    }
    profile = ScimProfile(
        provider_id="institutional-idp",
        base_url="https://idp.example.test/scim/v2",
        credential_ref="core-credential:scim",
        immutable_match_attribute="externalIdImmutable",
    )
    group["externalIdImmutable"] = "group-stable-100"
    page = parse_scim_list_response(
        {
            "schemas": [SCIM_LIST_SCHEMA],
            "totalResults": 1,
            "startIndex": 1,
            "itemsPerPage": 1,
            "Resources": [group],
        },
        resource_type="Group",
    )
    plan = ScimProvisioningPlanner(profile).plan(
        ScimSnapshot(
            provider_id="institutional-idp",
            observed_at=datetime(2026, 8, 23, tzinfo=UTC),
            groups=page.resources,
            complete=True,
        ),
        (),
    )

    assert plan.operations[0].resource_type == "Group"
    assert plan.operations[0].action == "create"
    assert profile.group_projection_mode == "business_membership_only"


def test_parser_and_profile_reject_unsafe_identity_assumptions() -> None:
    with pytest.raises(ValueError, match="immutable"):
        ScimProfile(
            provider_id="idp",
            base_url="https://idp.example.test/scim/v2",
            credential_ref="credential",
            immutable_match_attribute="userName",
        )
    with pytest.raises(ScimError, match="ListResponse"):
        parse_scim_list_response({"Resources": []}, resource_type="User")