mirror of
https://github.com/hasura/graphql-engine.git
synced 2024-12-16 18:42:30 +03:00
25d77afaff
Fixes https://github.com/hasura/graphql-engine-mono/issues/3695. Error: [BigQuery test failure Job exceeded rate limits](https://github.com/hasura/graphql-engine-mono/issues/3695) Cause: 1. [this command](2325755954/.circleci/test-server.sh (L1263)) runs tests matching the `Bigquery or Common` string, for the `test-oss-server-bigquery` CI job. 2. in this case, the pytest filter matched on `TestGraphQLQueryBoolExpSearchCommon`. Although unrelated pytests are skipped, BQ setup and teardown runs uneccesarily for the [MSSQL and Postgres backends](e444cf1f5d/server/tests-py/test_graphql_queries.py (L868)). 4. the setup and teardown runs three times in quick succession, _for each of_ SQL Server, Postgres and BigQuery. Occasionally, this surpassed [BigQuery's maximum rate of 5 table update operations in 10 seconds](https://cloud.google.com/bigquery/quotas#load_job_per_table.long). Fix: restrict setup/teardown to only the relevant backends... - Hotfix (this PR): ...by renaming pytest classes and changing the pytest filters in `test-server` - ok, this is faintly horrifying and an inelegant convention change. On the bright side, it shaves a minute or so off our integration test suite run by skipping fewer tests. Anecdata for `test-oss-server-bigquery` - before: 87 passed, 299 skipped, 1 warning, 1 error in 192.99s - after: 87 passed, 20 skipped, 1 warning in 170.82s - [`Common` was a terrible name, anyway](https://github.com/hasura/graphql-engine-mono/issues/2079), for `AnyCombinationOfBackends`. - Better fix: ...by refactoring the `conftest.py` helpers. I ran out of a timebox so will write up a separate issue. Given we're actively [porting pytests over to hspec](https://github.com/hasura/graphql-engine/issues/8432), I don't know how much it's worth investing time in a refactor. To verify the fix: I ran a full CI build a few times [[1]](https://buildkite.com/hasura/graphql-engine-mono/builds/8069#078c781a-c8ef-44f2-a400-15f91fb88e42)[[2]](https://buildkite.com/hasura/graphql-engine-mono/builds/8072#f9e7f59d-264f-46a4-973d-21aa762cca35)[[3]](https://buildkite.com/hasura/graphql-engine-mono/builds/8075#bb104e80-ff76-408c-a46b-6f40e92e6317) whilst troubleshooting to convince myself this fixed the problem. PR-URL: https://github.com/hasura/graphql-engine-mono/pull/4362 GitOrigin-RevId: 4c3283f0654b70e9dcda642d9012f6376aa95290
1760 lines
65 KiB
Python
1760 lines
65 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import pytest
|
|
import queue
|
|
import time
|
|
import json
|
|
import utils
|
|
from utils import *
|
|
from validate import check_query, check_query_f, check_event, check_event_transformed, check_events
|
|
|
|
usefixtures = pytest.mark.usefixtures
|
|
|
|
# Every test in this class requires the events webhook to be running first
|
|
# We are also going to mark as server upgrade tests are allowed
|
|
# A few tests are going to be excluded with skip_server_upgrade_test mark
|
|
pytestmark = [usefixtures('evts_webhook'), pytest.mark.allow_server_upgrade_test]
|
|
|
|
def select_last_event_fromdb(hge_ctx):
|
|
q = {
|
|
"type": "select",
|
|
"args": {
|
|
"table": {"schema": "hdb_catalog", "name": "event_log"},
|
|
"columns": ["*"],
|
|
"order_by": ["-created_at"],
|
|
"limit": 1
|
|
}
|
|
}
|
|
st_code, resp = hge_ctx.v1q(q)
|
|
return st_code, resp
|
|
|
|
def insert_mutation(hge_ctx, table, row, headers = {}):
|
|
return insert_many_mutation(hge_ctx, table, [row], headers)
|
|
|
|
def insert_many_mutation(hge_ctx, table, rows, headers = {}):
|
|
insert_mutation_field = ""
|
|
insert_values_type = ""
|
|
mutation_name = "insert" + "_" + table["name"]
|
|
if (table["schema"]):
|
|
insert_value_type = table["schema"] +"_" + table["name"] + "_" + "insert" + "_" + "input"
|
|
insert_mutation_field = "insert" + "_" + table["schema"] +"_" + table["name"]
|
|
else:
|
|
insert_value_type = table["name"] + "_" + "insert" + "_" + "input"
|
|
insert_mutation_field = "insert" + "_" + table["name"]
|
|
|
|
insert_mutation_query = """
|
|
mutation {mutation_name}($values: [{insert_value_type}!]!) {{
|
|
{insert_mutation_field}(objects: $values) {{
|
|
affected_rows
|
|
}}
|
|
}}
|
|
""".format(mutation_name = mutation_name, insert_value_type = insert_value_type, insert_mutation_field = insert_mutation_field )
|
|
|
|
variables = {'values': rows}
|
|
graphql_query = {'query': insert_mutation_query, 'variables': variables}
|
|
|
|
st_code, resp = hge_ctx.v1graphqlq(graphql_query, headers = headers)
|
|
return st_code, resp
|
|
|
|
def update_mutation(hge_ctx, table, where_exp, set_exp, headers = {}):
|
|
update_mutation_field = ""
|
|
mutation_name = "update" + "_" + table["name"]
|
|
|
|
if (table["schema"]):
|
|
update_mutation_field = "update" + "_" + table["schema"] +"_" + table["name"]
|
|
else:
|
|
update_mutation_field = "update" + "_" + table["name"]
|
|
|
|
update_mutation_query = """
|
|
mutation {mutation_name} {{
|
|
{update_mutation_field}(where: {where_exp}, _set: {set_exp}) {{
|
|
affected_rows
|
|
}}
|
|
}}
|
|
""".format(mutation_name = mutation_name,
|
|
update_mutation_field = update_mutation_field,
|
|
where_exp = where_exp,
|
|
set_exp = set_exp)
|
|
|
|
print("--- UPDATE MUTATION QUERY ---- \n", update_mutation_query)
|
|
|
|
graphql_query = {'query': update_mutation_query}
|
|
st_code, resp = hge_ctx.v1graphqlq(graphql_query, headers = headers)
|
|
|
|
#print(" ---- UPDATE MUTATION RESP ----", resp)
|
|
return st_code, resp
|
|
|
|
def delete_mutation(hge_ctx, table, where_exp, headers = {}):
|
|
delete_mutation_field = ""
|
|
mutation_name = "delete" + "_" + table["name"]
|
|
|
|
if (table["schema"]):
|
|
delete_mutation_field = "delete" + "_" + table["schema"] +"_" + table["name"]
|
|
else:
|
|
delete_mutation_field = "delete" + "_" + table["name"]
|
|
|
|
delete_mutation_query = """
|
|
mutation {mutation_name} {{
|
|
{delete_mutation_field}(where: {where_exp}) {{
|
|
affected_rows
|
|
}}
|
|
}}
|
|
""".format(mutation_name = mutation_name,
|
|
delete_mutation_field = delete_mutation_field,
|
|
where_exp = where_exp)
|
|
|
|
print("--- DELETE MUTATION QUERY ---- \n", delete_mutation_query)
|
|
|
|
graphql_query = {'query': delete_mutation_query}
|
|
st_code, resp = hge_ctx.v1graphqlq(graphql_query, headers = headers)
|
|
|
|
print(" ---- DELETE MUTATION RESP ----", resp)
|
|
return st_code, resp
|
|
|
|
@usefixtures("per_method_tests_db_state")
|
|
class TestEventCreateAndDelete:
|
|
|
|
def test_create_delete(self, hge_ctx):
|
|
check_query_f(hge_ctx, self.dir() + "/create_and_delete.yaml")
|
|
|
|
def test_create_reset(self, hge_ctx):
|
|
check_query_f(hge_ctx, self.dir() + "/create_and_reset.yaml")
|
|
|
|
# Can't run server upgrade tests, as this test has a schema change
|
|
@pytest.mark.skip_server_upgrade_test
|
|
def test_create_operation_spec_not_provider_err(self, hge_ctx):
|
|
check_query_f(hge_ctx, self.dir() + "/create_trigger_operation_specs_not_provided_err.yaml")
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/create-delete'
|
|
|
|
@usefixtures("per_method_tests_db_state")
|
|
class TestEventCreateAndResetNonDefaultSource:
|
|
|
|
def test_create_reset_non_default_source(self, hge_ctx):
|
|
check_query_f(hge_ctx, self.dir() + "/create_and_reset_non_default_source.yaml")
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/create_and_reset_non_default'
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures("per_method_tests_db_state")
|
|
class TestEventCreateAndDeleteMSSQL:
|
|
|
|
def test_create_delete(self, hge_ctx):
|
|
check_query_f(hge_ctx, self.dir() + "/create_and_delete_mssql.yaml")
|
|
|
|
def test_create_reset(self, hge_ctx):
|
|
check_query_f(hge_ctx, self.dir() + "/create_and_reset_mssql.yaml")
|
|
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
init_row = {"c1": 1, "c2": "world"}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
|
|
check_query_f(hge_ctx, self.dir() + "/create_and_reset_mssql_2.yaml")
|
|
|
|
# Can't run server upgrade tests, as this test has a schema change
|
|
@pytest.mark.skip_server_upgrade_test
|
|
def test_create_operation_spec_not_provider_err(self, hge_ctx):
|
|
check_query_f(hge_ctx, self.dir() + "/create_trigger_operation_specs_not_provided_err_mssql.yaml")
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/create-delete'
|
|
|
|
|
|
# Generates a backlog of events, then:
|
|
# - checks that we're processing with the concurrency and backpressure
|
|
# characteristics we expect
|
|
# - ensures all events are successfully processed
|
|
#
|
|
# NOTE: this expects:
|
|
# HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE=8
|
|
# HASURA_GRAPHQL_EVENTS_FETCH_BATCH_SIZE=100 (the default)
|
|
@pytest.mark.parametrize("backend", ['mssql','postgres'])
|
|
@usefixtures("per_method_tests_db_state")
|
|
class TestEventFloodPostgresMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/flood'
|
|
|
|
def test_flood(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_flood"}
|
|
|
|
# Trigger a bunch of events; hasura will begin processing but block on /block
|
|
payload = range(1,1001)
|
|
rows = list(map(lambda x: {"c1": x, "c2": "hello"}, payload))
|
|
|
|
if (hge_ctx.backend == "postgres"):
|
|
st_code, resp = insert_many(hge_ctx, table, rows)
|
|
elif (hge_ctx.backend == "mssql"):
|
|
st_code, resp = insert_many_mutation(hge_ctx, table, rows)
|
|
|
|
assert st_code == 200, resp
|
|
|
|
def check_backpressure():
|
|
# Expect that HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE webhooks are pending:
|
|
assert evts_webhook.blocked_count == 8
|
|
# ...Great, so presumably:
|
|
# - event handlers are run concurrently
|
|
# - with concurrency limited by HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE
|
|
|
|
if (hge_ctx.backend == "postgres"):
|
|
locked_counts = {
|
|
"type":"run_sql",
|
|
"args":{
|
|
"sql":'''
|
|
select
|
|
(select count(*) from hdb_catalog.event_log where locked IS NOT NULL) as num_locked,
|
|
count(*) as total
|
|
from hdb_catalog.event_log
|
|
where table_name = 'test_flood'
|
|
'''
|
|
}
|
|
}
|
|
st, resp = hge_ctx.v1q(locked_counts)
|
|
|
|
elif (hge_ctx.backend == "mssql"):
|
|
locked_counts = {
|
|
"type":"mssql_run_sql",
|
|
"args":{
|
|
"source": "mssql",
|
|
"sql":'''
|
|
select
|
|
(select count(*) from hdb_catalog.event_log where locked IS NOT NULL) as num_locked,
|
|
count(*) as total
|
|
from hdb_catalog.event_log
|
|
where table_name = 'test_flood'
|
|
'''
|
|
}
|
|
}
|
|
st, resp = hge_ctx.v2q(locked_counts)
|
|
|
|
assert st == 200, resp
|
|
# Make sure we have 2*HASURA_GRAPHQL_EVENTS_FETCH_BATCH_SIZE events checked out:
|
|
# - 100 prefetched
|
|
# - 100 being processed right now (but blocked on HTTP_POOL capacity)
|
|
# TODO it seems like we have some shared state in CI causing this to fail when we check 1000 below
|
|
# assert resp['result'][1] == ['200', '1000']
|
|
if (hge_ctx.backend == "postgres"):
|
|
assert resp['result'][1][0] == '200'
|
|
elif (hge_ctx.backend == "mssql"):
|
|
assert resp['result'][1][0] == 200
|
|
|
|
# Rather than sleep arbitrarily, loop until assertions pass:
|
|
utils.until_asserts_pass(30, check_backpressure)
|
|
# ...then make sure we're truly stable:
|
|
time.sleep(3)
|
|
check_backpressure()
|
|
|
|
# unblock open and future requests to /block; check all events processed
|
|
evts_webhook.unblock()
|
|
|
|
def get_evt():
|
|
# TODO ThreadedHTTPServer helps locally (I only need a timeout of
|
|
# 10 here), but we still need a bit of a long timeout here for CI
|
|
# it seems, since webhook can't keep up there:
|
|
ev_full = evts_webhook.get_event(600)
|
|
return ev_full['body']['event']['data']['new']['c1']
|
|
# Make sure we got all payloads (probably out of order):
|
|
ns = list(map(lambda _: get_evt(), payload))
|
|
ns.sort()
|
|
assert ns == list(payload)
|
|
|
|
@usefixtures("per_class_tests_db_state")
|
|
class TestEventDataFormat(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/data_format'
|
|
|
|
def test_bigint(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_bigint"}
|
|
|
|
init_row = {"id": 50755254975729665, "name": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": {"id": "50755254975729665", "name": "hello"}
|
|
}
|
|
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "bigint_all", table, "INSERT", exp_ev_data)
|
|
|
|
def test_geojson(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_geojson"}
|
|
|
|
exp_ev_data = {
|
|
"old": { "id" : 1,
|
|
"location":{
|
|
"coordinates":[
|
|
-43.77,
|
|
45.64
|
|
],
|
|
"crs":{
|
|
"type":"name",
|
|
"properties":{
|
|
"name":"urn:ogc:def:crs:EPSG::4326"
|
|
}
|
|
},
|
|
"type":"Point"
|
|
}
|
|
},
|
|
"new": { "id": 2,
|
|
"location":{
|
|
"coordinates":[
|
|
-43.77,
|
|
45.64
|
|
],
|
|
"crs":{
|
|
"type":"name",
|
|
"properties":{
|
|
"name":"urn:ogc:def:crs:EPSG::4326"
|
|
}
|
|
},
|
|
"type":"Point"
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
where_exp = {"id" : 1}
|
|
set_exp = {"id": 2}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "geojson_all", table, "UPDATE", exp_ev_data)
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures("per_class_tests_db_state")
|
|
class TestEventDataFormatBigIntMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/data_format/mssql/bigint'
|
|
|
|
def test_bigint(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_bigint"}
|
|
|
|
#init_row = {"id": 50755254975729665, "name": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": {"id": 50755254975729665, "name": "hello"}
|
|
}
|
|
|
|
# TODO: Naveen: Insert mutation on big int values in MSSQL source
|
|
# does not work as of now, hence using 'run_sql' to directly insert rows
|
|
# and trigger the event trigger. When they are supported in future, we
|
|
# might wanna use the insert_mutation here for consistency.
|
|
#st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
insert_bigint_sql = {
|
|
"type":"mssql_run_sql",
|
|
"args":{
|
|
"source": "mssql",
|
|
"sql":'''
|
|
INSERT INTO hge_tests.test_bigint ([id], [name]) VALUES (50755254975729665, 'hello')
|
|
'''
|
|
}
|
|
}
|
|
st_code, resp = hge_ctx.v2q(insert_bigint_sql)
|
|
print("----------- resp ----------\n", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "bigint_all", table, "INSERT", exp_ev_data)
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures("per_class_tests_db_state")
|
|
class TestEventDataFormatGeoJSONMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/data_format/mssql/geojson'
|
|
|
|
def test_geojson(self, hge_ctx, evts_webhook):
|
|
check_query_f(hge_ctx, self.dir() + '/create_geojson_event_trigger.yaml')
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql','postgres'])
|
|
@usefixtures("per_class_tests_db_state")
|
|
class TestCreateEventQueryPostgresMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/basic'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
# Check Insert Event Trigger Payload
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
|
|
if (hge_ctx.backend == "postgres"):
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
elif (hge_ctx.backend == "mssql"):
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
|
|
#assert st_code == 400, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data)
|
|
assert st_code == 200, resp
|
|
|
|
# Check Update Event Trigger Payload
|
|
if (hge_ctx.backend == "postgres"):
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c2": "world"}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
elif (hge_ctx.backend == "mssql"):
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c2: "world"}'
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "UPDATE", exp_ev_data)
|
|
|
|
# Check Delete Event Trigger Payload
|
|
if (hge_ctx.backend == "postgres"):
|
|
where_exp = {"c1": 1}
|
|
st_code, resp = delete(hge_ctx, table, where_exp)
|
|
elif (hge_ctx.backend == "mssql"):
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "DELETE", exp_ev_data)
|
|
|
|
|
|
def test_partitioned_table_basic_insert(self, hge_ctx, evts_webhook):
|
|
if (hge_ctx.backend == "postgres"):
|
|
if hge_ctx.pg_version < 110000:
|
|
pytest.skip('Event triggers on partioned tables are not supported in Postgres versions < 11')
|
|
return
|
|
st_code, resp = hge_ctx.v1q_f(self.dir() + '/partition_table_setup.yaml')
|
|
assert st_code == 200, resp
|
|
table = { "schema":"hge_tests", "name": "measurement"}
|
|
|
|
init_row = { "city_id": 1, "logdate": "2006-02-02", "peaktemp": 1, "unitsales": 1}
|
|
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "measurement_all", table, "INSERT", exp_ev_data)
|
|
st_code, resp = hge_ctx.v1q_f(self.dir() + '/partition_table_teardown.yaml')
|
|
assert st_code == 200, resp
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql','postgres'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventRetryConfPostgresMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/retry_conf'
|
|
|
|
# webhook: http://127.0.0.1:5592/fail
|
|
# retry_conf:
|
|
# num_retries: 4
|
|
# interval_sec: 1
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
if (hge_ctx.backend == "postgres"):
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
elif (hge_ctx.backend == "mssql"):
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_retry", table, "INSERT", exp_ev_data, webhook_path = "/fail", retry = 0)
|
|
check_event(hge_ctx, evts_webhook, "t1_retry", table, "INSERT", exp_ev_data, webhook_path = "/fail", retry = 1)
|
|
check_event(hge_ctx, evts_webhook, "t1_retry", table, "INSERT", exp_ev_data, webhook_path = "/fail", retry = 2)
|
|
check_event(hge_ctx, evts_webhook, "t1_retry", table, "INSERT", exp_ev_data, webhook_path = "/fail", retry = 3)
|
|
check_event(hge_ctx, evts_webhook, "t1_retry", table, "INSERT", exp_ev_data, webhook_path = "/fail", retry = 4)
|
|
|
|
# webhook: http://127.0.0.1:5592/sleep_2s
|
|
# retry_conf:
|
|
# num_retries: 2
|
|
# interval_sec: 1
|
|
# timeout_sec: 1
|
|
def test_timeout_short(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t2"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
if (hge_ctx.backend == "postgres"):
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
elif (hge_ctx.backend == "mssql"):
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t2_timeout_short", table, "INSERT", exp_ev_data, webhook_path = "/sleep_2s", retry = 0, get_timeout = 5)
|
|
check_event(hge_ctx, evts_webhook, "t2_timeout_short", table, "INSERT", exp_ev_data, webhook_path = "/sleep_2s", retry = 1, get_timeout = 5)
|
|
check_event(hge_ctx, evts_webhook, "t2_timeout_short", table, "INSERT", exp_ev_data, webhook_path = "/sleep_2s", retry = 2, get_timeout = 5)
|
|
|
|
# webhook: http://127.0.0.1:5592/sleep_2s
|
|
# retry_conf:
|
|
# num_retries: 0
|
|
# interval_sec: 2
|
|
# timeout_sec: 10
|
|
def test_timeout_long(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t3"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
if (hge_ctx.backend == "postgres"):
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
elif (hge_ctx.backend == "mssql"):
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
time.sleep(2)
|
|
check_event(hge_ctx, evts_webhook, "t3_timeout_long", table, "INSERT", exp_ev_data, webhook_path = "/sleep_2s")
|
|
|
|
# Keep this one last
|
|
def test_queue_empty(self, hge_ctx, evts_webhook):
|
|
try:
|
|
evts_webhook.get_event(3)
|
|
assert False, "expected queue to be empty"
|
|
except queue.Empty:
|
|
pass
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql','postgres'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventHeadersPostgresMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/headers'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
headers = {"X-Header-From-Value": "MyValue", "X-Header-From-Env": "MyEnvValue"}
|
|
if (hge_ctx.backend == "postgres"):
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
elif (hge_ctx.backend == "mssql"):
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data, headers = headers)
|
|
|
|
class TestUpdateEventQuery(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/update_query'
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def transact(self, request, hge_ctx, evts_webhook):
|
|
print("In setup method")
|
|
# Adds trigger on 'test_t1' with...
|
|
# insert:
|
|
# columns: '*'
|
|
# update:
|
|
# columns: [c2, c3]
|
|
st_code, resp = hge_ctx.v1q_f(self.dir() + '/create-setup.yaml')
|
|
assert st_code == 200, resp
|
|
|
|
# overwrites trigger added above, with...
|
|
# delete:
|
|
# columns: "*"
|
|
# update:
|
|
# columns: ["c1", "c3"]
|
|
st_code, resp = hge_ctx.v1q_f(self.dir() + '/update-setup.yaml')
|
|
assert st_code == 200, '{}'.format(resp)
|
|
assert resp[1]["sources"][0]["tables"][0]["event_triggers"][0]["webhook"] == 'http://127.0.0.1:5592/new'
|
|
yield
|
|
st_code, resp = hge_ctx.v1q_f(self.dir() + '/teardown.yaml')
|
|
assert st_code == 200, resp
|
|
|
|
def test_update_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
# Expect that inserting a row (which would have triggered in original
|
|
# create_event_trigger) does not trigger
|
|
init_row = {"c1": 1, "c2": "hello", "c3": {"name": "clarke"}}
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "INSERT", {}, webhook_path = "/new", get_timeout = 0)
|
|
|
|
# Likewise for an update on c2:
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c2": "world"}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", {}, webhook_path = "/new", get_timeout = 0)
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c3": {"name": "bellamy"}}
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world", "c3": {"name": "clarke"}},
|
|
"new": {"c1": 1, "c2": "world", "c3": {"name": "bellamy"}}
|
|
}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", exp_ev_data, webhook_path ="/new")
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c1": 2}
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world", "c3": {"name": "bellamy"}},
|
|
"new": {"c1": 2, "c2": "world", "c3": {"name": "bellamy"}}
|
|
}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", exp_ev_data, webhook_path ="/new")
|
|
|
|
where_exp = {"c1": 2}
|
|
exp_ev_data = {
|
|
"old": {"c1": 2, "c2": "world", "c3": {"name": "bellamy"}},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete(hge_ctx, table, where_exp)
|
|
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "DELETE", exp_ev_data, webhook_path = "/new")
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
class TestUpdateEventQueryMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/update_query'
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def transact(self, request, hge_ctx, evts_webhook):
|
|
print("In setup method")
|
|
# Adds trigger on 'test_t1' with...
|
|
# insert:
|
|
# columns: '*'
|
|
# update:
|
|
# columns: ["c3", "c4"]
|
|
st_code, resp = hge_ctx.v2q_f(self.dir() + '/schema-setup-mssql.yaml')
|
|
st_code, resp = hge_ctx.v1metadataq_f(self.dir() + '/create-setup-mssql.yaml')
|
|
assert st_code == 200, resp
|
|
|
|
# overwrites trigger added above, with...
|
|
# delete:
|
|
# columns: "*"
|
|
# update:
|
|
# columns: ["c1", "c2", "c4"]
|
|
|
|
st_code, resp = hge_ctx.v1metadataq_f(self.dir() + '/update-setup-mssql.yaml')
|
|
assert st_code == 200, '{}'.format(resp)
|
|
sources = resp[1]["sources"]
|
|
for source in sources:
|
|
if source["name"] == "mssql":
|
|
assert source["tables"][0]["event_triggers"][0]["webhook"] == 'http://127.0.0.1:5592/new'
|
|
|
|
yield
|
|
print("--- TEARDOWN STARTED -----")
|
|
st_code, resp = hge_ctx.v2q_f(self.dir() + '/teardown-mssql.yaml')
|
|
assert st_code == 200, resp
|
|
|
|
def test_update_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
# Expect that inserting a row (which would have triggered in original
|
|
# create_event_trigger) does not trigger
|
|
init_row = {"c1": 1, "c2": 100, "c3": "hello", "c4": "{'name': 'clarke'}"}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "INSERT", {}, webhook_path = "/new", get_timeout = 0)
|
|
|
|
# Likewise for an update on c3:
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c3: "world"}'
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("--- RESP 1 ---", resp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", {}, webhook_path = "/new", get_timeout = 0)
|
|
|
|
# Update on row c4 should initiate the event trigger
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c4: "{\'name\': \'bellamy\'}"}'
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2":100, "c3": "world", "c4": "{'name': 'clarke'}"},
|
|
"new": {"c1": 1, "c2":100, "c3": "world", "c4": "{'name': 'bellamy'}"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 2 ----", resp)
|
|
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", exp_ev_data, webhook_path ="/new")
|
|
|
|
# Update on row c2 should initiate the event trigger
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c2: 101}'
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2":100, "c3": "world", "c4": "{'name': 'bellamy'}"},
|
|
"new": {"c1": 1, "c2":101, "c3": "world", "c4": "{'name': 'bellamy'}"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 3 ----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", exp_ev_data, webhook_path ="/new")
|
|
|
|
# Test Delete Event Trigger
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2":101, "c3": "world", "c4": "{'name': 'bellamy'}"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
print("----- RESP 4 ----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "DELETE", exp_ev_data, webhook_path = "/new")
|
|
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestDeleteEventQuery(object):
|
|
|
|
directory = 'queries/event_triggers'
|
|
|
|
setup_files = [
|
|
directory + '/basic/setup.yaml',
|
|
directory + '/delete_query/setup.yaml'
|
|
]
|
|
|
|
teardown_files = [ directory + '/delete_query/teardown.yaml']
|
|
|
|
# Ensure deleting an event trigger works
|
|
def test_delete_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data, get_timeout=0)
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c2": "world"}
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "UPDATE", exp_ev_data, get_timeout=0)
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete(hge_ctx, table, where_exp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
# NOTE: use a bit of a delay here, to catch any stray events generated above
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "DELETE", exp_ev_data, get_timeout=2)
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestDeleteEventQueryMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/delete_query'
|
|
|
|
# Ensure deleting an event trigger works
|
|
def test_delete_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
print("----- RESP 1 -----", resp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data, get_timeout=0)
|
|
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c2: "world"}'
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 2 -----", resp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "UPDATE", exp_ev_data, get_timeout=0)
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
print("----- RESP 3 -----", resp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
# NOTE: use a bit of a delay here, to catch any stray events generated above
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "DELETE", exp_ev_data, get_timeout=2)
|
|
|
|
|
|
@usefixtures('per_class_tests_db_state')
|
|
class TestEventSelCols:
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/selected_cols'
|
|
|
|
def test_selected_cols(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": {"c1": 1, "c2": "hello"}
|
|
}
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "INSERT", exp_ev_data)
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c2": "world"}
|
|
# expected no event hence previous expected data
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", exp_ev_data, get_timeout=0)
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c1": 2}
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": {"c1": 2, "c2": "world"}
|
|
}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", exp_ev_data)
|
|
|
|
where_exp = {"c1": 2}
|
|
exp_ev_data = {
|
|
"old": {"c1": 2, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete(hge_ctx, table, where_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "DELETE", exp_ev_data)
|
|
|
|
@pytest.mark.skip_server_upgrade_test
|
|
def test_selected_cols_dep(self, hge_ctx, evts_webhook):
|
|
st_code, resp = hge_ctx.v1q({
|
|
"type": "run_sql",
|
|
"args": {
|
|
"sql": "alter table hge_tests.test_t1 drop column c1"
|
|
}
|
|
})
|
|
assert st_code == 400, resp
|
|
assert resp['code'] == "dependency-error", resp
|
|
|
|
st_code, resp = hge_ctx.v1q({
|
|
"type": "run_sql",
|
|
"args": {
|
|
"sql": "alter table hge_tests.test_t1 drop column c2"
|
|
}
|
|
})
|
|
assert st_code == 200, resp
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures('per_class_tests_db_state')
|
|
class TestEventSelColsMSSQL:
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/selected_cols'
|
|
|
|
def test_selected_cols(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello", "c3": "bellamy"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": {"c1": 1, "c2": "hello", "c3": "bellamy"}
|
|
}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
print("----- RESP 1 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "INSERT", exp_ev_data)
|
|
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c1: 2}'
|
|
|
|
# expected no event hence previous expected data
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 2 -----", resp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", exp_ev_data, get_timeout=0)
|
|
|
|
where_exp = '{c1: {_eq: 2}}'
|
|
set_exp = '{c2: "world"}'
|
|
exp_ev_data = {
|
|
"old": {"c1": 2, "c2": "hello", "c3": "bellamy"},
|
|
"new": {"c1": 2, "c2": "world", "c3": "bellamy"}
|
|
}
|
|
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 3 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "UPDATE", exp_ev_data)
|
|
|
|
where_exp = '{c1: {_eq: 2}}'
|
|
exp_ev_data = {
|
|
"old": {"c1": 2, "c2": "world", "c3": "bellamy"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
print("----- RESP 4 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_cols", table, "DELETE", exp_ev_data)
|
|
|
|
@pytest.mark.skip_server_upgrade_test
|
|
def test_selected_cols_dep(self, hge_ctx, evts_webhook):
|
|
# Dropping Primary Key is not allowed
|
|
st_code, resp = hge_ctx.v2q({
|
|
"type": "mssql_run_sql",
|
|
"args": {
|
|
"source": "mssql",
|
|
"sql": "alter table hge_tests.test_t1 drop column c1"
|
|
}
|
|
})
|
|
assert st_code == 400, resp
|
|
assert resp['code'] == "bad-request", resp
|
|
|
|
# 'C2' cannot be dropped because event trigger is created on that column
|
|
st_code, resp = hge_ctx.v2q({
|
|
"type": "mssql_run_sql",
|
|
"args": {
|
|
"source": "mssql",
|
|
"sql": "alter table hge_tests.test_t1 drop column c2"
|
|
}
|
|
})
|
|
print("----- RESP 5 -----", resp)
|
|
assert st_code == 400, resp
|
|
assert resp['code'] == "dependency-error", resp
|
|
|
|
st_code, resp = hge_ctx.v2q({
|
|
"type": "mssql_run_sql",
|
|
"args": {
|
|
"source": "mssql",
|
|
"sql": "alter table hge_tests.test_t1 drop column c3"
|
|
}
|
|
})
|
|
print("----- RESP 6 -----", resp)
|
|
assert st_code == 200, resp
|
|
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventInsertOnly:
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/insert_only'
|
|
|
|
def test_insert_only(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_insert", table, "INSERT", exp_ev_data)
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c2": "world"}
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_insert", table, "UPDATE", exp_ev_data, get_timeout=0)
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete(hge_ctx, table, where_exp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
# NOTE: use a bit of a delay here, to catch any stray events generated above
|
|
check_event(hge_ctx, evts_webhook, "t1_insert", table, "DELETE", exp_ev_data, get_timeout=2)
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventInsertOnlyMSSQL:
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/insert_only'
|
|
|
|
def test_insert_only(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
print("----- RESP 1 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_insert", table, "INSERT", exp_ev_data)
|
|
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c2: "world"}'
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
print("----- RESP 2 -----", resp)
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_insert", table, "UPDATE", exp_ev_data, get_timeout=0)
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
print("----- RESP 3 -----", resp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
# NOTE: use a bit of a delay here, to catch any stray events generated above
|
|
check_event(hge_ctx, evts_webhook, "t1_insert", table, "DELETE", exp_ev_data, get_timeout=2)
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventUpdateOnlyMSSQL:
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/update_only'
|
|
|
|
def test_update_only(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
print("----- RESP 1 -----", resp)
|
|
assert st_code == 200, resp
|
|
# INSERT operations will not fire event triggers
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_update", table, "INSERT", exp_ev_data, get_timeout=0)
|
|
|
|
# CASE 1: Primary key is not changed, and some updates happen
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c2: "world"}'
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "hello"},
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
print("----- RESP 2 -----", resp)
|
|
check_event(hge_ctx, evts_webhook, "t1_update", table, "UPDATE", exp_ev_data)
|
|
|
|
# CASE 2: Primary key has changed.
|
|
# When PK of MSSQL changes, then old data will be NULL
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c1: 2}'
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": {"c1": 2, "c2": "world"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 3 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_update", table, "UPDATE", exp_ev_data)
|
|
|
|
# DELETE operations will not fire event triggers
|
|
where_exp = '{c1: {_eq: 2}}'
|
|
exp_ev_data = {
|
|
"old": {"c1": 2, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
print("----- RESP 4 -----", resp)
|
|
assert st_code == 200, resp
|
|
with pytest.raises(queue.Empty):
|
|
# NOTE: use a bit of a delay here, to catch any stray events generated above
|
|
check_event(hge_ctx, evts_webhook, "t1_update", table, "DELETE", exp_ev_data, get_timeout=2)
|
|
|
|
# CASE 3: An Update transaction, which can give rise to both CASE 1 and CASE 2
|
|
# described above.
|
|
# i.e for a single update transaction which changes the primary key of a row
|
|
# and a non primary key of another row, 2 event triggers should be fired.
|
|
def test_update_both_cases(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
exp_insert_ev_data = {}
|
|
|
|
# Set up the table values to check the update transaction
|
|
insert_values_sql = {
|
|
"type":"mssql_run_sql",
|
|
"args":{
|
|
"source": "mssql",
|
|
"sql":'''
|
|
INSERT INTO hge_tests.test_t1 ([c1], [c2]) VALUES (1, 'hello'), (2, 'world')
|
|
'''
|
|
}
|
|
}
|
|
st_code, resp = hge_ctx.v2q(insert_values_sql)
|
|
print("----------- resp ----------\n", resp)
|
|
assert st_code == 200, resp
|
|
|
|
# INSERT operations will not fire event triggers
|
|
with pytest.raises(queue.Empty):
|
|
check_event(hge_ctx, evts_webhook, "t1_update", table, "INSERT", exp_insert_ev_data, get_timeout=0)
|
|
|
|
# An UPDATE SQL which will create two events, one for each case
|
|
# The following update transaction does the following changes
|
|
# We have the following values in table [(1, 'hello'), (2, 'world')]
|
|
# UPDATE transaction changes that to [(2, 'hello'), (3, 'clarke')]
|
|
update_values_sql = {
|
|
"type":"mssql_run_sql",
|
|
"args":{
|
|
"source": "mssql",
|
|
"sql":'''
|
|
UPDATE hge_tests.test_t1
|
|
SET c1 = (CASE WHEN c1 = 1 THEN 2
|
|
WHEN c1 = 2 THEN 3
|
|
ELSE c1 END),
|
|
c2 = (CASE WHEN c1 = 2 THEN N'clarke' ELSE c2 END)
|
|
'''
|
|
}
|
|
}
|
|
|
|
# CASE 2.1 : Primary key ('id') is updated, but the updated primary key value
|
|
# is already present in the table, then both data.old and data.new will be
|
|
# constructed
|
|
exp_ev_data_case_1 = {
|
|
"old": {"c1": 2, "c2": "world"},
|
|
"new": {"c1": 2, "c2": "hello"}
|
|
}
|
|
|
|
# CASE 2.2: Primary key ('id') is updated to a value which did not exists before
|
|
# in the table, so "data.old" will be NULL
|
|
exp_ev_data_case_2 = {
|
|
"old": None,
|
|
"new": {"c1": 3, "c2": "clarke"}
|
|
}
|
|
|
|
st_code, resp = hge_ctx.v2q(update_values_sql)
|
|
print("----------- resp ----------\n", resp)
|
|
assert st_code == 200, resp
|
|
|
|
exp_ev_datas = [exp_ev_data_case_1, exp_ev_data_case_2]
|
|
|
|
# The UPDATE SQL above will trigger exactly two triggers, one for each case
|
|
check_events(hge_ctx, evts_webhook, "t1_update", table, "UPDATE", 2, exp_ev_datas)
|
|
|
|
@usefixtures('per_class_tests_db_state')
|
|
class TestEventSelPayload:
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/selected_payload'
|
|
|
|
def test_selected_payload(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": {"c1": 1, "c2": "hello"}
|
|
}
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_payload", table, "INSERT", exp_ev_data)
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c2": "world"}
|
|
exp_ev_data = {
|
|
"old": {"c1": 1},
|
|
"new": {"c1": 1}
|
|
}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_payload", table, "UPDATE", exp_ev_data)
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c1": 2}
|
|
exp_ev_data = {
|
|
"old": {"c1": 1},
|
|
"new": {"c1": 2}
|
|
}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_payload", table, "UPDATE", exp_ev_data)
|
|
|
|
where_exp = {"c1": 2}
|
|
exp_ev_data = {
|
|
"old": {"c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete(hge_ctx, table, where_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_payload", table, "DELETE", exp_ev_data)
|
|
|
|
def test_selected_payload_dep(self, hge_ctx):
|
|
st_code, resp = hge_ctx.v1q({
|
|
"type": "run_sql",
|
|
"args": {
|
|
"sql": "alter table hge_tests.test_t1 drop column c1"
|
|
}
|
|
})
|
|
assert st_code == 400, resp
|
|
assert resp['code'] == "dependency-error", resp
|
|
|
|
st_code, resp = hge_ctx.v1q({
|
|
"type": "run_sql",
|
|
"args": {
|
|
"sql": "alter table hge_tests.test_t1 drop column c2"
|
|
}
|
|
})
|
|
assert st_code == 400, resp
|
|
assert resp['code'] == "dependency-error", resp
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures('per_class_tests_db_state')
|
|
class TestEventSelPayloadMSSQL:
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/selected_payload'
|
|
|
|
def test_selected_payload(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello", "c3": "bellamy"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": {"c1": 1, "c2": "hello", "c3": "bellamy"}
|
|
}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
print("----- RESP 1 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_payload", table, "INSERT", exp_ev_data)
|
|
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c2: "world"}'
|
|
exp_ev_data = {
|
|
"old": {"c2": "hello"},
|
|
"new": {"c2": "world"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 2 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_payload", table, "UPDATE", exp_ev_data)
|
|
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c3: "harry"}'
|
|
exp_ev_data = {
|
|
"old": {"c2": "world"},
|
|
"new": {"c2": "world"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 3 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_payload", table, "UPDATE", exp_ev_data)
|
|
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
exp_ev_data = {
|
|
"old": {"c3": "harry"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
print("----- RESP 4 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_payload", table, "DELETE", exp_ev_data)
|
|
|
|
def test_selected_payload_dep(self, hge_ctx):
|
|
# Dropping Primary Key is not allowed
|
|
st_code, resp = hge_ctx.v2q({
|
|
"type": "mssql_run_sql",
|
|
"args": {
|
|
"source": "mssql",
|
|
"sql": "alter table hge_tests.test_t1 drop column c1"
|
|
}
|
|
})
|
|
print("----- RESP 5 -----", resp)
|
|
assert st_code == 400, resp
|
|
assert resp['code'] == "bad-request", resp
|
|
|
|
# 'C2' cannot be dropped because event trigger is created on that column
|
|
st_code, resp = hge_ctx.v2q({
|
|
"type": "mssql_run_sql",
|
|
"args": {
|
|
"source": "mssql",
|
|
"sql": "alter table hge_tests.test_t1 drop column c2"
|
|
}
|
|
})
|
|
print("----- RESP 6 -----", resp)
|
|
assert st_code == 400, resp
|
|
assert resp['code'] == "dependency-error", resp
|
|
|
|
# 'C3' cannot be dropped because event trigger is created on that column
|
|
st_code, resp = hge_ctx.v2q({
|
|
"type": "mssql_run_sql",
|
|
"args": {
|
|
"source": "mssql",
|
|
"sql": "alter table hge_tests.test_t1 drop column c3"
|
|
}
|
|
})
|
|
print("----- RESP 7 -----", resp)
|
|
assert st_code == 400, resp
|
|
assert resp['code'] == "dependency-error", resp
|
|
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestWebhookEvent(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/webhook_env'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data)
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c2": "world"}
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "UPDATE", exp_ev_data)
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete(hge_ctx, table, where_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "DELETE", exp_ev_data)
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestWebhookEventMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/webhook_env'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
print("----- RESP 1 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data)
|
|
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c2: "world"}'
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 2 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "UPDATE", exp_ev_data)
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
print("----- RESP 3 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "DELETE", exp_ev_data)
|
|
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventWebhookTemplateURL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/webhook_template_url'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert(hge_ctx, table, init_row)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data, webhook_path = '/trigger')
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c2": "world"}
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "UPDATE", exp_ev_data, webhook_path = '/trigger')
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete(hge_ctx, table, where_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "DELETE", exp_ev_data, webhook_path = '/trigger')
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventWebhookTemplateURLMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/webhook_template_url'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row)
|
|
print("----- RESP 1 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data, webhook_path = '/trigger')
|
|
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c2: "world"}'
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp)
|
|
print("----- RESP 2 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "UPDATE", exp_ev_data, webhook_path = '/trigger')
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
print("----- RESP 3 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "DELETE", exp_ev_data, webhook_path = '/trigger')
|
|
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventSessionVariables(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/basic'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
session_variables = { 'x-hasura-role': 'admin', 'x-hasura-allowed-roles': "['admin','user']", 'x-hasura-user-id': '1'}
|
|
st_code, resp = insert(hge_ctx, table, init_row, headers = session_variables)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data, session_variables = session_variables)
|
|
|
|
where_exp = {"c1": 1}
|
|
set_exp = {"c2": "world"}
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
session_variables = { 'x-hasura-role': 'admin', 'x-hasura-random': 'some_random_info', 'X-Random-Header': 'not_session_variable'}
|
|
st_code, resp = update(hge_ctx, table, where_exp, set_exp, headers = session_variables)
|
|
assert st_code == 200, resp
|
|
session_variables.pop('X-Random-Header')
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "UPDATE", exp_ev_data, session_variables = session_variables)
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete(hge_ctx, table, where_exp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "DELETE", exp_ev_data)
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventSessionVariablesMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/basic'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
init_row = {"c1": 1, "c2": "hello"}
|
|
exp_ev_data = {
|
|
"old": None,
|
|
"new": init_row
|
|
}
|
|
session_variables = { 'x-hasura-role': 'admin', 'x-hasura-allowed-roles': "['admin','user']", 'x-hasura-user-id': '1'}
|
|
st_code, resp = insert_mutation(hge_ctx, table, init_row, headers = session_variables)
|
|
assert st_code == 200, resp
|
|
print("----- RESP 1 -----", resp)
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "INSERT", exp_ev_data, session_variables = session_variables)
|
|
|
|
where_exp = '{c1: {_eq: 1}}'
|
|
set_exp = '{c2: "world"}'
|
|
exp_ev_data = {
|
|
"old": init_row,
|
|
"new": {"c1": 1, "c2": "world"}
|
|
}
|
|
session_variables = { 'x-hasura-role': 'admin', 'x-hasura-random': 'some_random_info', 'X-Random-Header': 'not_session_variable'}
|
|
st_code, resp = update_mutation(hge_ctx, table, where_exp, set_exp, headers = session_variables)
|
|
assert st_code == 200, resp
|
|
print("----- RESP 2 -----", resp)
|
|
session_variables.pop('X-Random-Header')
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "UPDATE", exp_ev_data, session_variables = session_variables)
|
|
|
|
exp_ev_data = {
|
|
"old": {"c1": 1, "c2": "world"},
|
|
"new": None
|
|
}
|
|
st_code, resp = delete_mutation(hge_ctx, table, where_exp)
|
|
print("----- RESP 3 -----", resp)
|
|
assert st_code == 200, resp
|
|
check_event(hge_ctx, evts_webhook, "t1_all", table, "DELETE", exp_ev_data)
|
|
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestManualEvents(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/manual_events'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
st_code, resp = hge_ctx.v1metadataq_f('queries/event_triggers/manual_events/enabled.yaml')
|
|
assert st_code == 200, resp
|
|
print("----- RESP 1 -----", resp)
|
|
st_code, resp = hge_ctx.v1metadataq_f('queries/event_triggers/manual_events/disabled.yaml')
|
|
assert st_code == 400, resp
|
|
print("----- RESP 2 -----", resp)
|
|
|
|
# This test is being added to ensure that the manual events
|
|
# are not failing after any reload_metadata operation, this
|
|
# has been an issue of concern in some of the recent releases(v2.0.1 onwards)
|
|
def test_basic_with_reload_metadata(self, hge_ctx, evts_webhook):
|
|
reload_metadata_q = {
|
|
"type": "reload_metadata",
|
|
"args": {
|
|
"source": "mssql",
|
|
"reload_sources": True
|
|
}
|
|
}
|
|
|
|
for _ in range(5):
|
|
self.test_basic(hge_ctx, evts_webhook)
|
|
|
|
st_code, resp = hge_ctx.v1metadataq(reload_metadata_q)
|
|
assert st_code == 200, resp
|
|
print("----- RESP 3 -----", resp)
|
|
|
|
self.test_basic(hge_ctx, evts_webhook)
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestManualEventsMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/manual_events'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
st_code, resp = hge_ctx.v1metadataq_f('queries/event_triggers/manual_events/enabled-mssql.yaml')
|
|
assert st_code == 200, resp
|
|
st_code, resp = hge_ctx.v1metadataq_f('queries/event_triggers/manual_events/disabled-mssql.yaml')
|
|
assert st_code == 400, resp
|
|
|
|
# This test is being added to ensure that the manual events
|
|
# are not failing after any reload_metadata operation, this
|
|
# has been an issue of concern in some of the recent releases(v2.0.1 onwards)
|
|
def test_basic_with_reload_metadata(self, hge_ctx, evts_webhook):
|
|
reload_metadata_q = {
|
|
"type": "reload_metadata",
|
|
"args": {
|
|
"reload_sources": True
|
|
}
|
|
}
|
|
|
|
for _ in range(5):
|
|
self.test_basic(hge_ctx, evts_webhook)
|
|
|
|
st_code, resp = hge_ctx.v1metadataq(reload_metadata_q)
|
|
assert st_code == 200, resp
|
|
|
|
self.test_basic(hge_ctx, evts_webhook)
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql','postgres'])
|
|
@usefixtures('per_method_tests_db_state')
|
|
class TestEventsAsynchronousExecutionPostgresMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/async_execution'
|
|
|
|
def test_async_execution(self,hge_ctx,evts_webhook):
|
|
"""
|
|
A test to check if the events generated by the graphql-engine are
|
|
processed asynchronously. This test measures the time taken to process
|
|
all the events and that time should definitely be lesser than the time
|
|
taken if the events were to be executed sequentially.
|
|
|
|
This test inserts 5 rows and the webhook(/sleep_2s) takes
|
|
~2 seconds to process one request. So, if the graphql-engine
|
|
were to process the events sequentially it will take 5 * 2 = 10 seconds.
|
|
Theorotically, all the events should have been processed in ~2 seconds,
|
|
adding a 5 seconds buffer to the comparision, so that this test
|
|
doesn't flake in the CI.
|
|
"""
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
|
|
payload = range(1,6)
|
|
rows = list(map(lambda x: {"c1": x, "c2": "hello"}, payload))
|
|
if (hge_ctx.backend == "postgres"):
|
|
st_code, resp = insert_many(hge_ctx, table, rows)
|
|
elif (hge_ctx.backend == "mssql"):
|
|
st_code, resp = insert_many_mutation(hge_ctx, table, rows)
|
|
print("----- RESP 1 -----", resp)
|
|
start_time = time.perf_counter()
|
|
assert st_code == 200, resp
|
|
for i in range(1,6):
|
|
_ = evts_webhook.get_event(5) # webhook takes 2 seconds to process a request (+ buffer)
|
|
end_time = time.perf_counter()
|
|
time_elapsed = end_time - start_time
|
|
assert time_elapsed < 10
|
|
|
|
@usefixtures("per_class_tests_db_state")
|
|
class TestEventTransform(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/transform'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
# GIVEN
|
|
check_query_f(hge_ctx, self.dir() + '/basic_transform.yaml')
|
|
|
|
# WHEN
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
insert_row = {"id": 0, "first_name": "Simon", "last_name": "Marlow"}
|
|
st_code, resp = insert(hge_ctx, table, insert_row)
|
|
|
|
# THEN
|
|
expectedPath = "/?foo=bar"
|
|
expectedBody = insert_row
|
|
|
|
check_event_transformed(hge_ctx,
|
|
evts_webhook,
|
|
expectedBody,
|
|
headers={"foo": "bar"},
|
|
removedHeaders=["user-agent"],
|
|
webhook_path=expectedPath)
|
|
assert st_code == 200, resp
|
|
|
|
@pytest.mark.parametrize("backend", ['mssql'])
|
|
@usefixtures("per_method_tests_db_state")
|
|
class TestEventTransformMSSQL(object):
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/event_triggers/transform'
|
|
|
|
def test_basic(self, hge_ctx, evts_webhook):
|
|
# GIVEN
|
|
check_query_f(hge_ctx, self.dir() + '/basic_transform_mssql.yaml')
|
|
|
|
# WHEN
|
|
table = {"schema": "hge_tests", "name": "test_t1"}
|
|
insert_row = {"id": 0, "first_name": "Simon", "last_name": "Marlow"}
|
|
st_code, resp = insert_mutation(hge_ctx, table, insert_row)
|
|
print("----- RESP 1 -----", resp)
|
|
|
|
# THEN
|
|
expectedPath = "/?foo=bar"
|
|
expectedBody = insert_row
|
|
|
|
check_event_transformed(hge_ctx,
|
|
evts_webhook,
|
|
expectedBody,
|
|
headers={"foo": "bar"},
|
|
removedHeaders=["user-agent"],
|
|
webhook_path=expectedPath)
|
|
assert st_code == 200, resp
|