hasura/graphql-engine
1
0
Fork 0
mirror of https://github.com/hasura/graphql-engine.git synced 2024-12-15 09:22:43 +03:00
Code Issues Projects Releases Wiki Activity
881cc55e0e
graphql-engine/server/tests-py/test_graphql_mutations.py
Karthikeyan Chinnakonda 92026b769f [Preview] Inherited roles for postgres read queries 
fixes #3868

docker image - `hasura/graphql-engine:inherited-roles-preview-48b73a2de`

Note:

To be able to use the inherited roles feature, the graphql-engine should be started with the env variable `HASURA_GRAPHQL_EXPERIMENTAL_FEATURES` set to `inherited_roles`.

Introduction
------------

This PR implements the idea of multiple roles as presented in this [paper](https://www.microsoft.com/en-us/research/wp-content/uploads/2016/02/FGALanguageICDE07.pdf). The multiple roles feature in this PR can be used via inherited roles. An inherited role is a role which can be created by combining multiple singular roles. For example, if there are two roles `author` and `editor` configured in the graphql-engine, then we can create a inherited role with the name of `combined_author_editor` role which will combine the select permissions of the `author` and `editor` roles and then make GraphQL queries using the `combined_author_editor`.

How are select permissions of different roles are combined?
------------------------------------------------------------

A select permission includes 5 things:

1. Columns accessible to the role
2. Row selection filter
3. Limit
4. Allow aggregation
5. Scalar computed fields accessible to the role

 Suppose there are two roles, `role1` gives access to the `address` column with row filter `P1` and `role2` gives access to both the `address` and the `phone` column with row filter `P2` and we create a new role `combined_roles` which combines `role1` and `role2`.

Let's say the following GraphQL query is queried with the `combined_roles` role.

```graphql
query {
   employees {
     address
     phone
   }
}
```

This will translate to the following SQL query:

```sql

 select
    (case when (P1 or P2) then address else null end) as address,
    (case when P2 then phone else null end) as phone
 from employee
 where (P1 or P2)
```

The other parameters of the select permission will be combined in the following manner:

1. Limit - Minimum of the limits will be the limit of the inherited role
2. Allow aggregations - If any of the role allows aggregation, then the inherited role will allow aggregation
3. Scalar computed fields - same as table column fields, as in the above example

APIs for inherited roles:
----------------------

1. `add_inherited_role`

`add_inherited_role` is the [metadata API](https://hasura.io/docs/1.0/graphql/core/api-reference/index.html#schema-metadata-api) to create a new inherited role. It accepts two arguments

`role_name`: the name of the inherited role to be added (String)
`role_set`: list of roles that need to be combined (Array of Strings)

Example:

```json
{
  "type": "add_inherited_role",
  "args": {
      "role_name":"combined_user",
      "role_set":[
          "user",
          "user1"
      ]
  }
}
```

After adding the inherited role, the inherited role can be used like single roles like earlier

Note:

An inherited role can only be created with non-inherited/singular roles.

2. `drop_inherited_role`

The `drop_inherited_role` API accepts the name of the inherited role and drops it from the metadata. It accepts a single argument:

`role_name`: name of the inherited role to be dropped

Example:

```json

{
  "type": "drop_inherited_role",
  "args": {
      "role_name":"combined_user"
  }
}
```

Metadata
---------

The derived roles metadata will be included under the `experimental_features` key while exporting the metadata.

```json
{
  "experimental_features": {
    "derived_roles": [
      {
        "role_name": "manager_is_employee_too",
        "role_set": [
          "employee",
          "manager"
        ]
      }
    ]
  }
}
```

Scope
------

Only postgres queries and subscriptions are supported in this PR.

Important points:
-----------------

1. All columns exposed to an inherited role will be marked as `nullable`, this is done so that cell value nullification can be done.

TODOs
-------

- [ ] Tests
   - [ ] Test a GraphQL query running with a inherited role without enabling inherited roles in experimental features
   - [] Tests for aggregate queries, limit, computed fields, functions, subscriptions (?)
   - [ ] Introspection test with a inherited role (nullability changes in a inherited role)
- [ ] Docs
- [ ] Changelog

Co-authored-by: Vamshi Surabhi <6562944+0x777@users.noreply.github.com>
GitOrigin-RevId: 3b8ee1e11f5ceca80fe294f8c074d42fbccfec63
2021-03-08 11:15:10 +00:00

658 lines
27 KiB
Python
Raw Blame History

import pytest
from validate import check_query_f, check_query, get_conf_f
from conftest import use_inherited_roles_fixtures
# Marking all tests in this module that server upgrade tests can be run
# Few of them cannot be run, which will be marked skip_server_upgrade_test
pytestmark = pytest.mark.allow_server_upgrade_test
usefixtures = pytest.mark.usefixtures
use_mutation_fixtures = usefixtures(
'per_class_db_schema_for_mutation_tests',
'per_method_db_data_for_mutation_tests'
)
@pytest.mark.parametrize("transport", ['http', 'websocket'])
@use_mutation_fixtures
class TestGraphQLInsertWithTransport:
def test_inserts_author_article(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + "/author_article.yaml", transport)
@classmethod
def dir(cls):
return "queries/graphql_mutation/insert/basic"
@use_mutation_fixtures
class TestGraphQLInsert:
def test_inserts_various_postgres_types(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_various_postgres_types.yaml")
def test_insert_integer_overflow(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_integer_overflow.yaml")
@pytest.mark.xfail(reason="Refer https://github.com/hasura/graphql-engine/issues/348")
def test_insert_into_array_col_with_array_input(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_into_array_col_with_array_input.yaml")
def test_insert_using_variable(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_jsonb_variable.yaml")
def test_insert_using_array_variable(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_jsonb_variable_array.yaml")
def test_insert_person(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_jsonb.yaml")
def test_insert_person_array(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_jsonb_array.yaml")
def test_insert_null_col_value(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/order_col_shipped_null.yaml")
def test_insert_valid_variable_but_invalid_graphql_value(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_valid_variable_but_invalid_graphql_value.yaml")
def test_can_insert_in_insertable_view(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/can_insert_in_insertable_view.yaml")
def test_cannot_insert_in_non_insertable_view(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/cannot_insert_in_non_insertable_view.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/insert/basic"
@use_mutation_fixtures
class TestGraphqlInsertOnConflict:
def test_on_conflict_update(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/article_on_conflict_update.yaml")
def test_on_conflict_ignore(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/article_on_conflict_ignore_constraint.yaml")
def test_on_conflict_update_empty_cols(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/article_on_conflict_empty_update_columns.yaml")
def test_err_missing_article_constraint(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/article_on_conflict_error_missing_article_constraint.yaml")
def test_err_unexpected_action(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/article_unexpected_on_conflict_action.yaml")
def test_err_unexpected_constraint(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/article_unexpected_on_conflict_constraint_error.yaml")
def test_order_on_conflict_where(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + '/order_on_conflict_where.yaml')
@classmethod
def dir(cls):
return "queries/graphql_mutation/insert/onconflict"
@use_mutation_fixtures
class TestGraphqlInsertPermission:
def test_user_role_on_conflict_update(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/article_on_conflict_user_role.yaml")
def test_user_role_on_conflict_constraint_on_error(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/article_on_conflict_constraint_on_user_role_error.yaml")
def test_user_role_on_conflict_ignore(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_on_conflict_ignore_user_role.yaml")
def test_user_err_missing_article_constraint(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_article_on_conflict_error_missing_article_constraint.yaml")
def test_user_err_unexpected_action(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_article_error_unexpected_on_conflict_action.yaml")
def test_user_err_unexpected_constraint(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_article_unexpected_on_conflict_constraint_error.yaml")
def test_role_has_no_permissions_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/address_permission_error.yaml")
def test_author_user_role_insert_check_perm_success(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_user_role_insert_check_perm_success.yaml")
def test_user_role_insert_check_is_registered_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_user_role_insert_check_is_registered_fail.yaml")
def test_user_role_insert_check_user_id_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_user_role_insert_check_user_id_fail.yaml")
def test_student_role_insert_check_bio_success(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_student_role_insert_check_bio_success.yaml")
def test_student_role_insert_check_bio_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_student_role_insert_check_bio_fail.yaml")
def test_company_user_role_insert(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/company_user_role.yaml")
def test_company_user_role_insert_on_conflict(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/company_user_role_on_conflict.yaml")
def test_resident_user_role_insert(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/resident_user.yaml")
def test_resident_infant_role_insert(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/resident_infant.yaml")
def test_resident_infant_role_insert_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/resident_infant_fail.yaml")
def test_resident_5_modifies_resident_6_upsert(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/resident_5_modifies_resident_6_upsert.yaml")
def test_resident_on_conflict_where(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/resident_on_conflict_where.yaml")
def test_blog_on_conflict_update_preset(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/blog_on_conflict_update_preset.yaml")
def test_arr_sess_var_insert_article_as_editor_allowed_user_id(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_article_arr_sess_var_editor_allowed_user_id.yaml")
def test_arr_sess_var_insert_article_as_editor_err_not_allowed_user_id(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_article_arr_sess_var_editors_err_not_allowed_user_id.yaml")
def test_seller_insert_computer_json_has_keys_all(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/seller_insert_computer_has_keys_all_pass.yaml")
def test_seller_insert_computer_json_has_keys_all_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/seller_insert_computer_has_keys_all_fail.yaml")
def test_developer_insert_computer_json_has_keys_any(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/developer_insert_has_keys_any_pass.yaml")
def test_developer_insert_computer_json_has_keys_any_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/developer_insert_has_keys_any_fail.yaml")
def test_user_insert_account_success(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_insert_account_success.yaml")
def test_user_insert_account_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_insert_account_fail.yaml")
def test_backend_user_insert_fail(self, hge_ctx):
check_query_admin_secret(hge_ctx, self.dir() + "/backend_user_insert_fail.yaml")
def test_backend_user_insert_pass(self, hge_ctx):
check_query_admin_secret(hge_ctx, self.dir() + "/backend_user_insert_pass.yaml")
def test_backend_user_insert_invalid_bool(self, hge_ctx):
check_query_admin_secret(hge_ctx, self.dir() + "/backend_user_insert_invalid_bool.yaml")
def test_user_with_no_backend_privilege(self, hge_ctx):
check_query_admin_secret(hge_ctx, self.dir() + "/user_with_no_backend_privilege.yaml")
def test_backend_user_no_admin_secret_fail(self, hge_ctx):
if hge_ctx.hge_key and (hge_ctx.hge_jwt_key or hge_ctx.hge_webhook):
check_query_f(hge_ctx, self.dir() + "/backend_user_no_admin_secret_fail.yaml")
else:
pytest.skip("authorization not configured, skipping the test")
def test_check_set_headers_while_doing_upsert(self,hge_ctx):
check_query_f(hge_ctx, self.dir() + "/leads_upsert_check_with_headers.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/insert/permissions"
def check_query_admin_secret(hge_ctx, f, transport='http'):
conf = get_conf_f(f)
admin_secret = hge_ctx.hge_key
if admin_secret:
conf['headers']['x-hasura-admin-secret'] = admin_secret
check_query(hge_ctx, conf, transport, False)
@usefixtures('per_class_tests_db_state')
class TestGraphqlInsertConstraints:
def test_address_not_null_constraint_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/address_not_null_constraint_error.yaml")
def test_insert_unique_constraint_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_unique_constraint_error.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/insert/constraints"
@usefixtures('per_class_tests_db_state')
class TestGraphqlInsertNullPrefixedColumnOnConflict:
def test_address_not_null_constraint_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/null_prefixed_column_ok.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/insert/nullprefixcolumn"
@use_mutation_fixtures
class TestGraphqlInsertGeoJson:
def test_insert_point_landmark(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_landmark.yaml")
def test_insert_3d_point_drone_loc(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_drone_3d_location.yaml")
def test_insert_landmark_single_position_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_landmark_single_position_err.yaml")
def test_insert_line_string_road(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_road.yaml")
def test_insert_road_single_point_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_road_single_point_err.yaml")
def test_insert_multi_point_service_locations(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_service_locations.yaml")
def test_insert_multi_line_string_route(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_route.yaml")
def test_insert_polygon(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_area.yaml")
def test_insert_linear_ring_less_than_4_points_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_area_less_than_4_points_err.yaml")
def test_insert_linear_ring_last_point_not_equal_to_first_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_linear_ring_last_point_not_equal_to_first_err.yaml")
def test_insert_multi_polygon_compounds(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_compounds.yaml")
def test_insert_geometry_collection(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_geometry_collection.yaml")
def test_insert_unexpected_geometry_type_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_geometry_unexpected_type_err.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/insert/geojson"
@use_mutation_fixtures
# Skipping server upgrade tests for a few tests below
# Those tests capture bugs in the previous release
class TestGraphqlNestedInserts:
def test_author_with_detail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_with_detail.yaml")
def test_author_with_detail_fk(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_with_detail_fk.yaml")
def test_author_with_articles(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_with_articles.yaml")
def test_author_with_articles_empty(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_with_articles_empty.yaml")
def test_author_with_articles_null(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_with_articles_null.yaml")
def test_author_with_articles_author_id_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_with_articles_author_id_fail.yaml")
def test_articles_with_author(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/articles_with_author.yaml")
def test_articles_with_author_author_id_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/articles_with_author_author_id_fail.yaml")
def test_author_upsert_articles_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_upsert_articles_fail.yaml")
def test_articles_author_upsert_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/articles_author_upsert_fail.yaml")
def test_articles_with_author_returning(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/articles_with_author_returning.yaml")
def test_author_one(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_one.yaml")
def test_author_with_articles_one(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_with_articles_one.yaml")
def test_author_upsert_one_update(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_upsert_one_update.yaml")
def test_author_upsert_one_no_update(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_upsert_one_no_update.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/insert/nested"
@use_mutation_fixtures
class TestGraphqlInsertViews:
def test_insert_view_author_simple(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_view_author_simple.yaml")
def test_insert_view_author_complex_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/insert_view_author_complex_fail.yaml")
def test_nested_insert_article_author_simple_view(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/nested_insert_article_author_simple_view.yaml")
def test_nested_insert_article_author_complex_view_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/nested_insert_article_author_complex_view_fail.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/insert/views"
@use_mutation_fixtures
class TestGraphqlUpdateBasic:
def test_set_author_name(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_set_name.yaml")
def test_empty_set_author(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_empty_set.yaml")
hge_ctx.may_skip_test_teardown = True
def test_set_person_details(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_set_details.yaml")
def test_person_id_inc(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_inc.yaml")
def test_no_operator_err(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_error_no_operator.yaml")
def test_column_in_multiple_operators(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/article_column_multiple_operators.yaml")
def test_author_by_pk(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_by_pk.yaml")
def test_author_by_pk_null(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_by_pk_null.yaml")
def test_numerics_inc(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/numerics_inc.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/update/basic"
@use_mutation_fixtures
class TestGraphqlUpdateJsonB:
def test_jsonb_append_object(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_append_object.yaml")
def test_jsonb_append_array(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_append_array.yaml")
def test_jsonb_prepend_array(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_prepend_array.yaml")
def test_jsonb_delete_at_path(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_delete_at_path.yaml")
def test_jsonb_delete_array_element(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_delete_array_element.yaml")
def test_jsonb_delete_key(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/person_delete_key.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/update/jsonb"
@use_mutation_fixtures
class TestGraphqlUpdatePermissions:
def test_user_update_author(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_update_author.yaml")
def test_user_can_update_unpublished_article(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_can_update_unpublished_article.yaml")
def test_user_cannot_update_published_version_col(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_cannot_update_published_article_version.yaml")
def test_user_cannot_update_another_users_article(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_cannot_update_another_users_article.yaml")
def test_user_cannot_publish(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_cannot_publish.yaml")
def test_user_cannot_update_id_col(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_cannot_update_id_col_article.yaml")
def test_user_update_resident_preset(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + '/user_update_resident_preset.yaml')
def test_user_update_resident_preset_session_var(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + '/user_update_resident_preset_session_var.yaml')
def test_user_account_update_success(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + '/user_account_update_success.yaml')
def test_user_account_update_no_rows(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + '/user_account_update_no_rows.yaml')
@classmethod
def dir(cls):
return "queries/graphql_mutation/update/permissions"
@pytest.mark.parametrize("transport", ['http', 'websocket'])
@use_mutation_fixtures
class TestGraphqlDeleteBasic:
def test_article_delete(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + "/article.yaml", transport)
def test_article_delete_returning(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + "/article_returning.yaml", transport)
def test_article_delete_returning_author(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + "/article_returning_author.yaml", transport)
def test_author_returning_empty_articles(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + "/author_returning_empty_articles.yaml", transport)
def test_article_by_pk(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + "/article_by_pk.yaml", transport)
def test_article_by_pk_null(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + "/article_by_pk_null.yaml", transport)
@classmethod
def dir(cls):
return "queries/graphql_mutation/delete/basic"
@pytest.mark.parametrize("transport", ['http', 'websocket'])
@use_mutation_fixtures
class TestGraphqlDeleteConstraints:
def test_author_delete_foreign_key_violation(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + "/author_foreign_key_violation.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/delete/constraints"
@use_mutation_fixtures
class TestGraphqlDeletePermissions:
def test_author_can_delete_his_articles(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_can_delete_his_articles.yaml")
def test_author_cannot_delete_other_users_articles(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/author_cannot_delete_other_users_articles.yaml")
def test_resident_delete_without_select_perm_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/resident_delete_without_select_perm_fail.yaml")
def test_agent_delete_perm_arr_sess_var(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/agent_delete_perm_arr_sess_var.yaml")
def test_agent_delete_perm_arr_sess_var_fail(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/agent_delete_perm_arr_sess_var_fail.yaml")
def test_user_delete_author(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_delete_author.yaml")
def test_user_delete_account_success(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_delete_account_success.yaml")
def test_user_delete_account_no_rows(self, hge_ctx):
check_query_f(hge_ctx, self.dir() + "/user_delete_account_no_rows.yaml")
@classmethod
def dir(cls):
return "queries/graphql_mutation/delete/permissions"
@pytest.mark.parametrize("transport", ['http', 'websocket'])
@use_mutation_fixtures
class TestGraphqlMutationCustomSchema:
def test_insert_author(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/insert_author.yaml', transport)
def test_insert_article_author(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/insert_article_author.yaml', transport)
def test_update_article(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/update_article.yaml', transport)
def test_delete_article(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/delete_article.yaml', transport)
@classmethod
def dir(cls):
return "queries/graphql_mutation/custom_schema"
@pytest.mark.parametrize("transport", ['http', 'websocket'])
@use_mutation_fixtures
class TestGraphqlMutationCustomGraphQLTableName:
def test_insert_author(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/insert_author_details.yaml', transport)
def test_insert_article_author(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/insert_article_author.yaml', transport)
def test_update_author(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/update_author_details.yaml', transport)
def test_delete_author(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/delete_author_details.yaml', transport)
@classmethod
def dir(cls):
return "queries/graphql_mutation/custom_schema/custom_table_name"
@pytest.mark.parametrize('transport', ['http', 'websocket'])
@use_mutation_fixtures
class TestGraphQLMutateEnums:
@classmethod
def dir(cls):
return 'queries/graphql_mutation/enums'
def test_insert_enum_field(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/insert_enum_field.yaml', transport)
def test_insert_nullable_enum_field(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/insert_nullable_enum_field.yaml', transport)
def test_insert_enum_field_bad_value(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/insert_enum_field_bad_value.yaml', transport)
def test_update_enum_field(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/update_enum_field.yaml', transport)
def test_update_where_enum_field(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/update_where_enum_field.yaml', transport)
def test_delete_where_enum_field(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/delete_where_enum_field.yaml', transport)
use_function_permission_fixtures = usefixtures(
'per_class_db_schema_for_mutation_tests',
'per_method_db_data_for_mutation_tests',
'functions_permissions_fixtures'
)
# Tracking VOLATILE SQL functions as mutations, or queries (#1514)
@pytest.mark.parametrize('transport', ['http', 'websocket'])
@use_function_permission_fixtures
class TestGraphQLMutationFunctions:
@classmethod
def dir(cls):
return 'queries/graphql_mutation/functions'
# basic test that functions are added in the right places in the schema:
def test_smoke(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/smoke.yaml', transport)
# separate file since we this only works over http transport:
def test_smoke_errs(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/smoke_errs.yaml', 'http')
# Test tracking a VOLATILE function as top-level field of mutation root
# field, also smoke testing basic permissions on the table return type.
def test_functions_as_mutations(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/function_as_mutations.yaml', transport)
# When graphql-engine is started with `--infer-function-permissions=false` then
# a function is only accessible to a role when the permission is granted through
# the `pg_create_function_permission` definition
def test_function_as_mutation_without_function_permission(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/function_without_function_permission.yaml')
# Ensure select permissions on the corresponding SETOF table apply to
# the return set of the mutation field backed by the tracked function.
def test_functions_as_mutations_permissions(self, hge_ctx, transport):
st_code, resp = hge_ctx.v1metadataq_f(self.dir() + '/create_function_permission_add_to_score.yaml')
assert st_code == 200, resp
check_query_f(hge_ctx, self.dir() + '/function_as_mutations_permissions.yaml', transport)
st_code, resp = hge_ctx.v1metadataq_f(self.dir() + '/drop_function_permission_add_to_score.yaml')
assert st_code == 200, resp
def test_single_row_function_as_mutation(self, hge_ctx, transport):
st_code, resp = hge_ctx.v1metadataq_f(self.dir() + '/create_function_permission_add_to_score_by_user_id.yaml')
assert st_code == 200, resp
check_query_f(hge_ctx, self.dir() + '/single_row_function_as_mutation.yaml', transport)
st_code, resp = hge_ctx.v1metadataq_f(self.dir() + '/drop_function_permission_add_to_score_by_user_id.yaml')
assert st_code == 200, resp
@pytest.mark.parametrize('transport', ['http', 'websocket'])
@use_inherited_roles_fixtures
class TestGraphQLInheritedRoles:
@classmethod
def dir(cls):
return 'queries/graphql_mutation/insert/permissions/inherited_roles'
# This test exists here as a sanity check to check if mutations aren't exposed
# to an inherited role. When mutations are supported for everything, this test
# should be removed/modified.
def test_mutations_not_exposed_for_inherited_roles(self, hge_ctx, transport):
check_query_f(hge_ctx, self.dir() + '/mutation_not_exposed_to_inherited_roles.yaml')
Reference in New Issue View Git Blame Copy Permalink