287 lines
11 KiB
Python
287 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from urllib.parse import parse_qs, urlsplit
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_core.security.http_fetch import HttpFetchResponse
|
|
from govoplan_connectors.backend.service_desk_transport import (
|
|
HttpServiceDeskTransport,
|
|
ServiceDeskTransportError,
|
|
)
|
|
from govoplan_connectors.backend.service_desk_schemas import ServiceDeskRouteMapping
|
|
|
|
|
|
def response(payload: dict[str, object], *, headers: dict[str, str] | None = None):
|
|
return HttpFetchResponse(
|
|
status=200,
|
|
headers={"Content-Type": "application/json", **(headers or {})},
|
|
body=json.dumps(payload).encode(),
|
|
)
|
|
|
|
|
|
class ServiceDeskTransportTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.transport = HttpServiceDeskTransport()
|
|
self.endpoint = "https://support.example.test/znuny/nph-genericinterface.pl/Webservice/GovOPlaN"
|
|
self.routes = {
|
|
"search_path": "/Ticket/Search",
|
|
"ticket_path": "/Ticket/{ticket_id}",
|
|
"search_method": "POST",
|
|
"ticket_method": "GET",
|
|
}
|
|
|
|
def test_governed_routes_reject_embedded_authentication_controls(self) -> None:
|
|
for field, value in (
|
|
("search_path", "/Ticket/Search?Password=secret"),
|
|
(
|
|
"ticket_web_url_template",
|
|
"https://desk.example.test/ticket/{ticket_id}?SessionID=secret",
|
|
),
|
|
):
|
|
with self.subTest(field=field), self.assertRaisesRegex(
|
|
ValueError, "authentication controls"
|
|
):
|
|
ServiceDeskRouteMapping(**{field: value})
|
|
|
|
@patch("govoplan_connectors.backend.service_desk_transport.fetch_http")
|
|
def test_discovery_uses_governed_route_header_auth_and_version(self, fetch) -> None:
|
|
fetch.return_value = response(
|
|
{"TicketID": []}, headers={"X-Znuny-Version": "7.1.4"}
|
|
)
|
|
discovery = self.transport.discover(
|
|
endpoint_url=self.endpoint,
|
|
credential={"user_login": "connector", "password": "secret"},
|
|
routes={**self.routes, "search_filters": {"QueueIDs": [3, 7]}},
|
|
)
|
|
call = fetch.call_args
|
|
self.assertEqual("POST", call.kwargs["method"])
|
|
self.assertTrue(call.args[0].endswith("/Ticket/Search"))
|
|
self.assertEqual("connector", call.kwargs["headers"]["X-OTRS-Header-UserLogin"])
|
|
self.assertIn(
|
|
"X-OTRS-Header-Password",
|
|
call.kwargs["redirect_sensitive_headers"],
|
|
)
|
|
self.assertNotIn("secret", call.args[0])
|
|
self.assertEqual([3, 7], json.loads(call.kwargs["body"])["QueueIDs"])
|
|
self.assertEqual("znuny", discovery["product"])
|
|
self.assertEqual("synchronize", discovery["maturity"])
|
|
|
|
@patch("govoplan_connectors.backend.service_desk_transport.fetch_http")
|
|
def test_get_ticket_flags_are_query_parameters_without_secrets(self, fetch) -> None:
|
|
fetch.side_effect = (
|
|
response({"TicketID": ["42"]}),
|
|
response({"Ticket": [{"TicketID": "42", "Changed": "2026-08-22T10:00:00Z"}]}),
|
|
)
|
|
batch = self.transport.changes(
|
|
endpoint_url=self.endpoint,
|
|
credential={"user_login": "connector", "password": "secret"},
|
|
routes=self.routes,
|
|
cursor=None,
|
|
limit=100,
|
|
force_full=True,
|
|
)
|
|
query = parse_qs(urlsplit(fetch.call_args_list[1].args[0]).query)
|
|
self.assertEqual(["1"], query["AllArticles"])
|
|
self.assertEqual(["0"], query["GetAttachmentContents"])
|
|
self.assertNotIn("UserLogin", query)
|
|
self.assertEqual(1, len(batch.changes))
|
|
|
|
@patch("govoplan_connectors.backend.service_desk_transport.fetch_http")
|
|
def test_link_identity_reads_do_not_request_articles_attachments_or_dynamic_fields(
|
|
self, fetch
|
|
) -> None:
|
|
fetch.side_effect = (
|
|
response({"TicketID": ["42"]}),
|
|
response(
|
|
{"Ticket": [{"TicketID": "42", "Changed": "2026-08-22T10:00:00Z"}]}
|
|
),
|
|
)
|
|
self.transport.changes(
|
|
endpoint_url=self.endpoint,
|
|
credential=None,
|
|
routes={**self.routes, "_identity_only": True},
|
|
cursor=None,
|
|
limit=100,
|
|
force_full=True,
|
|
)
|
|
query = parse_qs(urlsplit(fetch.call_args_list[1].args[0]).query)
|
|
self.assertEqual(["0"], query["AllArticles"])
|
|
self.assertEqual(["0"], query["Attachments"])
|
|
self.assertEqual(["0"], query["DynamicFields"])
|
|
|
|
@patch("govoplan_connectors.backend.service_desk_transport.fetch_http")
|
|
def test_paged_full_cursor_preserves_cumulative_high_watermark(self, fetch) -> None:
|
|
fetch.side_effect = (
|
|
response({"TicketID": ["41", "42"]}),
|
|
response({"Ticket": [{"TicketID": "41", "Changed": "2026-08-22T12:00:00Z"}]}),
|
|
response({"TicketID": ["41", "42"]}),
|
|
response({"Ticket": [{"TicketID": "42", "Changed": "2026-08-22T10:00:00Z"}]}),
|
|
)
|
|
first = self.transport.changes(
|
|
endpoint_url=self.endpoint,
|
|
credential=None,
|
|
routes=self.routes,
|
|
cursor=None,
|
|
limit=1,
|
|
force_full=True,
|
|
)
|
|
second = self.transport.changes(
|
|
endpoint_url=self.endpoint,
|
|
credential=None,
|
|
routes=self.routes,
|
|
cursor=first.next_cursor,
|
|
limit=1,
|
|
force_full=True,
|
|
)
|
|
self.assertFalse(first.complete)
|
|
self.assertTrue(second.complete)
|
|
self.assertEqual("2026-08-22T12:00:00Z", second.high_watermark)
|
|
|
|
@patch("govoplan_connectors.backend.service_desk_transport.fetch_http")
|
|
def test_delta_overlaps_and_pages_all_ids_at_one_timestamp(self, fetch) -> None:
|
|
shared_revision = "2026-08-22T10:00:00Z"
|
|
fetch.side_effect = (
|
|
response({"TicketID": ["41", "42"]}),
|
|
response({"Ticket": [{"TicketID": "41", "Changed": shared_revision}]}),
|
|
response({"Ticket": [{"TicketID": "42", "Changed": shared_revision}]}),
|
|
response({"TicketID": ["41", "42"]}),
|
|
response({"Ticket": [{"TicketID": "41", "Changed": shared_revision}]}),
|
|
response({"Ticket": [{"TicketID": "42", "Changed": shared_revision}]}),
|
|
)
|
|
first = self.transport.changes(
|
|
endpoint_url=self.endpoint,
|
|
credential=None,
|
|
routes=self.routes,
|
|
cursor=json.dumps(
|
|
{"kind": "delta", "changed": "2026-08-22T09:59:59Z", "seen": []}
|
|
),
|
|
limit=1,
|
|
force_full=False,
|
|
)
|
|
second = self.transport.changes(
|
|
endpoint_url=self.endpoint,
|
|
credential=None,
|
|
routes=self.routes,
|
|
cursor=first.next_cursor,
|
|
limit=1,
|
|
force_full=False,
|
|
)
|
|
self.assertEqual(["41"], [item["TicketID"] for item in first.changes])
|
|
self.assertEqual(["42"], [item["TicketID"] for item in second.changes])
|
|
second_search = json.loads(fetch.call_args_list[3].kwargs["body"])
|
|
self.assertEqual(
|
|
"2026-08-22T09:59:59Z",
|
|
second_search["TicketChangeTimeNewerDate"],
|
|
)
|
|
|
|
@patch("govoplan_connectors.backend.service_desk_transport.fetch_http")
|
|
def test_delta_does_not_suppress_a_seen_ticket_that_changed_again(self, fetch) -> None:
|
|
fetch.side_effect = (
|
|
response({"TicketID": ["41"]}),
|
|
response(
|
|
{
|
|
"Ticket": [
|
|
{"TicketID": "41", "Changed": "2026-08-22T10:05:00Z"}
|
|
]
|
|
}
|
|
),
|
|
)
|
|
batch = self.transport.changes(
|
|
endpoint_url=self.endpoint,
|
|
credential=None,
|
|
routes=self.routes,
|
|
cursor=json.dumps(
|
|
{
|
|
"kind": "delta",
|
|
"changed": "2026-08-22T10:00:00Z",
|
|
"seen": ["41"],
|
|
}
|
|
),
|
|
limit=10,
|
|
force_full=False,
|
|
)
|
|
self.assertEqual(["41"], [item["TicketID"] for item in batch.changes])
|
|
self.assertEqual("2026-08-22T10:05:00Z", batch.high_watermark)
|
|
|
|
@patch("govoplan_connectors.backend.service_desk_transport.fetch_http")
|
|
def test_body_authentication_is_never_put_in_get_url(self, fetch) -> None:
|
|
fetch.return_value = response({"TicketID": ["42"]})
|
|
with self.assertRaisesRegex(ServiceDeskTransportError, "cannot be used with a GET"):
|
|
self.transport.changes(
|
|
endpoint_url=self.endpoint,
|
|
credential={
|
|
"auth_mode": "body",
|
|
"user_login": "connector",
|
|
"password": "secret",
|
|
},
|
|
routes={**self.routes, "search_method": "GET"},
|
|
cursor=None,
|
|
limit=1,
|
|
force_full=True,
|
|
)
|
|
fetch.assert_not_called()
|
|
|
|
@patch("govoplan_connectors.backend.service_desk_transport.fetch_http")
|
|
def test_legacy_body_authentication_is_post_only_and_not_duplicated(self, fetch) -> None:
|
|
fetch.return_value = response({"TicketID": []})
|
|
self.transport.discover(
|
|
endpoint_url=self.endpoint,
|
|
credential={
|
|
"auth_mode": "body",
|
|
"user_login": "connector",
|
|
"password": "secret",
|
|
},
|
|
routes=self.routes,
|
|
)
|
|
call = fetch.call_args
|
|
payload = json.loads(call.kwargs["body"])
|
|
self.assertEqual("connector", payload["UserLogin"])
|
|
self.assertEqual("secret", payload["Password"])
|
|
self.assertNotIn("X-OTRS-Header-UserLogin", call.kwargs["headers"])
|
|
self.assertNotIn("secret", call.args[0])
|
|
|
|
@patch("govoplan_connectors.backend.service_desk_transport.fetch_http")
|
|
def test_update_requires_revision_and_requested_field_verification(self, fetch) -> None:
|
|
fetch.side_effect = (
|
|
response(
|
|
{
|
|
"Ticket": [
|
|
{
|
|
"TicketID": "42",
|
|
"State": "open",
|
|
"Changed": "2026-08-22T10:00:00Z",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
response({"Success": 1}),
|
|
response(
|
|
{
|
|
"Ticket": [
|
|
{
|
|
"TicketID": "42",
|
|
"State": "open",
|
|
"Changed": "2026-08-22T10:05:00Z",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
)
|
|
with self.assertRaises(ServiceDeskTransportError) as raised:
|
|
self.transport.update_ticket(
|
|
endpoint_url=self.endpoint,
|
|
credential=None,
|
|
routes={**self.routes, "update_path": "/Ticket/{ticket_id}"},
|
|
ticket_id="42",
|
|
expected_revision="2026-08-22T10:00:00Z",
|
|
changes={"State": "pending reminder"},
|
|
)
|
|
self.assertTrue(raised.exception.outcome_unknown)
|
|
self.assertEqual("update_verification_failed", raised.exception.code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|