Commit ba564661 authored by Trond Aasan's avatar Trond Aasan
Browse files

CIM-58 Fix client

* Fixes CIM-59, CIM-60 and CIM-61

* Partial object updates are only partially supported.  Change logic
  to always serialise whole object, but exclude empty phone numbers.

* Remove unused code

* Validate models on assignment

* Simplify types.  Remove constr fields, mypy don't understand them and
  CIM validates data anyway.

* Use mixins to support optional functionality
parent a3aaf817
......@@ -57,7 +57,7 @@ Python client for accessing the CIM-API.
```python
from cim_client import CimClient, CimEndpoints
from cim_client.models import Person, PersonList
from cim_client.models import PersonBase, PersonList
c = CimClient(
CimEndpoints(url='https://example.com',
update_person_url='/_webservices/?ws=contacts/upsert/1.0',
......@@ -67,13 +67,13 @@ c = CimClient(
upd_schema = c.get_update_person_schema()
del_schema = c.get_delete_person_schema()
person = Person.from_dict({'username': 'foo', 'user_import_id': 'foo1'})
person = PersonBase.from_dict({'username': 'foo', 'firstname': 'John', 'lastname': 'Doe'})
response1 = c.update_person(person)
person2 = Person.from_dict({'username': 'bar', 'user_import_id': 'bar1'})
person2 = PersonBase.from_dict({'username': 'bar', 'firstname': 'Petter', 'lastname': 'Smart'})
persons = [person, person2]
personlist = PersonList(persons=persons)
# Note that delete_person supports both PersonList and [Person, Person, ...]
response2 = c.delete_person(personlist)
response3 = c.delete_person(persons)
response2 = c.update_person(personlist)
response3 = c.delete_person(person.username)
```
......@@ -10,8 +10,9 @@ from cim_client.models import (
Organisation,
OrganisationImportResult,
OrganisationList,
Person,
PersonBase,
PersonList,
PersonBase_T,
)
logger = logging.getLogger(__name__)
......@@ -20,6 +21,30 @@ logger = logging.getLogger(__name__)
JsonType = Any
def _serialise_upsert_person(
person: PersonBase_T,
) -> Dict[str, Any]:
"""
Make sure empty phone numbers are excluded.
A phone field is reset when the corresponding key is omitted.
Neither "" nor null are allowed values.
"""
empty_phone_fields = {
f
for f in {
"job_mobile",
"job_phone",
"secret_number",
"private_mobile",
"private_phone",
}
if not person.__dict__[f]
}
return person.dict(exclude=empty_phone_fields or None) # type: ignore[arg-type]
def merge_dicts(*dicts: Optional[Dict[Any, Any]]) -> Dict[Any, Any]:
"""
Combine a series of dicts without mutating any of them.
......@@ -203,11 +228,9 @@ class CimClient(object):
def post_person(
self,
persondata: Union[List[Person], Person, PersonList],
delete: bool = False,
overwrite: bool = True,
persondata: Union[List[PersonBase], PersonBase, PersonList],
) -> Optional[Tuple[str, bytes]]:
"""Update or delete one or more persons
"""Update one or more persons
Note that updating more than one person at a time may require parsing
the content of the response.
......@@ -216,26 +239,18 @@ class CimClient(object):
Note also that the response will not contain info about failing to find
someone. You will have to compare with what you posted.
"""
if delete:
url = self.endpoints.delete_person()
else:
url = self.endpoints.update_person()
url = self.endpoints.update_person()
if type(persondata) is list:
persondata = PersonList(persons=persondata)
elif type(persondata) is Person:
pass
else:
if isinstance(persondata, PersonList):
persondata = persondata.persons
elif isinstance(persondata, PersonBase):
persondata = [persondata]
elif not isinstance(persondata, list):
raise TypeError("persondata must be List of Persons or Person")
if overwrite:
data = persondata.json(by_alias=True) # type: ignore
else:
data = persondata.json(by_alias=True, exclude_unset=True) # type: ignore
headers = {"Content-Type": "application/json", "accept": "application/json"}
response = self.post(url, data=data, auth=self.auth, headers=headers)
if response.status_code == 404:
return None
data = [_serialise_upsert_person(x) for x in persondata]
headers = {"Content-Type": "application/json"}
response = self.post(url, json=data, auth=self.auth, headers=headers)
if response.status_code == 200:
logger.debug("import success, response was: %s", response.content)
return "Import success", response.content
......@@ -244,38 +259,48 @@ class CimClient(object):
def delete_person(
self,
persons: Union[List[Person], Person, PersonList],
overwrite: bool = True,
person_ids: List[str],
) -> Union[Tuple[str, bytes], None]:
"""Convenience method for deletion
It is unclear at the time of writing what the deletion endpoint needs
other than the username or user_import_id (if activated) of the person
we want to the delete. By default it seems to accept the same fields as
the update endpoint.
The request body should contain an array of unique identifiers
for the users or contacts you want to delete. You can find out
which property is used as a unique identifier in the installation
by generating a JSON schema.
:param persons: The person or people we want to delete from CIM
:param overwrite: In the case of overwrite it is logical that it serves
no purpose for this method but it has been included in case of any
shenanigans on the cim side.
:param person_ids: The person or people we want to delete from CIM
:return: String describing status or None if not found
"""
return self.post_person(persons, delete=True, overwrite=overwrite)
url = self.endpoints.delete_person()
headers = {"Content-Type": "application/json"}
response = self.post(url, json=person_ids, auth=self.auth, headers=headers)
if (
response.status_code == 404
and response.text == "No person found: No deletion"
):
logger.warning(
"No persons were deleted, response was: %s", response.content
)
return None
if response.status_code == 200:
logger.debug("Delete success, response was: %s", response.content)
return "Delete success", response.content
response.raise_for_status()
return None
def update_person(
self,
persons: Union[List[Person], Person, PersonList],
overwrite: bool = True,
persons: Union[List[PersonBase], PersonBase, PersonList],
) -> Union[Tuple[str, bytes], None]:
"""Convenience method for updating
:param persons: The person or people we want to update in CIM
:param overwrite: sets all unset fields to None causing the removal of
any information not included in the update for the user being
updated.
:return: String describing status or None if not found
"""
return self.post_person(persons, delete=False, overwrite=overwrite)
return self.post_person(persons)
def get_import_organisations_schema(self) -> JsonType:
"""GETs the current schema from CIM
......@@ -344,7 +369,7 @@ class CimClient(object):
url = self.endpoints.import_organisations()
result: JsonType = self.post(
url,
json=organisations.dict(exclude_unset=True)["__root__"],
json=organisations.dict()["__root__"],
auth=self.auth,
return_response=False,
)
......@@ -361,11 +386,11 @@ class CimClient(object):
def _find_unknown_parents(organisations: List[Organisation]) -> Set[str]:
keys = set(map(lambda x: x.key, organisations))
parent_keys = set(
filter(lambda x: x is not None, map(lambda x: x.parent_key, organisations))
filter(lambda x: bool(x), map(lambda x: x.parent_key, organisations))
)
if not parent_keys.issubset(keys):
return parent_keys - keys # type: ignore
return parent_keys - keys
return set()
......
......@@ -2,29 +2,17 @@
import json
from enum import Enum
from typing import TypeVar, Optional, List, Union, Type, Dict, Any
from typing import TypeVar, Optional, List, Type, Dict, Any
import pydantic.networks
from pydantic.class_validators import validator
from pydantic.types import constr
from pydantic import Extra
T = TypeVar("T")
BaseModel_T = TypeVar("BaseModel_T", bound="BaseModel")
PersonBase_T = TypeVar("PersonBase_T", bound="PersonBase")
NameType = TypeVar("NameType")
# TODO: Swap this regex with a validator that calls the parser in the
# phonenumbers library if we run into problems
# Optional '+' followed by one or more numbers (from cim documentation)
PHONE_REGEX: Type[str] = constr(regex=r"^\+?[ 0-9]+$")
class EmailEmptyAllowedStr(pydantic.networks.EmailStr):
@classmethod
def validate(cls, value: str) -> str:
if value == "":
return value
return super().validate(value)
class BaseModel(pydantic.BaseModel):
"""Expanded BaseModel for convenience"""
......@@ -32,7 +20,7 @@ class BaseModel(pydantic.BaseModel):
@classmethod
def from_dict(cls: Type[BaseModel_T], data: Dict[str, Any]) -> BaseModel_T:
"""Initialize class from dict"""
return cls(**data) # noqa
return cls(**data) # type: ignore[call-arg]
@classmethod
def from_json(cls: Type[BaseModel_T], json_data: str) -> BaseModel_T:
......@@ -40,6 +28,10 @@ class BaseModel(pydantic.BaseModel):
data = json.loads(json_data)
return cls.from_dict(data)
class Config:
extra = Extra.forbid
validate_assignment = True
class PersonTypeEnum(str, Enum):
user = "user"
......@@ -62,80 +54,96 @@ class NextOfKin(BaseModel):
firstname: str
lastname: str
id: str
phone: Optional[PHONE_REGEX] # type: ignore
phone: Optional[str]
email: Optional[str]
relation: Optional[RelationEnum]
date_of_birth: Optional[str] # Date format must agree with system admin
address: Optional[str]
state: Optional[str]
postal_code: Optional[str]
town: Optional[str]
country_code: Optional[str] # ISO 3166 two-letter country code (alpha-2)
class Person(BaseModel):
relation: RelationEnum = RelationEnum.unknown
# Date format must agree with system admin
date_of_birth: str = ""
address: str = ""
state: str = ""
postal_code: str = ""
town: str = ""
# ISO 3166 two-letter country code (alpha-2)
country_code: str = ""
@pydantic.validator(
"date_of_birth",
"address",
"state",
"postal_code",
"town",
"country_code",
pre=True,
)
def validate_parent_key(cls, value: Optional[str]) -> str: # noqa:[N805]
return "" if value is None else value
class NextOfKinMixin(BaseModel):
next_of_kin: List[NextOfKin] = []
class OrgImportIdMixin(BaseModel):
org_import_id: str = ""
class PersonBase(BaseModel):
"""Main model containing all fields for a CIM Person"""
username: str # Unique identifier if not user_import_id
lastname: Optional[str]
firstname: Optional[str]
person_type: Optional[PersonTypeEnum]
user_import_id: Optional[str] # Unique identifier if activated
password: Optional[str]
job_mobile: Optional[PHONE_REGEX] # type: ignore
job_phone: Optional[PHONE_REGEX] # type: ignore
secret_number: Optional[PHONE_REGEX] # type: ignore
private_mobile: Optional[PHONE_REGEX] # type: ignore
private_phone: Optional[PHONE_REGEX] # type: ignore
# Unique identifier if not user_import_id is activated
username: str
lastname: str
firstname: str
person_type: PersonTypeEnum = PersonTypeEnum.contact
job_mobile: str = ""
job_phone: str = ""
secret_number: str = ""
private_mobile: str = ""
private_phone: str = ""
email: Optional[str]
email_secondary: Optional[str]
org_import_id: Optional[str]
location_import_id: Optional[str]
company: Optional[str]
department: Optional[str]
subdep1: Optional[str]
subdep2: Optional[str]
subdep3: Optional[str]
subdep4: Optional[str]
subdep5: Optional[str]
subdep6: Optional[str]
subdep7: Optional[str]
subdep8: Optional[str]
subdep9: Optional[str]
subdep10: Optional[str]
job_title: Optional[str]
employee_id: Optional[str]
description: Optional[str]
roles: Optional[str] # Comma separated list ('foo' or 'foo,bar,baz')
dist_list: Optional[str] # List of dist_lists, same as above
job_title: str = ""
employee_id: str = ""
description: str = ""
# Comma separated list ('foo' or 'foo,bar,baz')
roles: str = ""
# List of dist_lists, same as above
dist_list: Optional[str]
# This is actually an enum of allowed values, see schema
timezone: Optional[str]
next_of_kin: Union[List[NextOfKin], None]
@validator("org_import_id")
def check_company_or_org_import_id(
cls, org_import_id: Optional[str], values: List[str] # noqa: N805
) -> Optional[str]:
if "company" not in values and not org_import_id:
raise ValueError("Either company or org_import_id must be supplied")
return org_import_id
@pydantic.validator(
"job_mobile",
"job_phone",
"secret_number",
"private_mobile",
"private_phone",
"job_title",
"employee_id",
"description",
"roles",
pre=True,
)
def validate_parent_key(cls, value: Optional[str]) -> str: # noqa:[N805]
return "" if value is None else value
class PersonList(BaseModel):
"""Model for making json formatted list of persons"""
persons: List[Person]
persons: List[PersonBase]
class Organisation(BaseModel):
"""Cim organisation"""
name: constr(max_length=150, min_length=1) # type: ignore
# FIXME: Looks like CIM API treats keys as case insensitive, but it's not documented.
key: constr(max_length=100, min_length=1) # type: ignore
parent_key: Optional[constr(max_length=100)] # type: ignore
name: str
key: str
parent_key: str = ""
class Config:
extra = "forbid"
@pydantic.validator("parent_key", pre=True)
def validate_parent_key(cls, value: Optional[str]) -> str: # noqa:[N805]
return "" if value is None else value
class OrganisationList(BaseModel):
......@@ -143,6 +151,8 @@ class OrganisationList(BaseModel):
class OrganisationImportResult(BaseModel):
"""API result"""
insert: Optional[List[str]]
update: Optional[List[str]]
delete: Optional[List[str]]
{
"username": "janedoe",
"user_import_id": "janedoe1"
"firstname": "Jane",
"lastname": "Doe"
}
......@@ -2,7 +2,6 @@
"lastname": "Doe",
"firstname": "John",
"username": "johndoe",
"user_import_id": "john123",
"job_mobile": "+4712345678",
"job_phone": "+4787654321",
"email": "johndoe@company.no",
......
......@@ -6,9 +6,18 @@ from cim_client.client import (
CimClient,
CimEndpoints,
logger as client_logger,
_find_unknown_parents, # noqa
_find_unknown_parents,
_serialise_upsert_person,
)
from cim_client.models import Person
from cim_client.models import (
PersonBase,
OrgImportIdMixin,
NextOfKinMixin,
)
class Person(PersonBase, OrgImportIdMixin, NextOfKinMixin):
pass
@pytest.fixture
......@@ -85,6 +94,39 @@ def test_get_delete_person_schema(client, requests_mock):
assert response == {"foo": "bar"}
def test_delete_person_success_200(client, john_doe, requests_mock):
"""Ensure 404 returns None"""
requests_mock.post(
"https://localhost/_webservices/?ws=contacts/delete/1.0",
status_code=200,
json=["Delete person: johndoe(9999)"],
)
response = client.delete_person(Person.from_dict(john_doe).username)
assert response == ("Delete success", b'["Delete person: johndoe(9999)"]')
def test_delete_person_success_404(client, john_doe, requests_mock):
"""Ensure 404 returns None when response text is `No person found: No deletion`"""
requests_mock.post(
"https://localhost/_webservices/?ws=contacts/delete/1.0",
status_code=404,
text="No person found: No deletion",
)
response = client.delete_person(Person.from_dict(john_doe).username)
assert response is None
def test_delete_person_failure_404(client, john_doe, requests_mock):
"""Ensure 404 fails when response text is not `No person found: No deletion`"""
requests_mock.post(
"https://localhost/_webservices/?ws=contacts/delete/1.0",
status_code=404,
text='["Something, something, something dark side"]',
)
with pytest.raises(HTTPError):
client.delete_person(Person.from_dict(john_doe).username)
def test_update_one_person(client, john_doe, requests_mock):
"""Ensure updating one works"""
requests_mock.post("https://localhost/_webservices/?ws=contacts/upsert/1.0")
......@@ -110,15 +152,6 @@ def test_update_failure_400(client, john_doe, requests_mock):
client.update_person(Person.from_dict(john_doe))
def test_update_failure_404(client, john_doe, requests_mock):
"""Ensure 404 returns None"""
requests_mock.post(
"https://localhost/_webservices/?ws=contacts/upsert/1.0", status_code=404
)
response = client.update_person(Person.from_dict(john_doe))
assert response is None
def test_update_failure_500(client, john_doe, requests_mock):
"""Ensure 500 raises an error"""
requests_mock.post(
......@@ -172,3 +205,14 @@ def test_import_organisations_missing_parent(
def test__find_unknown_parents(organisations_missing_parent):
unknown = _find_unknown_parents(organisations_missing_parent)
assert unknown == {organisations_missing_parent[1].parent_key}
def test__serialise_upsert_person():
p = Person(firstname="Ola", lastname="Dunk", username="donkey")
s = _serialise_upsert_person(p)
keys = s.keys()
assert "job_mobile" not in keys
assert "job_phone" not in keys
assert "secret_number" not in keys
assert "private_mobile" not in keys
assert "private_phone" not in keys
import pytest
from pydantic import ValidationError
from cim_client.models import Person, Organisation
from cim_client.models import (
PersonBase,
OrgImportIdMixin,
NextOfKinMixin,
Organisation,
)
class Person(PersonBase, OrgImportIdMixin, NextOfKinMixin):
pass
def test_person(john_doe, jane_doe, fai_lure):
......@@ -16,21 +25,13 @@ def test_person(john_doe, jane_doe, fai_lure):
def test_organisation():
with pytest.raises(ValidationError):
Organisation.from_dict({"name": "", "key": "key"})
with pytest.raises(ValidationError):
Organisation.from_dict({"name": "a" * 151, "key": "key"})
with pytest.raises(ValidationError):
Organisation.from_dict({"name": "name", "key": ""})
with pytest.raises(ValidationError):
Organisation.from_dict({"name": "name", "key": "a" * 101})
a_dict = {"name": "name", "key": "key"}
Organisation.from_dict(a_dict)
b_dict = {"name": "name", "key": "key", "parent_key": "parent_key"}
Organisation.from_dict(b_dict)
with pytest.raises(ValidationError):
Organisation.from_dict({"name": "name", "key": "key", "parent_key": "a" * 101})
def test_organisation_fails_with_extra_field():
with pytest.raises(ValidationError):
Organisation.from_dict(
{
......@@ -40,7 +41,3 @@ def test_organisation():
"extra field": "x",
}
)
Organisation.from_dict({"name": "name", "key": "key"})
Organisation.from_dict({"name": "name", "key": "key", "parent_key": "parent_key"})
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment