71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from projektmatch import espocrm
|
|
|
|
|
|
def make_client(responses):
|
|
"""Client whose session returns queued mock responses."""
|
|
client = espocrm.EspoClient("https://crm.example/api/v1", "k", retries=1,
|
|
backoff=0)
|
|
client.session = mock.Mock()
|
|
client.session.request.side_effect = responses
|
|
return client
|
|
|
|
|
|
def resp(status=200, body=None, reason=""):
|
|
r = mock.Mock()
|
|
r.status_code = status
|
|
r.json.return_value = body if body is not None else {}
|
|
r.text = "x" if body is not None else ""
|
|
r.headers = {"X-Status-Reason": reason}
|
|
return r
|
|
|
|
|
|
def test_find_opportunity_by_link():
|
|
client = make_client([resp(body={"list": [{"id": "1", "name": "P"}]})])
|
|
assert client.find_opportunity_by_link("https://x")["id"] == "1"
|
|
params = client.session.request.call_args.kwargs["params"]
|
|
assert params["where[0][type]"] == "equals"
|
|
assert params["where[0][attribute]"] == "cProjektlink"
|
|
|
|
|
|
def test_team_id_missing_raises():
|
|
client = make_client([resp(body={"list": []})])
|
|
with pytest.raises(espocrm.EspoError, match="DesTEngS"):
|
|
client.team_id("DesTEngS")
|
|
|
|
|
|
def test_ensure_account_exact_match_and_create():
|
|
hits = {"list": [{"id": "a1", "name": "Aristo Group", "type": "Reseller"}]}
|
|
client = make_client([resp(body=hits)])
|
|
assert client.ensure_account("Aristo Group", "Reseller") == "a1"
|
|
client = make_client([resp(body={"list": []}), resp(body={"id": "a2"})])
|
|
assert client.ensure_account("Neue GmbH", "Customer") == "a2"
|
|
payload = client.session.request.call_args.kwargs["json"]
|
|
assert payload == {"name": "Neue GmbH", "type": "Customer"}
|
|
|
|
|
|
def test_unique_opportunity_name_suffix():
|
|
hits = {"list": [{"name": "Projekt X"}, {"name": "Projekt X (2)"}]}
|
|
client = make_client([resp(body=hits)])
|
|
assert client.unique_opportunity_name("Projekt X") == "Projekt X (3)"
|
|
|
|
|
|
def test_retry_on_500_then_success():
|
|
client = make_client([resp(status=500), resp(body={"list": []})])
|
|
assert client.search("Team", "equals", "name", "X") == []
|
|
|
|
|
|
def test_400_raises_with_reason():
|
|
client = make_client([resp(status=400, body={}, reason="bad field")])
|
|
with pytest.raises(espocrm.EspoError, match="bad field"):
|
|
client.search("Team", "equals", "name", "X")
|
|
|
|
|
|
def test_helpers():
|
|
assert espocrm.core_token("Aristo Group GmbH") == "Aristo"
|
|
assert espocrm.split_person("Max Muster Mann") == ("Max Muster", "Mann")
|
|
assert espocrm.split_person("Mann") == ("", "Mann")
|