diff --git a/iga/__init__.py b/iga/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..380fb6f421159812100e0371b468707c88a52d0c --- /dev/null +++ b/iga/__init__.py @@ -0,0 +1,8 @@ +from .uib import UibSebra +from .uio import UioCerebrum + + +def get_iga_client(instance, config): + if instance == "uib": + return UibSebra(config) + return UioCerebrum(config) diff --git a/iga/iga.py b/iga/iga.py new file mode 100644 index 0000000000000000000000000000000000000000..cd43d80f85bd29c0cfcb02c40b1175cd2a51c5cb --- /dev/null +++ b/iga/iga.py @@ -0,0 +1,35 @@ +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Optional + +from greg.models import Identity + + +@dataclass +class IgaPerson: + """ + Simple Person representation + + Names are not required in sebra, thus we cannot assume everyone has them. + """ + + first: Optional[str] + last: Optional[str] + + def dict(self): + return { + "first": self.first, + "last": self.last, + } + + +class IgaImplementation(ABC): + @abstractmethod + def extid_search( + self, id_type: Identity.IdentityType, extid: str + ) -> Optional[IgaPerson]: + pass + + +class IgaException(Exception): + pass diff --git a/iga/tests/__init__.py b/iga/tests/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/iga/tests/test_uib.py b/iga/tests/test_uib.py new file mode 100644 index 0000000000000000000000000000000000000000..74d09edabe093b6feee54c785112135cc7a28994 --- /dev/null +++ b/iga/tests/test_uib.py @@ -0,0 +1,71 @@ +from greg.models import Identity +from ..iga import IgaPerson +from ..uib import UibSebra + + +UIB_USERS = { + "Resources": [ + { + "id": "1", + } + ] +} +UIB_SEARCH = { + "schemas": [""], + "id": 1, + "meta": { + "created": "2020-11-04T07:35:18Z", + "lastModified": "2020-11-04T07:35:18Z", + "resourceType": "User", + }, + "username": "", + "active": True, + "no:edu:scim:user": { + "accountType": "primary", + "employeeNumber": "102160", + "eduPersonPrincipalName": "ruped001@uib.no", + "userPrincipalName": "Ruth.Pedersen@uibtest.no", + }, +} + + +def test_uib_search(requests_mock): + """Regular search works as expected""" + requests_mock.get( + "http://example.com/sebra/Users?norEduPersonNIN=1", + json=UIB_USERS, + ) + uib_search_wname = UIB_SEARCH.copy() + uib_search_wname["name"] = { + "formatted": "Ruth Pedersen", + "givenName": "Ruth", + "familyName": "Pedersen", + } + requests_mock.get( + "http://example.com/sebra/Users/1", + json=uib_search_wname, + ) + + client = UibSebra({"url": "http://example.com/sebra/", "headers": {"bar": "baz"}}) + assert not client.extid_search(Identity.IdentityType.PASSPORT_NUMBER, 1) + assert client.extid_search( + Identity.IdentityType.NORWEGIAN_NATIONAL_ID_NUMBER, 1 + ) == IgaPerson(first="Ruth", last="Pedersen") + + +def test_uib_search_nameless(requests_mock): + """Nameless people can be also be found""" + requests_mock.get( + "http://example.com/sebra/Users?norEduPersonNIN=1", + json=UIB_USERS, + ) + requests_mock.get( + "http://example.com/sebra/Users/1", + json=UIB_SEARCH, + ) + + client = UibSebra({"url": "http://example.com/sebra/", "headers": {"bar": "baz"}}) + assert not client.extid_search(Identity.IdentityType.PASSPORT_NUMBER, 1) + assert client.extid_search( + Identity.IdentityType.NORWEGIAN_NATIONAL_ID_NUMBER, 1 + ) == IgaPerson(first=None, last=None) diff --git a/iga/tests/test_uio.py b/iga/tests/test_uio.py new file mode 100644 index 0000000000000000000000000000000000000000..24cd55d7be905c5799b394b7723383485d33037a --- /dev/null +++ b/iga/tests/test_uio.py @@ -0,0 +1,80 @@ +from greg.models import Identity +from ..iga import IgaPerson +from ..uio import UioCerebrum + + +def test_uio_search_hit(requests_mock): + requests_mock.get( + "http://example.com/cerebrum/v1/search/persons/external-ids?id_type=NO_BIRTHNO&external_id=123", + json={ + "external_ids": [ + { + "person_id": 1, + "source_system": "dfo_sap", + "external_id": "123", + "id_type": "NO_BIRTHNO", + } + ] + }, + ) + requests_mock.get( + "http://example.com/cerebrum/v1/search/persons/external-ids?id_type=PASSNR&external_id=12345", + json={ + "external_ids": [ + { + "person_id": 1, + "source_system": "dfo_sap", + "external_id": "12345", + "id_type": "PASSNR", + } + ] + }, + ) + requests_mock.get( + "http://example.com/cerebrum/v1/persons/1", + json={ + "contexts": ["string"], + "created_at": "2022-02-17T10:17:31.305Z", + "href": "http://example.com/cerebrum/v1/search/persons/1", + "names": [ + {"source_system": "dfo_sap", "variant": "FIRST", "name": "Ola"}, + {"source_system": "dfo_sap", "variant": "LAST", "name": "Nordmann"}, + ], + "birth_date": "2022-02-17T10:17:31.305Z", + "id": 1, + }, + ) + + client = UioCerebrum( + {"url": "http://example.com/cerebrum/", "headers": {"bar": "baz"}} + ) + resp = client.extid_search(Identity.IdentityType.PASSPORT_NUMBER, "12345") + assert resp == IgaPerson(first="Ola", last="Nordmann") + + resp = client.extid_search( + Identity.IdentityType.NORWEGIAN_NATIONAL_ID_NUMBER, "123" + ) + assert resp == IgaPerson(first="Ola", last="Nordmann") + + +def test_uio_search_miss(requests_mock): + """Verify no matches returns empty list""" + requests_mock.get( + "http://example.com/cerebrum/v1/search/persons/external-ids?id_type=NO_BIRTHNO&external_id=123", + json={"external_ids": []}, + ) + requests_mock.get( + "http://example.com/cerebrum/v1/search/persons/external-ids?id_type=PASSNR&external_id=12345", + json={"external_ids": []}, + ) + + client = UioCerebrum( + {"url": "http://example.com/cerebrum/", "headers": {"bar": "baz"}} + ) + resp = client.extid_search(Identity.IdentityType.PASSPORT_NUMBER, "12345") + assert not resp + + resp = client.extid_search( + Identity.IdentityType.NORWEGIAN_NATIONAL_ID_NUMBER, "123" + ) + assert not resp diff --git a/iga/uib.py b/iga/uib.py new file mode 100644 index 0000000000000000000000000000000000000000..ec22d712c96befec512cc8dc6ea3a80fb179cfb6 --- /dev/null +++ b/iga/uib.py @@ -0,0 +1,30 @@ +from typing import Optional +from scim_client import ScimClient +from scim_client.models import User + +from greg.models import Identity + +from .iga import IgaImplementation, IgaPerson + + +class UibSebra(IgaImplementation): + def __init__(self, config) -> None: + self.client = ScimClient(**config) + self.idtype_methodmap = { + Identity.IdentityType.NORWEGIAN_NATIONAL_ID_NUMBER: self.client.get_by_nin, + Identity.IdentityType.PASSPORT_NUMBER: None, # not supported by uib scim + } + + def extid_search( + self, id_type: Identity.IdentityType, extid: str + ) -> Optional[IgaPerson]: + search = self.idtype_methodmap.get(id_type) + if not search: + return [] + user: User = search(extid) + if not user: + return None + return IgaPerson( + first=user.name.given_name if user.name else None, + last=user.name.family_name if user.name else None, + ) diff --git a/iga/uio.py b/iga/uio.py new file mode 100644 index 0000000000000000000000000000000000000000..93abc939d04c4a6a01489f2ea23fd9bf27d97406 --- /dev/null +++ b/iga/uio.py @@ -0,0 +1,44 @@ +from typing import Optional +import structlog + +from cerebrum_client import CerebrumClient + +from greg.models import Identity +from .iga import IgaException, IgaImplementation, IgaPerson + +logger = structlog.getLogger(__name__) + + +class UioCerebrum(IgaImplementation): + def __init__(self, config) -> None: + self.client = CerebrumClient(**config) + self.idtype2cerebrum = { + Identity.IdentityType.NORWEGIAN_NATIONAL_ID_NUMBER: "NO_BIRTHNO", + Identity.IdentityType.PASSPORT_NUMBER: "PASSNR", + } + + def extid_search( + self, id_type: Identity.IdentityType, extid: str + ) -> Optional[IgaPerson]: + idtype = self.idtype2cerebrum.get(id_type, None) + search = self.client.search_person_external_ids( + id_type=idtype, external_id=extid + ) + try: + persons = [self.client.get_person(i.person_id) for i in search] + except AttributeError as ae: + logger.exception("bad_cerebrum_response", external_id=extid) + raise IgaException("Failed to fetch IGA info") from ae + + if persons: + for pers in persons: + first = None + last = None + for pname in pers.names: + if pname.variant == "FIRST": + first = pname.name + elif pname.variant == "LAST": + last = pname.name + if first and last: + return IgaPerson(first=first, last=last) + return None