Method test_mappings_trigger

aleph/tests/test_mappings_api.py:221–273

(self)

        assert res.json["results"][0]["id"] == mapping_id

    def test_mappings_trigger(self):
        table = self.create_table_entity(self.col, entity_id="foo")
        mapping = self.create_mapping(self.col, table=table, role=self.role)

        # There are no entities before running the mapping
        entities_url = "/api/2/entities"
        entities_query_string = {
            "filter:collection_id": self.col.id,
            "filter:schemata": "Person",
        }

        res = self.client.get(
            entities_url,
            query_string=entities_query_string,
            headers=self.headers,
        )
        assert res.json["total"] == 0

        # Mapping and collection IDs have to match
        trigger_url = (
            f"/api/2/collections/{self.col_other.id}/mappings/{mapping.id}/trigger"
        )
        res = self.client.post(trigger_url, headers=self.headers_other)
        assert res.status_code == 404

        # Trigger the mapping to generate entities
        trigger_url = f"/api/2/collections/{self.col.id}/mappings/{mapping.id}/trigger"

        # Only users with access to the collection can trigger mappings
        res = self.client.post(trigger_url, headers=self.headers_other)
        assert res.status_code == 403

        # Read access is not enough to trigger a mapping
        res = self.client.post(trigger_url, headers=self.headers_read)
        assert res.status_code == 403

        res = self.client.post(trigger_url, headers=self.headers)
        assert res.status_code == 202

        view_url = f"/api/2/collections/{self.col.id}/mappings/{mapping.id}"
        res = self.client.get(view_url, headers=self.headers)
        assert res.json["last_run_status"] == "successful"
        assert "last_run_err_msg" not in res.json

        # 14 entities have been generated after triggering the mapping
        res = self.client.get(
            entities_url,
            query_string=entities_query_string,
            headers=self.headers,
        )
        assert len(res.json["results"]) == 14
        assert res.json["results"][0]["schema"] == "Person"
        assert res.json["results"][0]["properties"]["proof"][0]["id"] == table.id

    def test_mappings_flush(self):
        table = self.create_table_entity(collection=self.col, entity_id="foo")