Viabiliza filtro por valores mascarados em campos mascarados

core/models.py

@@ -14,6 +14,7 @@
 from django.contrib.postgres.search import SearchQuery, SearchRank, SearchVectorField
 from django.db import connection, models, transaction
 from django.db.models import F
+from django.db.models.functions import Substr
 from django.db.models.signals import post_delete, pre_delete
 from django.db.utils import ProgrammingError
 from django.urls import reverse
@@ -94,7 +95,27 @@ def apply_filters(self, filtering):
         # settings.
         model_filtering = self.model.extra["filtering"]
         processor = DynamicModelFilterProcessor(filtering, model_filtering)
-        return self.filter(**processor.filters)
+        kwargs = processor.filters
+
+        qs = self
+        for field in [f for f in self.model.extra["obfuscated"] if f in kwargs]:
+            value = kwargs[field]
+            if "*" not in value:
+                continue
+
+            # filter by obfuscated values such as CPFs ***34561****
+            kwargs.pop(field)
+            clean_value = value.replace("*", "")
+            start = value.index(clean_value) + 1
+
+            field_substr = f"{field}_substr"
+            annotate_kwargs = {
+                field_substr: Substr(field, start, len(clean_value)),
+            }
+            filter_kwargs = {field_substr: clean_value}
+            qs = qs.annotate(**annotate_kwargs).filter(**filter_kwargs)
+
+        return qs.filter(**kwargs)
 
     def apply_ordering(self, query):
         qs = self
@@ -410,6 +431,7 @@ def get_model(self, cache=True, data_table=None):
         # in DYNAMIC_MODEL_REGISTRY and not cache)
         fields = {field.name: field.field_class for field in self.fields}
         fields["search_data"] = SearchVectorField(null=True)
+        obfuscated = [f.name for f in self.fields if f.obfuscate]
         ordering = self.ordering or []
         filtering = self.filtering or []
         search = self.search or []
@@ -439,6 +461,7 @@ def get_model(self, cache=True, data_table=None):
         Model.extra = {
             "filtering": filtering,
             "ordering": ordering,
+            "obfuscated": obfuscated,
             "search": search,
             "table": self,
         }

core/tests/test_models.py

@@ -10,6 +10,7 @@
 
 from core.dynamic_models import DynamicModelMixin
 from core.models import Dataset, DataTable, Field, Table, TableFile, Version
+from core.tests.utils import BaseTestCaseWithSampleDataset
 from utils.file_info import human_readable_size
 
 
@@ -265,3 +266,49 @@ def test_return_empty_list_if_no_visible_table(self):
         Table.objects.all().update(hidden=True)
 
         assert [] == self.dataset.all_files
+
+
+class DynamicTableModelTest(BaseTestCaseWithSampleDataset):
+    DATASET_SLUG = "sample"
+    TABLE_NAME = "sample_table"
+    FIELDS_KWARGS = [
+        {
+            "name": "sample_field",
+            "options": {"max_length": 10},
+            "type": "text",
+            "null": False,
+            "filtering": True,
+            "choices": {"data": ["foo", "bar"]},
+        },
+        {
+            "name": "obfuscated_field",
+            "options": {"max_length": 10},
+            "type": "text",
+            "null": False,
+            "filtering": True,
+            "choices": {},
+            "obfuscate": True,
+        },
+    ]
+
+    def test_filter_by_obfuscate_exact_value(self):
+        value = "123456"
+        entry = baker.make(self.TableModel, obfuscated_field=value)
+        baker.make(self.TableModel, obfuscated_field="other")
+
+        query = {"obfuscated_field": value}
+        qs = self.TableModel.objects.apply_filters(query)
+
+        assert 1 == qs.count()
+        assert entry in qs
+
+    def test_filter_by_obfuscate_partial_value(self):
+        value = "123456"
+        entry = baker.make(self.TableModel, obfuscated_field=value)
+        baker.make(self.TableModel, obfuscated_field="other")
+
+        query = {"obfuscated_field": "**34**"}
+        qs = self.TableModel.objects.apply_filters(query)
+
+        assert 1 == qs.count()
+        assert entry in qs