tlg
86bd7f6a07
Every opportunity created by the workflow now carries the CTag "Auto: Durch Suchagent-Treffer erstellt" (id 6a83292a186dce7b2), set via cTagsIds directly in the create POST. The tag id is resolved at runtime by name (EspoClient.ensure_ctag, same pattern as team_id; self-healing after a CRM rebuild); the read-back verification now also checks the tag. No backfill: existing opportunities and the saved "Via cowork-api erstellt" list filter are untouched. Live-verified end to end: synthetic trigger mail, opportunity 6a832b8f29cc86bdf created with cTagsNames containing the tag (test artifact removed afterwards). 65/65 pytest. Also adds the complete German technical documentation of the workflow (docs/projekt-matching-dokumentation.md, sections 1-13 + S68 appendix). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
84 lines
3.2 KiB
Python
84 lines
3.2 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_ensure_ctag_find_and_create():
|
|
hits = {"list": [{"id": "t1", "name": "Auto: Durch Suchagent-Treffer erstellt"}]}
|
|
client = make_client([resp(body=hits)])
|
|
assert client.ensure_ctag("Auto: Durch Suchagent-Treffer erstellt") == "t1"
|
|
params = client.session.request.call_args.kwargs["params"]
|
|
assert params["where[0][type]"] == "equals"
|
|
assert params["where[0][attribute]"] == "name"
|
|
client = make_client([resp(body={"list": []}), resp(body={"id": "t2"})])
|
|
assert client.ensure_ctag("Auto: Durch Suchagent-Treffer erstellt") == "t2"
|
|
payload = client.session.request.call_args.kwargs["json"]
|
|
assert payload == {"name": "Auto: Durch Suchagent-Treffer erstellt"}
|
|
|
|
|
|
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")
|