Skip to content

Commit

Permalink
tests(merge attributes): update tests for replacement of view template
Browse files Browse the repository at this point in the history
Signed-off-by: David Wallace <[email protected]>
  • Loading branch information
MyPyDavid committed Jun 18, 2024
1 parent 13506f7 commit b99df98
Showing 1 changed file with 114 additions and 44 deletions.
158 changes: 114 additions & 44 deletions rdmo/management/tests/test_merge_attributes.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,14 @@
if i.is_relation and not i.many_to_many and i.related_model is not Attribute]

EXAMPLE_URI_PREFIX = 'http://example.com/terms'
foo_merge_uri_prefix = 'http://foo-merge.com/terms'
bar_merge_uri_prefix = 'http://bar-merge.com/terms'
FOO_MERGE_URI_PREFIX = 'http://foo-merge.com/terms'
BAR_MERGE_URI_PREFIX = 'http://bar-merge.com/terms'
EXAMPLE_VIEW_URI_PATH = "views/view_a"
merge_view_template_addition_uri_path = 'individual/single/textarea'
merge_view_template_addition = Template("{% render_value '$new_uri' %}")
EXAMPLE_VIEW_URI = Attribute.build_uri(EXAMPLE_URI_PREFIX, merge_view_template_addition_uri_path)
new_merge_uri_prefixes = [foo_merge_uri_prefix, bar_merge_uri_prefix]
VIEW_TEMPLATE_URI_PATH = 'individual/single/textarea'
VIEW_TEMPLATE_URI_PATH_ADDITIONS = ['', '/test1', '/test2', '/testfoo']
VIEW_TEMPLATE_RENDER_VALUE = Template("{% render_value '$new_uri' %}")
EXAMPLE_VIEW_URI = Attribute.build_uri(EXAMPLE_URI_PREFIX, VIEW_TEMPLATE_URI_PATH)
NEW_MERGE_URI_PREFIXES = [FOO_MERGE_URI_PREFIX, BAR_MERGE_URI_PREFIX]

def _prepare_instance_for_copy(instance, uri_prefix=None, uri_path=None):
instance.pk = None
Expand All @@ -43,23 +44,32 @@ def _get_queryset(related_field, attribute=None):
lookup_field = related_field.remote_field.name
return model.objects.filter(**{lookup_field: attribute})

def _add_new_line_to_view_template(instance: View, new_uri_prefix: str) -> View:
current_field_value = instance.template
new_field_value = current_field_value + "\n"
new_uri = Attribute.build_uri(new_uri_prefix, merge_view_template_addition_uri_path)
new_field_value += merge_view_template_addition.substitute(new_uri=new_uri)
instance.template = new_field_value
return instance
def create_new_uris_with_uri_prefix_for_template(new_uri_prefix: str) -> List[str]:
new_uris = []
for extra_path in VIEW_TEMPLATE_URI_PATH_ADDITIONS:
new_uri_path = VIEW_TEMPLATE_URI_PATH + extra_path
new_uri = Attribute.build_uri(new_uri_prefix, new_uri_path)
new_uris.append(new_uri)
return new_uris

def create_copy_of_view_that_uses_new_attribute(new_prefixes: List[str]):
def create_copy_of_view_that_uses_new_attribute(db, new_prefixes: List[str]):
# TODO search in View.template for the attribute uri
# example_uri = f"{EXAMPLE_URI_PREFIX}/{EXAMPLE_VIEW_URI_PATH}"
# source = Attribute.objects.get(uri=example_uri)
qs = View.objects.filter(**{"template__contains": EXAMPLE_VIEW_URI_PATH})
for new_prefix in new_prefixes:
for instance in qs:
instance = _prepare_instance_for_copy(instance)
instance = _add_new_line_to_view_template(instance, new_prefix)
qs = View.objects.filter(**{"uri__contains": EXAMPLE_VIEW_URI_PATH}).all()
if not qs.exists():
raise ValueError("Views for tests should exist here.")
for instance in qs:
original_template = instance.template
for new_prefix in new_prefixes:
instance = _prepare_instance_for_copy(instance, uri_prefix=new_prefix, uri_path=EXAMPLE_VIEW_URI_PATH)
new_template_uris = create_new_uris_with_uri_prefix_for_template(new_prefix)
new_template = ''
new_template += original_template
for uri in new_template_uris:
new_template += '\n'
new_template += VIEW_TEMPLATE_RENDER_VALUE.substitute(new_uri=uri)
instance.template = new_template
instance.save()

def create_copies_of_related_models_with_new_uri_prefix(new_prefixes):
Expand Down Expand Up @@ -103,16 +113,16 @@ def get_related_affected_instances(attribute) -> list:


@pytest.fixture
def create_merge_attributes(db, settings):
def create_new_merge_attributes_and_views(db, settings):
""" Creates model instances for merge attributes tests """
example_attributes = Attribute.objects.filter(uri_prefix=EXAMPLE_URI_PREFIX).all()
create_copies_of_attributes_with_new_uri_prefix(example_attributes, new_merge_uri_prefixes)
create_copies_of_related_models_with_new_uri_prefix(new_merge_uri_prefixes)
create_copy_of_view_that_uses_new_attribute(new_merge_uri_prefixes)
create_copies_of_attributes_with_new_uri_prefix(example_attributes, NEW_MERGE_URI_PREFIXES)
create_copies_of_related_models_with_new_uri_prefix(NEW_MERGE_URI_PREFIXES)
create_copy_of_view_that_uses_new_attribute(db, NEW_MERGE_URI_PREFIXES)


@pytest.mark.parametrize('uri_prefix', new_merge_uri_prefixes)
def test_that_the_freshly_created_merge_attributes_are_present(create_merge_attributes, uri_prefix):
@pytest.mark.parametrize('uri_prefix', NEW_MERGE_URI_PREFIXES)
def test_that_the_freshly_created_merge_attributes_are_present(db, create_new_merge_attributes_and_views, uri_prefix):
merge_attributes = Attribute.objects.filter(uri_prefix=uri_prefix).all()
assert len(merge_attributes) > 2
unique_uri_prefixes = set(Attribute.objects.values_list("uri_prefix", flat=True))
Expand All @@ -127,12 +137,25 @@ def test_that_the_freshly_created_merge_attributes_are_present(create_merge_attr
models_qs = [_get_queryset(i, attribute=attribute) for i in ATTRIBUTE_RELATED_MODELS_FIELDS]
assert any(len(i) > 0 for i in models_qs)

# assert new views created by create_copy_of_view_that_uses_new_attribute
# foo-merge
foo_merge_view_qs = View.objects.filter(template__contains=FOO_MERGE_URI_PREFIX).exclude(
template__contains=BAR_MERGE_URI_PREFIX)
assert foo_merge_view_qs.count() == 1
assert foo_merge_view_qs.first().uri_prefix == FOO_MERGE_URI_PREFIX
# bar-merge
bar_merge_view_qs = View.objects.filter(template__contains=BAR_MERGE_URI_PREFIX).exclude(
template__contains=FOO_MERGE_URI_PREFIX)
assert bar_merge_view_qs.count() == 1
assert bar_merge_view_qs.first().uri_prefix == BAR_MERGE_URI_PREFIX

@pytest.mark.parametrize('source_uri_prefix', new_merge_uri_prefixes)

@pytest.mark.parametrize('source_uri_prefix', NEW_MERGE_URI_PREFIXES)
@pytest.mark.parametrize('save', [False, True])
@pytest.mark.parametrize('delete', [False, True])
@pytest.mark.parametrize('view', [False, True])
def test_command_merge_attributes(create_merge_attributes, source_uri_prefix, save, delete, view):
def test_command_merge_attributes(db, settings, create_new_merge_attributes_and_views,
source_uri_prefix, save, delete, view):
source_attributes = Attribute.objects.filter(uri_prefix=source_uri_prefix).all()
assert len(source_attributes) > 2
unique_uri_prefixes = set(Attribute.objects.values_list("uri_prefix", flat=True))
Expand All @@ -143,7 +166,6 @@ def test_command_merge_attributes(create_merge_attributes, source_uri_prefix, sa
target_attribute = Attribute.objects.get(uri_prefix=EXAMPLE_URI_PREFIX, path=source_attribute.path)
stdout, stderr = io.StringIO(), io.StringIO()
before_source_related_qs = get_related_affected_instances(source_attribute)
before_source_related_view_qs = View.objects.filter(**{"template__contains": source_attribute.path})
# before_target_related_qs = get_related_affected_instances(target_attribute)

command_kwargs = {'source': source_attribute.uri,
Expand Down Expand Up @@ -172,30 +194,78 @@ def test_command_merge_attributes(create_merge_attributes, source_uri_prefix, sa
else:
assert Attribute.objects.get(id=source_attribute.id)


after_source_related_qs = get_related_affected_instances(source_attribute)
after_source_related_view_qs = View.objects.filter(**{"template__contains": source_attribute.path})
after_target_related_qs = get_related_affected_instances(target_attribute)
# after_target_related_view_qs = View.objects.filter(**{"template__contains": target_attribute.path})

if save and not failed:

if any(i.exists() for i in before_source_related_qs):
assert not any(i.exists() for i in after_source_related_qs)
assert any(i.exists() for i in after_target_related_qs)

if source_attribute.path in merge_view_template_addition_uri_path:
# assert that the attribute in the view template was replaced as well
# the EXAMPLE_VIEW_URI is from the target attribute
# uri_prefix = source_uri_prefix, uri_path = EXAMPLE_VIEW_URI_PATH
if before_source_related_view_qs.exists():
if view and source_attribute.path != target_attribute.path:
assert not after_source_related_view_qs.exists()
for instance in after_source_related_view_qs:
assert any(target_attribute.path in i for i in instance.template.splitlines())
assert not any(source_attribute.path in i for i in instance.template.splitlines())
else:
assert after_source_related_view_qs.exists()

else:
if any(i.exists() for i in before_source_related_qs):
assert any(i.exists() for i in after_source_related_qs)


@pytest.mark.parametrize('source_uri_prefix', NEW_MERGE_URI_PREFIXES)
@pytest.mark.parametrize('save', [False, True])
@pytest.mark.parametrize('delete', [False, True])
@pytest.mark.parametrize('view', [False, True])
def test_command_merge_attributes_for_views(db, settings, create_new_merge_attributes_and_views,
source_uri_prefix, save, delete, view):
source_attributes = Attribute.objects.filter(uri_prefix=source_uri_prefix).all()
assert len(source_attributes) > 2
unique_uri_prefixes = set(Attribute.objects.values_list("uri_prefix", flat=True))
# test that the currently selected uri_prefix is in db
assert source_uri_prefix in unique_uri_prefixes
source_attribute_uri = Attribute.build_uri(uri_prefix=source_uri_prefix, path=VIEW_TEMPLATE_URI_PATH)
target_attribute_uri = Attribute.build_uri(uri_prefix=EXAMPLE_URI_PREFIX, path=VIEW_TEMPLATE_URI_PATH)

stdout, stderr = io.StringIO(), io.StringIO()
before_source_related_view_uri_qs = View.objects.filter(**{"template__contains": f"'{source_attribute_uri}'"}).all()

before_source_related_view_uri_templates = {i.uri: i.template for i in before_source_related_view_uri_qs}

command_kwargs = {'source': source_attribute_uri,
'target': target_attribute_uri,
'save': save, 'delete': delete, 'view': view}
failed = False
call_command('merge_attributes',
stdout=stdout, stderr=stderr, **command_kwargs)

if delete and not failed:
# assert that the source attribut was deleted
with pytest.raises(Attribute.DoesNotExist):
Attribute.objects.get(uri=source_attribute_uri)
else:
assert Attribute.objects.get(uri=source_attribute_uri)

after_source_related_view_uri_qs = View.objects.filter(**{"template__contains": f"'{source_attribute_uri}'"})
after_target_related_view_uri_qs = View.objects.filter(**{"template__contains": f"'{target_attribute_uri}'"})

if not save or not view:
assert not after_target_related_view_uri_qs.exists()
return

if save and not failed:
pass


if (save and not failed and view
and VIEW_TEMPLATE_URI_PATH in source_attribute_uri):
# assert that the attribute in the view template was replaced as well
# the EXAMPLE_VIEW_URI is from the target attribute
# uri_prefix = source_uri_prefix, uri_path = EXAMPLE_VIEW_URI_PATH
assert not after_source_related_view_uri_qs.exists()
assert after_target_related_view_uri_qs.exists()

if (before_source_related_view_uri_templates and
source_attribute_uri != target_attribute_uri):

after_occurrences = list(filter(lambda x:target_attribute_uri in x,
after_target_related_view_uri_qs.first().template.splitlines()))
assert len(after_occurrences) == 1
before_occurrences = list(filter(lambda x: target_attribute_uri in x,
before_source_related_view_uri_qs.first().template.splitlines()))
assert len(before_occurrences) == 0

0 comments on commit b99df98

Please sign in to comment.