Coverage for tests / conftest.py: 94%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5"""Pytest configuration and fixtures for sparqlite tests.""" 

6 

7import subprocess 

8import tempfile 

9import time 

10from pathlib import Path 

11from urllib.error import URLError 

12from urllib.request import Request, urlopen 

13 

14import pytest 

15from virtuoso_utilities.launch_virtuoso import launch_virtuoso 

16 

17from sparqlite import SPARQLClient 

18 

19 

20TEST_GRAPH = "https://w3id.org/oc/meta/test" 

21 

22PREFIXES = """ 

23PREFIX fabio: <http://purl.org/spar/fabio/> 

24PREFIX dcterms: <http://purl.org/dc/terms/> 

25PREFIX prism: <http://prismstandard.org/namespaces/basic/2.0/> 

26PREFIX datacite: <http://purl.org/spar/datacite/> 

27PREFIX literal: <http://www.essepuntato.it/2010/06/literalreification/> 

28PREFIX pro: <http://purl.org/spar/pro/> 

29PREFIX foaf: <http://xmlns.com/foaf/0.1/> 

30""" 

31 

32OPENCITATIONS_TEST_DATA = f""" 

33{PREFIXES} 

34 

35INSERT DATA {{ 

36 GRAPH <{TEST_GRAPH}> {{ 

37 <https://w3id.org/oc/meta/br/1> a fabio:JournalArticle ; 

38 dcterms:title "A study on citation networks" ; 

39 prism:publicationDate "2024-01-15" ; 

40 datacite:hasIdentifier <https://w3id.org/oc/meta/id/1> ; 

41 pro:isDocumentContextFor <https://w3id.org/oc/meta/ar/1> . 

42 

43 <https://w3id.org/oc/meta/id/1> 

44 datacite:usesIdentifierScheme datacite:doi ; 

45 literal:hasLiteralValue "10.1000/test.001" . 

46 

47 <https://w3id.org/oc/meta/ar/1> 

48 pro:withRole pro:author ; 

49 pro:isHeldBy <https://w3id.org/oc/meta/ra/1> . 

50 

51 <https://w3id.org/oc/meta/ra/1> foaf:name "John Smith" . 

52 

53 <https://w3id.org/oc/meta/br/2> a fabio:JournalArticle ; 

54 dcterms:title "Machine learning in bibliometrics" ; 

55 prism:publicationDate "2024-03-20" ; 

56 datacite:hasIdentifier <https://w3id.org/oc/meta/id/2> ; 

57 pro:isDocumentContextFor <https://w3id.org/oc/meta/ar/2> . 

58 

59 <https://w3id.org/oc/meta/id/2> 

60 datacite:usesIdentifierScheme datacite:doi ; 

61 literal:hasLiteralValue "10.1000/test.002" . 

62 

63 <https://w3id.org/oc/meta/ar/2> 

64 pro:withRole pro:author ; 

65 pro:isHeldBy <https://w3id.org/oc/meta/ra/2> . 

66 

67 <https://w3id.org/oc/meta/ra/2> foaf:name "Jane Doe" . 

68 

69 <https://w3id.org/oc/meta/br/3> a fabio:Book ; 

70 dcterms:title "Introduction to Semantic Web" ; 

71 prism:publicationDate "2023-06-01" . 

72 }} 

73}} 

74""" 

75 

76 

77def _wait_for_endpoint(url: str, timeout: int = 60) -> None: 

78 start = time.time() 

79 while time.time() - start < timeout: 

80 try: 

81 req = Request( 

82 f"{url}?query=ASK+%7B+%3Fs+%3Fp+%3Fo+%7D", 

83 headers={"Accept": "application/sparql-results+json"}, 

84 ) 

85 urlopen(req, timeout=2) 

86 return 

87 except (URLError, OSError): 

88 time.sleep(1) 

89 raise RuntimeError(f"Endpoint {url} not ready within {timeout}s") 

90 

91 

92@pytest.fixture(scope="session") 

93def virtuoso_endpoint(): 

94 container_name = "sparqlite-test-virtuoso" 

95 http_port = 8895 

96 

97 with tempfile.TemporaryDirectory() as data_dir: 

98 launch_virtuoso( 

99 name=container_name, 

100 data_dir=data_dir, 

101 http_port=http_port, 

102 isql_port=11115, 

103 memory="2g", 

104 dba_password="dba", 

105 detach=True, 

106 wait_ready=True, 

107 enable_write_permissions=True, 

108 force_remove=True, 

109 ) 

110 

111 yield f"http://localhost:{http_port}/sparql" 

112 

113 subprocess.run( 

114 ["docker", "rm", "-f", container_name], 

115 stdout=subprocess.DEVNULL, 

116 stderr=subprocess.DEVNULL, 

117 ) 

118 

119 

120@pytest.fixture(scope="session") 

121def qlever_endpoint(): 

122 container_name = "sparqlite-test-qlever" 

123 http_port = 7019 

124 

125 with tempfile.TemporaryDirectory() as data_dir: 

126 input_file = Path(data_dir) / "input.nt" 

127 input_file.write_text( 

128 "<http://example.org/s> <http://example.org/p> <http://example.org/o> .\n" 

129 ) 

130 

131 subprocess.run( 

132 [ 

133 "docker", "run", "--rm", 

134 "--user", "root", 

135 "-v", f"{data_dir}:/data", 

136 "--entrypoint", "qlever-index", 

137 "adfreiburg/qlever", 

138 "-i", "/data/index", 

139 "-f", "/data/input.nt", 

140 "-F", "nt", 

141 ], 

142 check=True, 

143 capture_output=True, 

144 ) 

145 

146 subprocess.run( 

147 [ 

148 "docker", "run", "-d", 

149 "--name", container_name, 

150 "--user", "root", 

151 "-p", f"{http_port}:7001", 

152 "-v", f"{data_dir}:/data", 

153 "--entrypoint", "qlever-server", 

154 "adfreiburg/qlever", 

155 "-i", "/data/index", 

156 "-j", "2", 

157 "-p", "7001", 

158 "-m", "1G", 

159 "-n", 

160 ], 

161 check=True, 

162 capture_output=True, 

163 ) 

164 

165 _wait_for_endpoint(f"http://localhost:{http_port}") 

166 yield f"http://localhost:{http_port}" 

167 

168 subprocess.run( 

169 ["docker", "rm", "-f", container_name], 

170 stdout=subprocess.DEVNULL, 

171 stderr=subprocess.DEVNULL, 

172 ) 

173 

174 

175@pytest.fixture(params=["virtuoso", "qlever"], scope="session") 

176def endpoint(request): 

177 return request.getfixturevalue(f"{request.param}_endpoint") 

178 

179 

180@pytest.fixture 

181def client(endpoint): 

182 with SPARQLClient(endpoint) as c: 

183 yield c 

184 

185 

186@pytest.fixture 

187def test_data(client): 

188 client.update(OPENCITATIONS_TEST_DATA) 

189 yield 

190 client.update(f"CLEAR GRAPH <{TEST_GRAPH}>")