graphql-engine/server/tests-py/test_endpoints.py
Samir Talwar 44ad07eeab server/tests-py: Fix up the Redis tests so they run.
This fixes the existing tests around query caching and other Redis-related behaviors so they actually run.

Some of the tests still fail because of a couple of bugs, but the error messages are now meaningful.

They do not yet run in CI. We will enable them once they are fixed.

I added a bit of extra logging; we now log the Redis server host and port. I also left in the `Show` instance derivations I used for debugging, as I don't think they'll harm anything and might be useful in the future.

The changes are as follows:

1. I added a `redis` server to the tests and configured both HGE and the test runner to talk to it.
2. I fixed up the action tests so they set up and tear down correctly, including a fix to the output type of the action.
3. Caching has been implemented for actions with client header forwarding, so I have changed the test accordingly.
4. Tests now fail correctly if the response headers are wrong but the body is correct.
5. I have updated the keys in the response headers as the algorithm for generating them seems to have changed.
6. A few improvements have been made to the Python tests to handle lint warnings and improve logging.

PR-URL: https://github.com/hasura/graphql-engine-mono/pull/10839
GitOrigin-RevId: 5d31a376346190b5c7b398b4dee84b88a9d1b18b
2024-05-28 15:12:10 +00:00

84 lines
4.0 KiB
Python

import pytest
from validate import check_query_f
@pytest.mark.parametrize("transport", ['http'])
@pytest.mark.usefixtures('postgis', 'per_class_tests_db_state')
class TestCustomEndpoints:
@classmethod
def dir(cls):
return 'queries/endpoints'
def test_missing_endpoint(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_missing.yaml', transport)
def test_endpoint_as_user_err(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_as_user_err.yaml', transport)
def test_simple_endpoint(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_simple.yaml', transport)
def test_simple_endpoint_wrong_method(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_simple_wrong_method.yaml', transport)
@pytest.mark.usefixtures('pro_tests_fixtures', 'flush_redis')
def test_simple_cached_endpoint(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_simple_cached.yaml', transport)
def test_endpoint_with_query_arg(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_with_query_arg.yaml', transport)
def test_endpoint_with_body_arg(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_with_body_arg.yaml', transport)
def test_endpoint_with_body_list_arg(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_with_list_arg.yaml', transport)
def test_endpoint_with_query_arg_url_encoded(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_with_query_arg_url_encoded.yaml', transport)
def test_endpoint_with_query_arg_url_encoded_2(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_with_query_arg_url_encoded_2.yaml', transport)
def test_endpoint_with_query_args_missing_arg(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_with_query_arg_missing_arg.yaml', transport)
def test_endpoint_with_multiple_query_args(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_with_multiple_query_args.yaml', transport)
def test_endpoint_with_template(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_with_template.yaml', transport)
def test_endpoint_dropped(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_dropped.yaml', transport)
def test_endpoint_conflicting(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_conflicting.yaml', transport)
def test_endpoint_duplicate_param(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_duplicate_param.yaml', transport)
def test_endpoint_empty_path(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_empty_path.yaml', transport)
def test_endpoint_trailing_slash(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_trailing_slash.yaml', transport)
def test_endpoint_empty_path_segment(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_empty_path_segment.yaml', transport)
def test_endpoint_empty_path_param(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_empty_path_param.yaml', transport)
def test_endpoint_uuid_arg(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_with_uuid_arg.yaml', transport)
def test_endpoint_subscription(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_subscription.yaml', transport)
def test_query_validation(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_query_validation.yaml', transport)
def test_endpoint_with_pg_date_url_variable(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/endpoint_pg_date_url_variable.yaml', transport)