mirror of
https://github.com/hasura/graphql-engine.git
synced 2024-12-21 06:21:39 +03:00
a0c83b8e99
PR-URL: https://github.com/hasura/graphql-engine-mono/pull/3860 GitOrigin-RevId: 02d1963946d69fdf6886aa61806eb71db60f8931
80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
import pytest
|
|
import json
|
|
import jsondiff
|
|
from ruamel.yaml import YAML
|
|
|
|
yaml=YAML(typ='safe', pure=True)
|
|
|
|
class TestInconsistentObjects():
|
|
|
|
reload_metadata = {
|
|
"type": "reload_metadata",
|
|
"args": {}
|
|
}
|
|
drop_inconsistent_metadata = {
|
|
"type": "drop_inconsistent_metadata",
|
|
"args": {}
|
|
}
|
|
export_metadata = {
|
|
"type": "export_metadata",
|
|
"args": {}
|
|
}
|
|
|
|
def test_inconsistent_objects(self, hge_ctx):
|
|
with open(self.dir() + "/test.yaml") as c:
|
|
test = yaml.load(c)
|
|
|
|
# setup
|
|
st_code, resp = hge_ctx.v1q(json.loads(json.dumps(test['setup'])))
|
|
assert st_code == 200, resp
|
|
|
|
try:
|
|
# exec sql to cause inconsistentancy
|
|
sql_res = hge_ctx.sql(test['sql'])
|
|
|
|
# reload metadata
|
|
st_code, resp = hge_ctx.v1q(q=self.reload_metadata)
|
|
assert st_code == 200, resp
|
|
# check inconsistent objects
|
|
incons_objs_test = test['inconsistent_objects']
|
|
incons_objs_resp = resp['inconsistent_objects']
|
|
|
|
assert resp['is_consistent'] == False, resp
|
|
assert incons_objs_resp == incons_objs_test, yaml.dump({
|
|
'response': incons_objs_resp,
|
|
'expected': incons_objs_test,
|
|
'diff': jsondiff.diff(incons_objs_test, incons_objs_resp)
|
|
})
|
|
|
|
# export metadata
|
|
st_code, export = hge_ctx.v1q(q=self.export_metadata)
|
|
assert st_code == 200, export
|
|
|
|
# apply metadata
|
|
st_code, resp = hge_ctx.v1q(
|
|
q={
|
|
"type": "replace_metadata",
|
|
"args": export
|
|
}
|
|
)
|
|
assert st_code == 400, resp
|
|
|
|
finally:
|
|
# drop inconsistent objects
|
|
st_code, resp = hge_ctx.v1q(q=self.drop_inconsistent_metadata)
|
|
assert st_code == 200, resp
|
|
|
|
# reload metadata
|
|
st_code, resp = hge_ctx.v1q(q=self.reload_metadata)
|
|
assert st_code == 200, resp
|
|
# check inconsistent objects
|
|
assert resp['is_consistent'] == True, resp
|
|
|
|
# teardown
|
|
st_code, resp = hge_ctx.v1q(json.loads(json.dumps(test['teardown'])))
|
|
assert st_code == 200, resp
|
|
|
|
@classmethod
|
|
def dir(cls):
|
|
return 'queries/inconsistent_objects'
|