from django.contrib.contenttypes.models import ContentType import pytest from pytest_unordered import unordered from baserow.contrib.database.fields.dependencies.exceptions import ( SelfReferenceFieldDependencyError, ) from baserow.contrib.database.fields.dependencies.handler import FieldDependencyHandler from baserow.contrib.database.fields.dependencies.models import FieldDependency from baserow.contrib.database.fields.field_cache import FieldCache from baserow.contrib.database.fields.handler import FieldHandler from baserow.contrib.database.fields.models import LinkRowField from baserow.contrib.database.fields.registries import field_type_registry @pytest.mark.django_db def test_get_same_table_deps(data_fixture): field_a = data_fixture.create_text_field() field_b = data_fixture.create_text_field(table=field_a.table) field_c = data_fixture.create_text_field(table=field_a.table) field_in_other_table = data_fixture.create_text_field() FieldDependency.objects.create(dependant=field_a, dependency=field_b) FieldDependency.objects.create(dependant=field_b, dependency=field_c) FieldDependency.objects.create(dependant=field_c, dependency=field_in_other_table) assert FieldDependencyHandler().get_same_table_dependencies(field_a) == [field_b] assert FieldDependencyHandler().get_same_table_dependencies(field_b) == [field_c] assert FieldDependencyHandler().get_same_table_dependencies(field_c) == [] def when_field_updated(field, via=None, relation_changed=True): result = [] for ( dependant_field, dependant_field_type, via_path_to_starting_table, ) in field.dependant_fields_with_types( field_cache=FieldCache(), starting_via_path_to_starting_table=via, associated_relation_changed=relation_changed, ): if via_path_to_starting_table is None: via_path_to_starting_table = [] result_dict = { "field": dependant_field, "via": via_path_to_starting_table, } then = when_field_updated( dependant_field, via=via_path_to_starting_table, relation_changed=False ) if then: result_dict["then"] = then result.append(result_dict) return result @pytest.mark.django_db def test_dependencies_for_primary_lookup(data_fixture): user = data_fixture.create_user() table_a, table_b, table_a_link_field = data_fixture.create_two_linked_tables( user=user ) table_a_primary = table_a.field_set.get(primary=True) table_b_link_field = table_a_link_field.link_row_related_field table_b_primary = table_b.field_set.get(primary=True) FieldHandler().update_field( user, table_a_primary.specific, new_type_name="formula", formula=f"lookup('{table_a_link_field.name}', '{table_b_primary.name}')", ) assert when_field_updated(table_b_primary) == causes( a_field_update_for(table_a_link_field, via=[table_a_link_field]), a_field_update_for( table_a_primary.specific, via=[table_a_link_field], then=causes( a_field_update_for( table_b_link_field, via=[table_a_link_field, table_b_link_field], ) ), ), ) assert when_field_updated(table_a_primary) == causes( a_field_update_for(table_b_link_field, via=[table_b_link_field]), ) assert when_field_updated(table_b_link_field) == causes( a_field_update_for(table_a_link_field, via=[table_a_link_field]), a_field_update_for( table_a_primary.specific, via=[table_a_link_field], then=causes( a_field_update_for( table_b_link_field, via=[table_a_link_field, table_b_link_field], ) ), ), ) assert when_field_updated(table_a_link_field) == causes( a_field_update_for(table_b_link_field, via=[table_b_link_field]), a_field_update_for( table_a_primary.specific, via=[], then=causes( a_field_update_for( table_b_link_field, via=[table_b_link_field], ) ), ), ) @pytest.mark.django_db def test_dependencies_for_triple_lookup(data_fixture): user = data_fixture.create_user() table_a, table_b, table_a_to_b_link_field = data_fixture.create_two_linked_tables( user=user ) table_c, _, table_c_to_b_link_field = data_fixture.create_two_linked_tables( user=user, table_b=table_b ) table_a_primary = table_a.field_set.get(primary=True).specific table_b_to_a_link_field = table_a_to_b_link_field.link_row_related_field table_b_to_c_link_field = table_c_to_b_link_field.link_row_related_field table_b_primary = table_b.field_set.get(primary=True).specific table_c_primary = table_c.field_set.get(primary=True).specific lookup_of_linked_field = FieldHandler().create_field( user, table=table_a, type_name="formula", name="lookup_of_link_field", formula=f"lookup('{table_a_to_b_link_field.name}', " f"'{table_b_to_c_link_field.name}')", ) assert lookup_of_linked_field.error is None assert when_field_updated(table_c_primary) == causes( a_field_update_for( table_b_to_c_link_field, via=[table_b_to_c_link_field], then=causes( a_field_update_for( lookup_of_linked_field, via=[table_b_to_c_link_field, table_a_to_b_link_field], ) ), ) ) assert when_field_updated(table_b_to_c_link_field) == causes( a_field_update_for( table_c_to_b_link_field, via=[table_c_to_b_link_field], ), a_field_update_for( lookup_of_linked_field, via=[table_a_to_b_link_field], ), ) @pytest.mark.django_db @pytest.mark.field_link_row def test_dependencies_for_link_row_link_row_self_reference(data_fixture): user = data_fixture.create_user() table_a = data_fixture.create_database_table(user=user) table_a_primary = data_fixture.create_text_field(table=table_a, primary=True) table_a_self_link = FieldHandler().create_field( user, table=table_a, type_name="link_row", name="self", link_row_table=table_a, ) assert when_field_updated(table_a_primary) == causes( a_field_update_for(field=table_a_self_link, via=[table_a_self_link]) ) @pytest.mark.django_db def test_self_reference_raises(data_fixture): user = data_fixture.create_user() table_a = data_fixture.create_database_table(user=user) with pytest.raises(SelfReferenceFieldDependencyError): FieldHandler().create_field( user, table=table_a, type_name="formula", name="self_ref", formula="field('self_ref')", ) def causes(*result): return list(result) def a_field_update_for(field, via, then=None): r = { "field": field, "via": via, } if then: r["then"] = then return r @pytest.mark.django_db @pytest.mark.field_link_row def test_get_all_dependant_fields_with_type(data_fixture): table = data_fixture.create_database_table() text_field_1 = data_fixture.create_text_field(table=table) text_field_2 = data_fixture.create_text_field(table=table) text_field_3 = data_fixture.create_text_field(table=table) text_field_1_dependency_1 = data_fixture.create_text_field(table=table) text_field_1_dependency_2 = data_fixture.create_text_field(table=table) text_field_2_dependency_1 = data_fixture.create_text_field(table=table) text_field_2_dependency_2 = data_fixture.create_text_field(table=table) text_field_2_dependency_3 = data_fixture.create_text_field(table=table) text_field_1_dependency_1_dependency_1 = data_fixture.create_text_field(table=table) text_field_1_dependency_2_dependency_1 = data_fixture.create_text_field(table=table) text_field_2_dependency_1_dependency_1 = data_fixture.create_text_field(table=table) link_field_to_table = data_fixture.create_link_row_field(link_row_table=table) text_field_1_and_2_dependency_1 = data_fixture.create_text_field(table=table) other_table_field = data_fixture.create_text_field( table=link_field_to_table.link_row_table ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_1 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_2 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_and_2_dependency_1 ) FieldDependency.objects.create( dependency=text_field_2, dependant=text_field_2_dependency_1 ) FieldDependency.objects.create( dependency=text_field_2, dependant=other_table_field, via=link_field_to_table, ) FieldDependency.objects.create( dependency=text_field_2, dependant=text_field_2_dependency_3 ) FieldDependency.objects.create( dependency=text_field_2, dependant=text_field_1_and_2_dependency_1 ) FieldDependency.objects.create( dependency=text_field_1_dependency_1, dependant=text_field_1_dependency_1_dependency_1, ) FieldDependency.objects.create( dependency=text_field_1_dependency_2, dependant=text_field_1_dependency_2_dependency_1, ) FieldDependency.objects.create( dependency=text_field_2_dependency_1, dependant=text_field_2_dependency_1_dependency_1, ) field_cache = FieldCache() text_field_type = field_type_registry.get_by_model(text_field_1_dependency_1) results = FieldDependencyHandler.get_all_dependent_fields_with_type( table.id, field_ids=[text_field_1.id], field_cache=field_cache, associated_relations_changed=True, ) expected_text_field_1_dependants = [ (text_field_1_dependency_1, text_field_type, None), (text_field_1_dependency_2, text_field_type, None), (text_field_1_and_2_dependency_1, text_field_type, None), (text_field_1_dependency_1_dependency_1, text_field_type, None), (text_field_1_dependency_2_dependency_1, text_field_type, None), ] assert results == unordered(expected_text_field_1_dependants) results = FieldDependencyHandler.get_all_dependent_fields_with_type( table.id, field_ids=[text_field_2.id], field_cache=field_cache, associated_relations_changed=True, ) expected_text_field_2_dependants = [ (text_field_2_dependency_1, text_field_type, None), (other_table_field, text_field_type, [link_field_to_table]), ( text_field_2_dependency_3, text_field_type, None, ), (text_field_1_and_2_dependency_1, text_field_type, None), (text_field_2_dependency_1_dependency_1, text_field_type, None), ] assert results == unordered(expected_text_field_2_dependants) results = FieldDependencyHandler.get_all_dependent_fields_with_type( table.id, field_ids=[text_field_3.id], field_cache=field_cache, associated_relations_changed=True, ) assert len(results) == 0 results = FieldDependencyHandler.get_all_dependent_fields_with_type( table.id, field_ids=[text_field_1.id, text_field_2.id, text_field_3.id], field_cache=field_cache, associated_relations_changed=True, ) expected_text_field_1_2_3_dependants = [ # This is a combination of `expected_text_field_1_dependants` and # `expected_text_field_2_dependants`, but without the duplicates. (text_field_1_dependency_1, text_field_type, None), (text_field_1_dependency_2, text_field_type, None), (text_field_1_dependency_1_dependency_1, text_field_type, None), (text_field_1_dependency_2_dependency_1, text_field_type, None), (text_field_2_dependency_1, text_field_type, None), (other_table_field, text_field_type, [link_field_to_table]), ( text_field_2_dependency_3, text_field_type, None, ), (text_field_1_and_2_dependency_1, text_field_type, None), (text_field_2_dependency_1_dependency_1, text_field_type, None), ] assert results == unordered(expected_text_field_1_2_3_dependants) @pytest.mark.django_db @pytest.mark.field_link_row def test_get_all_dependant_fields_with_type_num_queries( data_fixture, django_assert_num_queries ): table = data_fixture.create_database_table() text_field_1 = data_fixture.create_text_field(table=table) text_field_1_dependency_1 = data_fixture.create_text_field(table=table) text_field_1_dependency_2 = data_fixture.create_number_field(table=table) text_field_1_and_2_dependency_1 = data_fixture.create_text_field(table=table) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_1 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_2 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_and_2_dependency_1 ) # This content type is fetched in the method, but it's also cached after. # Fetching it in the test, ensures that we see a correct number of queries after. ContentType.objects.get_for_model(LinkRowField) with django_assert_num_queries(3): FieldDependencyHandler.get_all_dependent_fields_with_type( table.id, field_ids=[text_field_1.id], field_cache=FieldCache(), associated_relations_changed=True, ) text_field_1_dependency_3 = data_fixture.create_text_field(table=table) text_field_1_dependency_4 = data_fixture.create_number_field(table=table) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_3 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_4 ) with django_assert_num_queries(3): FieldDependencyHandler.get_all_dependent_fields_with_type( table.id, field_ids=[text_field_1.id], field_cache=FieldCache(), associated_relations_changed=True, ) @pytest.mark.django_db @pytest.mark.field_link_row def test_get_all_dependant_fields_with_type_via_field_num_queries( data_fixture, django_assert_num_queries ): table = data_fixture.create_database_table() table2 = data_fixture.create_database_table() text_field_1 = data_fixture.create_text_field(table=table) text_field_2 = data_fixture.create_text_field(table=table2) link_field_to_table = data_fixture.create_link_row_field(table=table) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_2, via=link_field_to_table, ) with django_assert_num_queries(3): dependant_fields = FieldDependencyHandler.get_all_dependent_fields_with_type( table.id, field_ids=[text_field_1.id], field_cache=FieldCache(), associated_relations_changed=True, ) ( dependant_field, dependant_field_type, path_to_starting_table, ) = dependant_fields[0] str(dependant_field.table.id) str(path_to_starting_table[0].table.id) str(path_to_starting_table[0].link_row_table.id) @pytest.mark.django_db @pytest.mark.field_link_row def test_get_dependant_fields_with_type(data_fixture): table = data_fixture.create_database_table() text_field_1 = data_fixture.create_text_field(table=table) text_field_2 = data_fixture.create_text_field(table=table) text_field_3 = data_fixture.create_text_field(table=table) text_field_1_dependency_1 = data_fixture.create_text_field(table=table) text_field_1_dependency_2 = data_fixture.create_text_field(table=table) text_field_2_dependency_1 = data_fixture.create_text_field(table=table) text_field_2_dependency_2 = data_fixture.create_text_field(table=table) text_field_2_dependency_3 = data_fixture.create_text_field(table=table) link_field_to_table = data_fixture.create_link_row_field(link_row_table=table) text_field_1_and_2_dependency_1 = data_fixture.create_text_field(table=table) other_table_field = data_fixture.create_text_field( table=link_field_to_table.link_row_table ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_1 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_2 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_and_2_dependency_1 ) FieldDependency.objects.create( dependency=text_field_2, dependant=text_field_2_dependency_1 ) FieldDependency.objects.create( dependency=text_field_2, dependant=other_table_field, via=link_field_to_table, ) FieldDependency.objects.create( dependency=text_field_2, dependant=text_field_2_dependency_3 ) FieldDependency.objects.create( dependency=text_field_2, dependant=text_field_1_and_2_dependency_1 ) field_cache = FieldCache() text_field_type = field_type_registry.get_by_model(text_field_1_dependency_1) results = FieldDependencyHandler.get_dependant_fields_with_type( table.id, field_ids=[text_field_1.id], associated_relations_changed=True, field_cache=field_cache, ) expected_text_field_1_dependants = [ (text_field_1_dependency_1, text_field_type, None), (text_field_1_dependency_2, text_field_type, None), (text_field_1_and_2_dependency_1, text_field_type, None), ] assert results == unordered(expected_text_field_1_dependants) results = FieldDependencyHandler.get_dependant_fields_with_type( table.id, field_ids=[text_field_2.id], associated_relations_changed=True, field_cache=field_cache, ) expected_text_field_2_dependants = [ (text_field_2_dependency_1, text_field_type, None), (other_table_field, text_field_type, [link_field_to_table]), ( text_field_2_dependency_3, text_field_type, None, ), (text_field_1_and_2_dependency_1, text_field_type, None), ] assert results == unordered(expected_text_field_2_dependants) results = FieldDependencyHandler.get_dependant_fields_with_type( table.id, field_ids=[text_field_3.id], associated_relations_changed=True, field_cache=field_cache, ) assert len(results) == 0 results = FieldDependencyHandler.get_dependant_fields_with_type( table.id, field_ids=[text_field_1.id, text_field_2.id, text_field_3.id], associated_relations_changed=True, field_cache=field_cache, ) assert results == unordered( expected_text_field_1_dependants + expected_text_field_2_dependants ) @pytest.mark.django_db @pytest.mark.field_link_row def test_get_dependant_fields_with_type_num_queries( data_fixture, django_assert_num_queries ): table = data_fixture.create_database_table() text_field_1 = data_fixture.create_text_field(table=table) text_field_1_dependency_1 = data_fixture.create_text_field(table=table) text_field_1_dependency_2 = data_fixture.create_number_field(table=table) text_field_1_and_2_dependency_1 = data_fixture.create_text_field(table=table) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_1 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_2 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_and_2_dependency_1 ) with django_assert_num_queries(3): FieldDependencyHandler.get_dependant_fields_with_type( table.id, field_ids=[text_field_1.id], associated_relations_changed=True, field_cache=FieldCache(), ) text_field_1_dependency_3 = data_fixture.create_text_field(table=table) text_field_1_dependency_4 = data_fixture.create_number_field(table=table) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_3 ) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_1_dependency_4 ) with django_assert_num_queries(3): FieldDependencyHandler.get_dependant_fields_with_type( table.id, field_ids=[text_field_1.id], associated_relations_changed=True, field_cache=FieldCache(), ) @pytest.mark.django_db @pytest.mark.field_link_row def test_get_dependant_fields_with_type_via_field_num_queries( data_fixture, django_assert_num_queries ): table = data_fixture.create_database_table() text_field_1 = data_fixture.create_text_field(table=table) text_field_2 = data_fixture.create_text_field(table=table) link_field_to_table = data_fixture.create_link_row_field(table=table) FieldDependency.objects.create( dependency=text_field_1, dependant=text_field_2, via=link_field_to_table, ) with django_assert_num_queries(2): dependant_fields = FieldDependencyHandler.get_dependant_fields_with_type( table.id, field_ids=[text_field_1.id], associated_relations_changed=True, field_cache=FieldCache(), ) ( dependant_field, dependant_field_type, path_to_starting_table, ) = dependant_fields[0] str(dependant_field.table.id) str(path_to_starting_table[0].table.id) str(path_to_starting_table[0].link_row_table.id)