Coverage for tests / test_client.py: 100%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from unittest.mock import MagicMock, patch 

6 

7import pycurl 

8 

9from sparqlite import SPARQLClient 

10 

11 

12class TestEndpointQueryParameters: 

13 

14 def test_get_preserves_endpoint_query_params(self): 

15 mock_curl = MagicMock() 

16 mock_curl.getinfo.return_value = 200 

17 setopt_calls: list[tuple[int, str]] = [] 

18 

19 def track_setopt(opt: int, val: str) -> None: 

20 setopt_calls.append((opt, val)) 

21 

22 mock_curl.setopt = track_setopt 

23 

24 with patch("pycurl.Curl", return_value=mock_curl): 

25 client = SPARQLClient( 

26 "http://example.org/sparql?default-graph-uri=http%3A%2F%2Fexample.org%2Fgraph" 

27 ) 

28 client._curl = mock_curl 

29 client._request("SELECT * WHERE { ?s ?p ?o }", "application/json") 

30 client.close() 

31 

32 url_calls = [val for opt, val in setopt_calls if opt == pycurl.URL] 

33 assert len(url_calls) == 1 

34 assert "default-graph-uri=http%3A%2F%2Fexample.org%2Fgraph" in url_calls[0] 

35 assert "query=" in url_calls[0] 

36 

37 def test_post_preserves_endpoint_query_params(self): 

38 mock_curl = MagicMock() 

39 mock_curl.getinfo.return_value = 200 

40 setopt_calls: list[tuple[int, str]] = [] 

41 

42 def track_setopt(opt: int, val: str) -> None: 

43 setopt_calls.append((opt, val)) 

44 

45 mock_curl.setopt = track_setopt 

46 

47 with patch("pycurl.Curl", return_value=mock_curl): 

48 client = SPARQLClient( 

49 "http://example.org/sparql?default-graph-uri=http%3A%2F%2Fexample.org%2Fgraph" 

50 ) 

51 client._curl = mock_curl 

52 client._request( 

53 "SELECT * WHERE { ?s ?p ?o }", "application/json", method="POST" 

54 ) 

55 client.close() 

56 

57 post_calls = [val for opt, val in setopt_calls if opt == pycurl.POSTFIELDS] 

58 assert len(post_calls) == 1 

59 assert "default-graph-uri=http%3A%2F%2Fexample.org%2Fgraph" in post_calls[0] 

60 assert "query=" in post_calls[0] 

61 

62 url_calls = [val for opt, val in setopt_calls if opt == pycurl.URL] 

63 assert url_calls == ["http://example.org/sparql"] 

64 

65 def test_endpoint_without_query_params_unchanged(self): 

66 mock_curl = MagicMock() 

67 mock_curl.getinfo.return_value = 200 

68 setopt_calls: list[tuple[int, str]] = [] 

69 

70 def track_setopt(opt: int, val: str) -> None: 

71 setopt_calls.append((opt, val)) 

72 

73 mock_curl.setopt = track_setopt 

74 

75 with patch("pycurl.Curl", return_value=mock_curl): 

76 client = SPARQLClient("http://example.org/sparql") 

77 client._curl = mock_curl 

78 client._request("SELECT * WHERE { ?s ?p ?o }", "application/json") 

79 client.close() 

80 

81 url_calls = [val for opt, val in setopt_calls if opt == pycurl.URL] 

82 assert len(url_calls) == 1 

83 assert url_calls[0].startswith("http://example.org/sparql?query=")