(self)
| 342 | assert res.json["results"][0]["id"] == person.id |
| 343 | |
| 344 | def test_mappings_update(self): |
| 345 | table = self.create_table_entity(collection=self.col, entity_id="foo") |
| 346 | mapping = self.create_mapping(collection=self.col, table=table, role=self.role) |
| 347 | |
| 348 | url = f"/api/2/collections/{self.col.id}/mappings/{mapping.id}" |
| 349 | trigger_url = f"/api/2/collections/{self.col.id}/mappings/{mapping.id}/trigger" |
| 350 | |
| 351 | # Trigger mapping to generate entities |
| 352 | res = self.client.post(trigger_url, headers=self.headers) |
| 353 | assert res.status_code == 202 |
| 354 | |
| 355 | entities_url = "/api/2/entities" |
| 356 | entities_query_string = { |
| 357 | "filter:schema": "Person", |
| 358 | "filter:collection_id": self.col.id, |
| 359 | } |
| 360 | |
| 361 | res = self.client.get( |
| 362 | entities_url, |
| 363 | query_string=entities_query_string, |
| 364 | headers=self.headers, |
| 365 | ) |
| 366 | assert len(res.json["results"]) == 14 |
| 367 | assert set(res.json["results"][0]["properties"].keys()) == { |
| 368 | "name", |
| 369 | "nationality", |
| 370 | "proof", |
| 371 | } |
| 372 | |
| 373 | new_query = { |
| 374 | "person": { |
| 375 | "schema": "Person", |
| 376 | "keys": ["name", "nationality"], |
| 377 | "properties": { |
| 378 | "name": {"column": "name"}, |
| 379 | "nationality": {"column": "nationality"}, |
| 380 | # The following column mapping is new |
| 381 | "gender": {"column": "gender"}, |
| 382 | }, |
| 383 | } |
| 384 | } |
| 385 | data = {"table_id": table.id, "mapping_query": new_query} |
| 386 | |
| 387 | # Mapping and collection IDs have to match |
| 388 | url = f"/api/2/collections/{self.col_other.id}/mappings/{mapping.id}" |
| 389 | res = self.client.post(url, json=data, headers=self.headers_other) |
| 390 | assert res.status_code == 404 |
| 391 | |
| 392 | # Update the mapping |
| 393 | url = f"/api/2/collections/{self.col.id}/mappings/{mapping.id}" |
| 394 | |
| 395 | # Only users with access to the collection can update mappings |
| 396 | res = self.client.post(url, json=data, headers=self.headers_other) |
| 397 | assert res.status_code == 403 |
| 398 | |
| 399 | # Read access is not enough to update a mapping |
| 400 | res = self.client.post(url, json=data, headers=self.headers_read) |
| 401 | assert res.status_code == 403 |
nothing calls this directly
no test coverage detected