Coverage for tests / unit / rdflibplus / test_datasetview.py: 100%

1"""Test DatasetView.""" 

2 

3# ruff: noqa: D103, PLR2004 

4import pytest 

5from rdflib import Dataset, Graph, Literal, Namespace, URIRef 

6 

7from pythinfer.rdflibplus import DatasetView 

8 

9EX = Namespace("http://example.org/") 

10 

11 

12@pytest.fixture 

13def g0() -> Graph: 

14 """Create a named graph with zero triples (for immutability tests).""" 

15 return Graph(identifier=EX.graph0) 

16 

17 

18@pytest.fixture 

19def g1() -> Graph: 

20 """Create a named graph with one triple.""" 

21 graph = Graph(identifier=EX.graph1) 

22 graph.add((EX.subject1, EX.predicate1, Literal("object1"))) 

23 return graph 

24 

25 

26@pytest.fixture 

27def g2() -> Graph: 

28 """Create a named graph with three triples.""" 

29 graph = Graph(identifier=EX.graph2) 

30 graph.add((EX.subject2, EX.predicate2, Literal("object2"))) 

31 graph.add((EX.subject3, EX.predicate3, Literal("object3"))) 

32 graph.add((EX.subject4, EX.predicate4, Literal("object4"))) 

33 return graph 

34 

35 

36@pytest.fixture 

37def g3() -> Graph: 

38 """Create a named graph with five triples.""" 

39 graph = Graph(identifier=EX.graph3) 

40 graph.add((EX.subject5, EX.predicate5, Literal("object5"))) 

41 graph.add((EX.subject6, EX.predicate6, Literal("object6"))) 

42 graph.add((EX.subject7, EX.predicate7, Literal("object7"))) 

43 graph.add((EX.subject8, EX.predicate8, Literal("object8"))) 

44 graph.add((EX.subject9, EX.predicate9, Literal("object9"))) 

45 return graph 

46 

47 

48@pytest.fixture 

49def ds(g1: Graph, g2: Graph, g3: Graph) -> Dataset: 

50 """Create a Dataset containing all three graphs.""" 

51 dataset = Dataset(default_union=True) 

52 dataset.graph(g1) 

53 dataset.graph(g2) 

54 dataset.graph(g3) 

55 return dataset 

56 

57 

58def test_starting_assumptions(g1: Graph, g2: Graph, g3: Graph, ds: Dataset) -> None: 

59 # Starting assumption: separate graphs with different sizes and with 

60 # different stores 

61 assert len(g1) == 1 

62 assert len(g2) == 3 

63 assert len(g3) == 5 

64 assert g1.store != g2.store != g3.store 

65 

66 # A single Dataset contains all graphs 

67 assert len(ds.graph(g1.identifier)) == 1 

68 assert len(ds.graph(g2.identifier)) == 3 

69 assert len(ds.graph(g3.identifier)) == 5 

70 assert len(ds) == 9 

71 

72 

73def test_dataset_behaviour_is_insufficient( 

74 g1: Graph, 

75 g3: Graph, 

76 ds: Dataset, 

77) -> None: 

78 # Try to use a new Dataset as a subset view (and show it doesn't work) 

79 ds2 = Dataset(store=ds.store) 

80 ds2.add_graph(g1.identifier) 

81 ds2.add_graph(g3.identifier) 

82 

83 # First, check that the explicitly added graphs are present: 

84 assert len(ds2.graph(g1.identifier)) == 1 

85 assert len(ds2.graph(g3.identifier)) == 5 

86 

87 # Now, show that it is *not a subset view*: all graphs are there. 

88 # This would be 6 if Dataset only contained the two graphs added above 

89 assert len(ds2) == 9 

90 

91 

92def test_datasetview_basic_usage(g1: Graph, g2: Graph, g3: Graph, ds: Dataset) -> None: 

93 ds_view = DatasetView( 

94 original_ds=ds, 

95 included_graph_ids=[ 

96 g1.identifier, 

97 g3.identifier, 

98 ], 

99 ) 

100 

101 assert len(ds_view.graph(g1.identifier)) == 1 

102 assert len(ds_view.graph(g3.identifier)) == 5 

103 assert len(ds_view) == 6 

104 

105 # Check that excluded graph is indeed excluded 

106 with pytest.raises(PermissionError): 

107 ds_view.graph(g2.identifier) 

108 

109 # Also check that works when providing full Graph object 

110 assert len(ds_view.graph(g1)) == 1 

111 assert len(ds_view.graph(g3)) == 5 

112 with pytest.raises(PermissionError): 

113 ds_view.graph(g2) 

114 

115 

116def test_datasetview_different_selection( 

117 g1: Graph, 

118 g2: Graph, 

119 g3: Graph, 

120 ds: Dataset, 

121) -> None: 

122 ds_view2 = DatasetView( 

123 original_ds=ds, 

124 included_graph_ids=[ 

125 g2.identifier, 

126 g3.identifier, 

127 ], 

128 ) 

129 

130 assert len(ds_view2.graph(g2.identifier)) == 3 

131 assert len(ds_view2.graph(g3.identifier)) == 5 

132 assert len(ds_view2) == 8 

133 

134 with pytest.raises(PermissionError): 

135 ds_view2.graph(g1.identifier) 

136 

137 

138def test_quads_method(g1: Graph, g2: Graph, g3: Graph, ds: Dataset) -> None: 

139 ds_view = DatasetView( 

140 original_ds=ds, 

141 included_graph_ids=[ 

142 g1.identifier, 

143 g3.identifier, 

144 ], 

145 ) 

146 quads = list(ds_view.quads((None, None, None, None))) 

147 expected_quads = list(ds.quads((None, None, None, None))) 

148 # Filter expected_quads to only those in g1 and g3 

149 expected_quads = [ 

150 quad for quad in expected_quads if quad[3] in {g1.identifier, g3.identifier} 

151 ] 

152 assert len(expected_quads) == 6 

153 assert len(quads) == len(expected_quads) 

154 assert set(quads) == set(expected_quads) 

155 

156 

157def test_triples_method(g1: Graph, g2: Graph, g3: Graph, ds: Dataset) -> None: 

158 ds_view = DatasetView( 

159 original_ds=ds, 

160 included_graph_ids=[ 

161 g1.identifier, 

162 g3.identifier, 

163 ], 

164 ) 

165 triples = list(ds_view.triples((None, None, None))) 

166 expected_triples = list(g1.triples((None, None, None))) + list( 

167 g3.triples((None, None, None)), 

168 ) 

169 

170 assert len(expected_triples) == 6 

171 assert len(triples) == len(expected_triples) 

172 assert set(triples) == set(expected_triples) 

173 

174 # Also test with explicit context as kwarg 

175 triples_ctx = list( 

176 ds_view.triples((None, None, None), context=g1), 

177 ) 

178 expected_triples_ctx = list(g1.triples((None, None, None))) 

179 assert len(triples_ctx) == len(expected_triples_ctx) 

180 assert set(triples_ctx) == set(expected_triples_ctx) 

181 

182 

183def test_contains_method(g1: Graph, g2: Graph, g3: Graph, ds: Dataset) -> None: 

184 ds_view = DatasetView( 

185 original_ds=ds, 

186 included_graph_ids=[ 

187 g1.identifier, 

188 g3.identifier, 

189 ], 

190 ) 

191 # Triples in g1 and g3 should be found 

192 assert (EX.subject1, EX.predicate1, Literal("object1")) in ds_view 

193 assert (EX.subject5, EX.predicate5, Literal("object5")) in ds_view 

194 

195 # Triples in g2 should not be found 

196 assert (EX.subject2, EX.predicate2, Literal("object2")) not in ds_view 

197 

198 

199def test_iterating_over_dataset(g1: Graph, g2: Graph, g3: Graph, ds: Dataset) -> None: 

200 ds_view = DatasetView( 

201 original_ds=ds, 

202 included_graph_ids=[ 

203 g1.identifier, 

204 g3.identifier, 

205 ], 

206 ) 

207 triples = set(ds_view) 

208 expected_triples = {(s, p, o, g1.identifier) for (s, p, o) in g1} | { 

209 (s, p, o, g3.identifier) for (s, p, o) in g3 

210 } 

211 

212 assert len(expected_triples) == 6 

213 assert len(triples) == len(expected_triples) 

214 assert triples == expected_triples 

215 

216 

217def test_graph_method(g0: Graph, g1: Graph, g2: Graph, ds: Dataset) -> None: 

218 ds_view = DatasetView( 

219 original_ds=ds, 

220 included_graph_ids=[ 

221 g0.identifier, 

222 g2.identifier, 

223 ], 

224 ) 

225 # Accessing included graphs should work 

226 graph0 = ds_view.graph(g0.identifier) 

227 assert isinstance(graph0, Graph) 

228 assert len(graph0) == 0 

229 assert graph0 == g0 

230 

231 graph2 = ds_view.graph(g2.identifier) 

232 assert isinstance(graph2, Graph) 

233 assert len(graph2) == 3 

234 assert graph2 == g2 

235 

236 # Accessing excluded graphs should raise an error 

237 with pytest.raises(PermissionError): 

238 ds_view.graph(g1.identifier) 

239 

240 # Accessing the default graph should raise an error because it is not 

241 # explicitly included 

242 with pytest.raises(PermissionError): 

243 ds_view.graph(ds.default_graph.identifier) 

244 

245 # Make sure that the deprecated add_graph method behaves the same way 

246 with pytest.raises(PermissionError): 

247 ds_view.add_graph(g1.identifier) 

248 

249 

250def test_remove_graph(g0: Graph, g1: Graph, g2: Graph, ds: Dataset) -> None: 

251 ds_view = DatasetView( 

252 original_ds=ds, 

253 included_graph_ids=[ 

254 g1.identifier, 

255 g2.identifier, 

256 ], 

257 ) 

258 n_orig_ds = len(ds) 

259 assert len(ds_view) == 4 

260 # Try to remove a graph from the view by identifier 

261 ds_view.remove_graph(g1.identifier) 

262 assert len(ds_view.graph(g1.identifier)) == 0 

263 assert len(ds_view) == 3 

264 

265 # Check that the original dataset is also affected 

266 assert len(ds.graph(g1.identifier)) == 0 

267 assert len(ds) == n_orig_ds - 1 

268 

269 # Try to remove a graph from the view by Graph object 

270 ds_view.remove_graph(g2) 

271 assert len(ds_view.graph(g2.identifier)) == 0 

272 assert len(ds_view) == 0 

273 assert len(ds.graph(g2.identifier)) == 0 

274 assert len(ds) == n_orig_ds - 4 

275 

276 # Now check that attempting to remove a graph not in the view raises an error 

277 with pytest.raises(PermissionError): 

278 ds_view.remove_graph(g0.identifier) 

279 

280 # Now check that None or default graph raises an error 

281 with pytest.raises(PermissionError): 

282 ds_view.remove_graph(None) 

283 with pytest.raises(PermissionError): 

284 ds_view.remove_graph(ds_view.default_graph) 

285 

286 

287def test_add_triple(g0: Graph, g1: Graph, g2: Graph, ds: Dataset) -> None: 

288 ds_view = DatasetView( 

289 original_ds=ds, 

290 included_graph_ids=[ 

291 g0.identifier, 

292 g2.identifier, 

293 ], 

294 ) 

295 assert len(ds_view.graph(g0.identifier)) == 0 

296 assert len(ds.graph(g0.identifier)) == 0 

297 

298 # Adding a triple to a graph included in the view should work like normal 

299 ds_view.graph(g0.identifier).add( 

300 (EX.subjectX, EX.predicateX, Literal("objectX")), 

301 ) 

302 assert len(ds_view.graph(g0.identifier)) == 1 

303 # Check that the original dataset reflects the change 

304 assert len(ds.graph(g0.identifier)) == 1 

305 

306 # Adding a triple to a graph not included in the view should fail 

307 with pytest.raises(PermissionError): 

308 ds_view.graph(g1.identifier).add( 

309 (EX.subjectY, EX.predicateY, Literal("objectY")), 

310 ) 

311 

312 # This holds for the default graph as well 

313 with pytest.raises(PermissionError): 

314 ds_view.graph(ds_view.default_graph.identifier).add( 

315 (EX.subjectY, EX.predicateY, Literal("objectY")), 

316 ) 

317 

318 # Adding directly to included graphs should work 

319 ds_view.add((EX.subjectY, EX.predicateY, Literal("objectY"), g0.identifier)) 

320 

321 # But adding directly to the View also fails for excluded graphs 

322 with pytest.raises(PermissionError): 

323 ds_view.add((EX.subjectY, EX.predicateY, Literal("objectY"), g1.identifier)) 

324 with pytest.raises(PermissionError): 

325 ds_view.add((EX.subjectY, EX.predicateY, Literal("objectY"))) 

326 

327 

328def test_remove_triple(g0: Graph, g1: Graph, g2: Graph, ds: Dataset) -> None: 

329 ds_view = DatasetView( 

330 original_ds=ds, 

331 included_graph_ids=[ 

332 g0.identifier, 

333 g2.identifier, 

334 ], 

335 ) 

336 # Removing a triple from a graph in the view should work like normal 

337 ds_view.graph(g2.identifier).remove( 

338 (EX.subject2, EX.predicate2, Literal("object2")), 

339 ) 

340 assert len(ds_view.graph(g2.identifier)) == 2 

341 # Check that the original dataset reflects the change 

342 assert len(ds.graph(g2.identifier)) == 2 

343 assert len(ds_view) == 2 

344 

345 # Removing a triple from a graph not included in the view should fail 

346 with pytest.raises(PermissionError): 

347 ds_view.graph(g1.identifier).remove( 

348 (EX.subject1, EX.predicate1, Literal("object1")), 

349 ) 

350 

351 # Also check that removing directly from the view fails for excluded graphs 

352 with pytest.raises(PermissionError): 

353 ds_view.remove( 

354 (EX.subject1, EX.predicate1, Literal("object1"), g1.identifier), 

355 ) 

356 # This holds for the default graph as well 

357 with pytest.raises(PermissionError): 

358 ds_view.remove( 

359 (EX.subject1, EX.predicate1, Literal("object1")), 

360 ) 

361 

362 # Now check removing directly from included graphs works 

363 assert len(ds_view.graph(g2.identifier)) == 2 

364 ds_view.remove( 

365 (EX.subject3, EX.predicate3, Literal("object3"), g2.identifier), 

366 ) 

367 assert len(ds_view.graph(g2.identifier)) == 1 

368 assert len(ds.graph(g2.identifier)) == 1 

369 assert len(ds_view) == 1 

370 

371 

372def test_datasetview_preserves_namespace_bindings( 

373 g1: Graph, 

374 g2: Graph, 

375 ds: Dataset, 

376) -> None: 

377 """Test that namespace bindings are preserved when serializing a DatasetView.""" 

378 # Bind custom namespaces to the dataset 

379 CUSTOM1 = Namespace("http://custom1.example.org/") 

380 CUSTOM2 = Namespace("http://custom2.example.org/") 

381 ds.bind("custom1", CUSTOM1) 

382 ds.bind("custom2", CUSTOM2) 

383 

384 # Add triples using these custom namespaces via the dataset's graphs 

385 # (not the original graph objects, which have separate stores) 

386 ds.graph(g1.identifier).add( 

387 (CUSTOM1.subject_custom1, CUSTOM1.predicate_custom1, Literal("value1")) 

388 ) 

389 ds.graph(g2.identifier).add( 

390 (CUSTOM2.subject_custom2, CUSTOM2.predicate_custom2, Literal("value2")) 

391 ) 

392 

393 # Create a DatasetView that excludes one graph (g3) 

394 ds_view = DatasetView( 

395 original_ds=ds, 

396 included_graph_ids=[g1.identifier, g2.identifier], 

397 ) 

398 

399 # Serialize the view 

400 serialized = str(ds_view.serialize(format="trig")) 

401 

402 # Check that namespace bindings are preserved in serialization 

403 assert "@prefix custom1: <http://custom1.example.org/>" in serialized 

404 assert "@prefix custom2: <http://custom2.example.org/>" in serialized