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()