(query)
| 189 | |
| 190 | @pytest.mark.parametrize("query", [ADD_NODE, ADD_EDGE, ADD_TEMP_PROP, ADD_CONST_PROP]) |
| 191 | def test_update_graph(query): |
| 192 | work_dir = tempfile.mkdtemp() |
| 193 | with GraphServer( |
| 194 | work_dir, config={"auth": {"public_key": PUB_KEY}} |
| 195 | ).start() as server: |
| 196 | port = server.port() |
| 197 | add_test_graph(port) |
| 198 | data = json.dumps({"query": query}) |
| 199 | |
| 200 | response = requests.post(raphtory_url(port), data=data) |
| 201 | assert response.status_code == 401 |
| 202 | |
| 203 | response = requests.post(raphtory_url(port), headers=READ_HEADERS, data=data) |
| 204 | assert response.json()["data"] is None |
| 205 | assert ( |
| 206 | response.json()["errors"][0]["message"] |
| 207 | == "The requested endpoint requires write access" |
| 208 | ) |
| 209 | |
| 210 | response = requests.post(raphtory_url(port), headers=WRITE_HEADERS, data=data) |
| 211 | assert_successful_response(response) |
| 212 | |
| 213 | |
| 214 | NEW_GRAPH = """mutation { newGraph(path:"new", graphType:EVENT) }""" |
nothing calls this directly
no test coverage detected