Skip to content
Snippets Groups Projects
Verified Commit c746f2ce authored by Andreas Ellewsen's avatar Andreas Ellewsen
Browse files

Add iga module

Makes it possible to search for external ids (nin/passport) in iga
without refering directly to the respective clients.
parent 17a7efe8
No related branches found
No related tags found
1 merge request!286UiO: Sjekk om fødselsnummer/passnummer er i bruk i IGA
from .uib import UibSebra
from .uio import UioCerebrum
def get_iga_client(instance, config):
if instance == "uib":
return UibSebra(config)
return UioCerebrum(config)
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
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)
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
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,
)
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
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment