3399 controlled access fra data files page

Taskfile.yml

@@ -238,7 +238,7 @@ tasks:
     desc: Run cypress tests
     dir: tdrs-frontend
     cmds:
-      - docker-compose -f docker-compose.local.yml up --build tdp-frontend-test -d
+      - docker compose -f docker-compose.local.yml up --build tdp-frontend-test -d
       - npm run test:e2e
 
   k6:

tdrs-backend/tdpservice/data_files/admin/admin.py

@@ -37,6 +37,16 @@ class Media:
 
     actions = ['reparse']
 
+    def get_queryset(self, request):
+        """Return the queryset."""
+        qs = super().get_queryset(request)
+        # return data files based on user's section
+        if not request.user.has_fra_access:
+            filtered_for_fra = qs.exclude(section__in=DataFile.get_fra_section_list())
+            return filtered_for_fra
+        else:
+            return qs
+
     def reparse(self, request, queryset):
         """Reparse the selected data files."""
         files = queryset.values_list("id", flat=True)
@@ -135,6 +145,28 @@ def queryset(self, request, queryset):
             else:
                 return queryset
 
+    class FRA_AccessFilter(admin.SimpleListFilter):
+        """Filter datafile based on user access to FRA files."""
+
+        title = 'FRA/Non FRA Files'
+        parameter_name = 'fra_access'
+
+        def lookups(self, request, model_admin):
+            """Return a list of tuples."""
+            return [
+                ('1', 'FRA Datafile'),
+                ('0', 'Non FRA Datafile'),
+            ]
+
+        def queryset(self, request, queryset):
+            """Return a queryset."""
+            if self.value() == '1':
+                return queryset.filter(section__in=DataFile.get_fra_section_list())
+            elif self.value() == '0':
+                return queryset.exclude(section__in=DataFile.get_fra_section_list())
+            else:
+                return queryset
+
     inlines = [DataFileInline]
 
     list_display = [
@@ -162,6 +194,16 @@ def queryset(self, request, queryset):
         VersionFilter,
     ]
 
+    def get_list_filter(self, request):
+        """Get filter list in DataFile admin page."""
+        list_filter = super().get_list_filter(request)
+        user = request.user
+        if (user.is_an_admin or user.has_fra_access) and self.FRA_AccessFilter not in list_filter:
+            list_filter.append(self.FRA_AccessFilter)
+        elif not user.has_fra_access and self.FRA_AccessFilter in list_filter:
+            list_filter.remove(self.FRA_AccessFilter)
+        return list_filter
+
 @admin.register(LegacyFileTransfer)
 class LegacyFileTransferAdmin(ReadOnlyAdminMixin, admin.ModelAdmin):
     """Admin class for LegacyFileTransfer models."""

tdrs-backend/tdpservice/data_files/migrations/0017_alter_datafile_section.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.15 on 2025-01-08 13:42
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('data_files', '0016_remove_datafile_reparse_meta_models'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='datafile',
+            name='section',
+            field=models.CharField(choices=[('Tribal Closed Case Data', 'Tribal Closed Case Data'), ('Tribal Active Case Data', 'Tribal Active Case Data'), ('Tribal Aggregate Data', 'Tribal Aggregate Data'), ('Tribal Stratum Data', 'Tribal Stratum Data'), ('SSP Aggregate Data', 'Ssp Aggregate Data'), ('SSP Closed Case Data', 'Ssp Closed Case Data'), ('SSP Active Case Data', 'Ssp Active Case Data'), ('SSP Stratum Data', 'Ssp Stratum Data'), ('Active Case Data', 'Active Case Data'), ('Closed Case Data', 'Closed Case Data'), ('Aggregate Data', 'Aggregate Data'), ('Stratum Data', 'Stratum Data'), ('Work Outcomes for TANF Exiters', 'Fra Work Outcome Tanf Exiters'), ('Secondary School Attainment', 'Fra Secondry School Attainment'), ('Supplemental Work Outcomes', 'Fra Supplement Work Outcomes')], max_length=32),
+        ),
+    ]

tdrs-backend/tdpservice/data_files/models.py

@@ -119,6 +119,58 @@ class Section(models.TextChoices):
         AGGREGATE_DATA = "Aggregate Data"
         STRATUM_DATA = "Stratum Data"
 
+        FRA_WORK_OUTCOME_TANF_EXITERS = "Work Outcomes for TANF Exiters"
+        FRA_SECONDRY_SCHOOL_ATTAINMENT = "Secondary School Attainment"
+        FRA_SUPPLEMENT_WORK_OUTCOMES = "Supplemental Work Outcomes"
+
+        @classmethod
+        def is_fra(cls, section: str) -> bool:
+            """Determine if the section is a FRA section."""
+            return section in [
+                cls.FRA_WORK_OUTCOME_TANF_EXITERS,
+                cls.FRA_SECONDRY_SCHOOL_ATTAINMENT,
+                cls.FRA_SUPPLEMENT_WORK_OUTCOMES
+            ]
+
+        @classmethod
+        def is_ssp(cls, section: str) -> bool:
+            """Determine if the section is an SSP section."""
+            return section in [
+                cls.SSP_AGGREGATE_DATA,
+                cls.SSP_ACTIVE_CASE_DATA,
+                cls.SSP_CLOSED_CASE_DATA,
+                cls.SSP_STRATUM_DATA
+            ]
+
+        @classmethod
+        def is_tribal(cls, section: str) -> bool:
+            """Determine if the section is a Tribal section."""
+            return section in [
+                cls.TRIBAL_AGGREGATE_DATA,
+                cls.TRIBAL_ACTIVE_CASE_DATA,
+                cls.TRIBAL_CLOSED_CASE_DATA,
+                cls.TRIBAL_STRATUM_DATA
+            ]
+
+        @classmethod
+        def is_tanf(cls, section: str) -> bool:
+            """Determine if the section is a TANF section."""
+            return section in [
+                cls.ACTIVE_CASE_DATA,
+                cls.CLOSED_CASE_DATA,
+                cls.AGGREGATE_DATA,
+                cls.STRATUM_DATA
+            ]
+
+    @staticmethod
+    def get_fra_section_list():
+        """Return FRA section list."""
+        return [
+                DataFile.Section.FRA_WORK_OUTCOME_TANF_EXITERS,
+                DataFile.Section.FRA_SECONDRY_SCHOOL_ATTAINMENT,
+                DataFile.Section.FRA_SUPPLEMENT_WORK_OUTCOMES
+            ]
+
     class Quarter(models.TextChoices):
         """Enum for data file Quarter."""
 
@@ -182,14 +234,13 @@ class Meta:
     def prog_type(self):
         """Return the program type for a given section."""
         # e.g., 'SSP Closed Case Data'
-        if self.section.startswith('SSP'):
+        if self.Section.is_ssp(self.section):
             return 'SSP'
+        elif self.Section.is_fra(self.section):
+            return 'FRA'
         else:
             return 'TAN'
 
-        # TODO: if given a datafile (section), we can reverse back to the program b/c the
-        # section string has "tribal/ssp" in it, then process of elimination we have tanf
-
     @property
     def filename(self):
         """Return the correct filename for this data file."""

tdrs-backend/tdpservice/data_files/serializers.py

@@ -114,3 +114,9 @@ def validate_file(self, file):
         validate_file_extension(file.name)
         validate_file_infection(file, file.name, user)
         return file
+
+    def validate_section(self, section):
+        """Validate the section field."""
+        if DataFile.Section.is_fra(section):
+            raise serializers.ValidationError("Section cannot be FRA")
+        return section

tdrs-backend/tdpservice/data_files/test/test_api.py

@@ -1,6 +1,7 @@
 """Tests for DataFiles Application."""
 import os
 from rest_framework import status
+from rest_framework.exceptions import ErrorDetail
 import pytest
 import base64
 import openpyxl
@@ -67,6 +68,11 @@ def assert_data_file_created(response):
         """Assert that the data file was created."""
         assert response.status_code == status.HTTP_201_CREATED
 
+    @staticmethod
+    def assert_data_file_error(response):
+        """Assert that the data file was created."""
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
     @staticmethod
     def assert_data_file_rejected(response):
         """Assert that a given data file submission was rejected."""
@@ -154,6 +160,15 @@ def post_data_file_file(self, api_client, data_file_data):
             format='multipart'
         )
 
+    def post_data_file_fra(self, api_client, data_file_data):
+        """Submit a data file with the given data."""
+        data_file_data['section'] = 'Secondary School Attainment'
+        return api_client.post(
+            self.root_url,
+            data_file_data,
+            format='multipart'
+        )
+
     def get_data_file_files(self, api_client):
         """Submit a data file with the given data."""
         return api_client.get(
@@ -213,6 +228,12 @@ def test_create_data_file_file_entry(self, api_client, data_file_data, user):
         self.assert_data_file_created(response)
         self.assert_data_file_exists(data_file_data, 1, user)
 
+    def test_create_data_file_fra(self, api_client, data_file_data, user):
+        """Test ability to create data file metadata registry."""
+        response = self.post_data_file_fra(api_client, data_file_data)
+        assert response.data == {'section': [ErrorDetail(string='Section cannot be FRA', code='invalid')]}
+        self.assert_data_file_error(response)
+
     def test_data_file_file_version_increment(
         self,
         api_client,

tdrs-backend/tdpservice/data_files/test/test_models.py

@@ -84,7 +84,18 @@ def test_data_files_filename_is_expected(user):
             assert new_data_file.filename == stt.filenames[section]
 
 @pytest.mark.django_db
-def test_prog_type(data_file_instance):
+@pytest.mark.parametrize('section, program_type',
+                         [('Tribal Closed Case Data', 'TAN'),
+                          ('Tribal Active Case Data', 'TAN'),
+                          ('SSP Aggregate Data', 'SSP'),
+                          ('SSP Closed Case Data', 'SSP'),
+                          ('Active Case Data', 'TAN'),
+                          ('Aggregate Data', 'TAN'),
+                          ('Work Outcomes for TANF Exiters', 'FRA'),
+                          ('Secondary School Attainment', 'FRA'),
+                          ('Supplemental Work Outcomes', 'FRA')
+                          ])
+def test_prog_type(data_file_instance, section, program_type):
     """Test propert prog_type."""
     df = DataFile.create_new_version({
         "year": data_file_instance.year,
@@ -97,11 +108,8 @@ def test_prog_type(data_file_instance):
         "user": data_file_instance.user,
     })
 
-    assert df.prog_type == "TAN"
-    df.section = "SSP Active Case Data"
-    assert df.prog_type == "SSP"
-    df.section = "Tribal Active Case Data"
-    assert df.prog_type == "TAN"
+    df.section = section
+    assert df.prog_type == program_type
 
 @pytest.mark.django_db
 def test_fiscal_year(data_file_instance):

tdrs-backend/tdpservice/data_files/views.py

@@ -54,6 +54,7 @@ class DataFileViewSet(ModelViewSet):
     def create(self, request, *args, **kwargs):
         """Override create to upload in case of successful scan."""
         logger.debug(f"{self.__class__.__name__}: {request}")
+
         response = super().create(request, *args, **kwargs)
 
         # only if file is passed the virus scan and created successfully will we perform side-effects:
@@ -69,6 +70,10 @@ def create(self, request, *args, **kwargs):
                         f"Datafile META -> datafile: {data_file_id}, section: {data_file.section}, " +
                         f"quarter {data_file.quarter}, year {data_file.year}.")
 
+            if data_file.prog_type == 'FRA':
+                logger.debug(f"{self.__class__.__name__}: return val: {response}")
+                return response
+
             parser_task.parse.delay(data_file_id)
             logger.info("Submitted parse task to queue for datafile %s.", data_file_id)
 
@@ -97,9 +102,16 @@ def get_s3_versioning_id(self, file_name, prefix):
     def get_queryset(self):
         """Apply custom queryset filters."""
         queryset = super().get_queryset().order_by('-created_at')
+        FRA_SECTION_LIST = [
+                DataFile.Section.FRA_WORK_OUTCOME_TANF_EXITERS,
+                DataFile.Section.FRA_SECONDRY_SCHOOL_ATTAINMENT,
+                DataFile.Section.FRA_SUPPLEMENT_WORK_OUTCOMES
+                ]
         if self.action == 'list':
             if self.request.query_params.get('file_type') == 'ssp-moe':
                 queryset = queryset.filter(section__contains='SSP')
+            elif self.request.query_params.get('file_type') == 'fra':
+                queryset = queryset.filter(section__in=FRA_SECTION_LIST)
             else:
                 queryset = queryset.exclude(section__contains='SSP')
 

tdrs-backend/tdpservice/users/admin.py

@@ -8,6 +8,8 @@
 
 from .models import User
 
+import logging
+logger = logging.getLogger()
 
 class UserForm(forms.ModelForm):
     """Customize the user admin form."""
@@ -26,8 +28,12 @@ def clean(self):
         if len(groups) > 1:
             raise ValidationError("User should not have multiple groups")
 
-        return cleaned_data
+        feature_flags = cleaned_data.get('feature_flags', {})
+        if not feature_flags:
+            feature_flags = {}
+        cleaned_data['feature_flags'] = feature_flags
 
+        return cleaned_data
 
 class UserAdmin(admin.ModelAdmin):
     """Customize the user admin functions."""

tdrs-backend/tdpservice/users/migrations/0042_user_feature_flags.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.15 on 2025-01-28 14:53
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('users', '0041_users_digit_group_add_datafile_permission'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='user',
+            name='feature_flags',
+            field=models.JSONField(blank=True, default=dict, help_text='Feature flags for this user. This is a JSON field that can be used to store key-value pairs. E.g: {"fra_access": true}'),
+        ),
+    ]

tdrs-backend/tdpservice/users/models.py

@@ -114,6 +114,19 @@ class Meta:
     _loaded_values = None
     _adding = True
 
+    # Feature flag for the user to enable or disable FRA access
+    feature_flags = models.JSONField(
+        default=dict,
+        help_text='Feature flags for this user. This is a JSON field that can be used to store key-value pairs. ' +
+        'E.g: {"fra_access": true}',
+        blank=True,
+    )
+
+    @property
+    def has_fra_access(self):
+        """Return whether or not the user has FRA access."""
+        return self.feature_flags.get('fra_access', False)
+
     def __str__(self):
         """Return the username as the string representation of the object."""
         return self.username

tdrs-backend/tdpservice/users/test/test_models.py

@@ -4,6 +4,9 @@
 import pytest
 
 from tdpservice.stts.models import STT, Region
+from tdpservice.data_files.models import DataFile
+from tdpservice.data_files.test.factories import DataFileFactory
+from django.test import Client
 
 
 @pytest.mark.django_db
@@ -70,3 +73,30 @@ def test_user_can_only_have_stt_or_region(user, stt, region):
 
         user.clean()
         user.save()
+
+@pytest.mark.django_db
+def test_user_with_fra_access(client, admin_user, stt):
+    """Test that a user with FRA access can only have an STT."""
+    admin_user.stt = stt
+    admin_user.is_superuser = True
+    admin_user.feature_flags = {"fra_access": False}
+
+    admin_user.clean()
+    admin_user.save()
+
+    client = Client()
+    client.login(username=admin_user.username, password="test_password")
+
+    datafile = DataFileFactory()
+    datafile.section = DataFile.Section.FRA_WORK_OUTCOME_TANF_EXITERS
+    datafile.save()
+
+    response = client.get(f"/admin/data_files/datafile/{datafile.id}/change/")
+    assert response.status_code == 302
+
+    admin_user.feature_flags = {"fra_access": True}
+    admin_user.save()
+
+    response = client.get(f"/admin/data_files/datafile/{datafile.id}/change/")
+    assert response.status_code == 200
+    assert '<div class="readonly">Fra Work Outcome Tanf Exiters</div>' in response.content.decode('utf-8')