add django app for GCE screener responses (#2465)

Adds a new app to save user responses on the standalone Good Cause Eviction screener tool. We set up the model for responses, and a single API endpoint for POST requests from the screener. On first the first request with address it creates a new record and returns the ID that's autogenerated in the DB. On subsequent requests with that ID included it will update the fields with new data.

Since all data being sent it directly controlled by us on the GCE screener side (no direct user input strings), I don't think we need extensive validation with forms.py, and it wasn't working as it does in wow), so we use a simple pydantic model with basic tyes. But in future if we wanted we can use pydantics validation options to directly match the django model (eg. options for yes/no/unsure, etc.)

Since we are not using the frontend of tenants2, we need to log errors for 500 responses so they get tracked by rollbar.

I have the GCE_ORIGN set in justfix_environment.py with the temporary one we are using, but we'll need to update this once we settle on a final domain for the screener.

Also, unrelated to this project, I added port forwarding for the postgres DB in docker-compose so we can connect locally (eg. via postico) for easier debugging.

Companion PR for gce-screener: JustFixNYC/gce-screener#39

docker-compose.yml

@@ -30,6 +30,8 @@ services:
     environment:
       - POSTGRES_DB=justfix
       - POSTGRES_USER=justfix
+    ports:
+      - "127.0.0.1:5432:5432"
 volumes:
   node-modules:
   unused-node-modules:

gce/apps.py

@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+
+class GoodCauseEvictionScreenerConfig(AppConfig):
+    name = "gce"

gce/migrations/0001_initial.py

@@ -0,0 +1,35 @@
+# Generated by Django 3.2.13 on 2024-12-04 21:39
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='GoodCauseEvictionScreenerResponse',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+                ('bbl', models.CharField(help_text='The zero-padded borough, block and lot (BBL) number for the search address property.', max_length=10)),
+                ('house_number', models.TextField()),
+                ('street_name', models.TextField()),
+                ('borough', models.CharField(choices=[('BROOKLYN', 'Brooklyn'), ('QUEENS', 'Queens'), ('BRONX', 'Bronx'), ('MANHATTAN', 'Manhattan'), ('STATEN_ISLAND', 'Staten Island')], max_length=20)),
+                ('zipcode', models.CharField(max_length=5)),
+                ('address_confirmed', models.BooleanField(default=False, help_text='Whether the user has clicked to confirm the search address is correct.')),
+                ('nycdb_results', models.JSONField(blank=True, help_text='Response from the WOW gce/screener API for the search address. Schema may change over time.', null=True)),
+                ('form_answers_initial', models.JSONField(blank=True, help_text='The initial set of user responses to the survey form saved on submission, before taking and necessary next steps to confirm criteria.', null=True)),
+                ('result_coverage_initial', models.TextField(blank=True, choices=[('COVERED', 'Covered by GCE'), ('NOT_COVERED', 'Not covered by GCE'), ('UNKNOWN', 'Unknown if covered by GCE')], help_text='The initial GCE coverage result based on building data and user form responses, before taking any necessary next steps to confirm criteria.', null=True)),
+                ('result_criteria_initial', models.JSONField(blank=True, help_text='An object with each GCE criteria and the initial eligibility determination (eligible, ineligible, unknown), before taking any next steps to confirm criteria.', null=True)),
+                ('form_answers_final', models.JSONField(blank=True, help_text='The final set of user responses to the survey form saved on submission, after taking any necessary next steps to confirm criteria.', null=True)),
+                ('result_coverage_final', models.TextField(blank=True, choices=[('COVERED', 'Covered by GCE'), ('NOT_COVERED', 'Not covered by GCE'), ('UNKNOWN', 'Unknown if covered by GCE')], help_text='The final GCE coverage result taking into account any confirmed criteria from next steps.', null=True)),
+                ('result_criteria_final', models.JSONField(blank=True, help_text='An object with each GCE criteria and the final eligibility determination (eligible, ineligible, unknown), after taking any next steps to confirm criteria.', null=True)),
+            ],
+        ),
+    ]

gce/models.py

@@ -0,0 +1,105 @@
+from django.db import models
+
+from project.common_data import Choices
+from project.util.address_form_fields import BOROUGH_FIELD_KWARGS
+
+COVERAGE = Choices(
+    [
+        ("COVERED", "Covered by GCE"),
+        ("NOT_COVERED", "Not covered by GCE"),
+        ("UNKNOWN", "Unknown if covered by GCE"),
+    ]
+)
+
+
+class GoodCauseEvictionScreenerResponse(models.Model):
+
+    created_at = models.DateTimeField(auto_now_add=True)
+
+    updated_at = models.DateTimeField(auto_now=True)
+
+    bbl: str = models.CharField(
+        max_length=10,  # One for the borough, 5 for the block, 4 for the lot.
+        help_text=(
+            "The zero-padded borough, block and lot (BBL) number for the "
+            "search address property."
+        ),
+    )
+
+    house_number: str = models.TextField()
+
+    street_name: str = models.TextField()
+
+    borough: str = models.CharField(**BOROUGH_FIELD_KWARGS)
+
+    zipcode: str = models.CharField(max_length=5)
+
+    address_confirmed: bool = models.BooleanField(
+        help_text="Whether the user has clicked to confirm the search address is correct.",
+        default=False,
+    )
+
+    nycdb_results = models.JSONField(
+        help_text=(
+            "Response from the WOW gce/screener API for the search address. "
+            "Schema may change over time."
+        ),
+        blank=True,
+        null=True,
+    )
+
+    form_answers_initial = models.JSONField(
+        help_text=(
+            "The initial set of user responses to the survey form saved on submission, before "
+            "taking and necessary next steps to confirm criteria."
+        ),
+        blank=True,
+        null=True,
+    )
+
+    result_coverage_initial: str = models.TextField(
+        help_text=(
+            "The initial GCE coverage result based on building data and user form responses, "
+            "before taking any necessary next steps to confirm criteria."
+        ),
+        choices=COVERAGE.choices,
+        null=True,
+        blank=True,
+    )
+
+    result_criteria_initial = models.JSONField(
+        help_text=(
+            "An object with each GCE criteria and the initial eligibility determination "
+            "(eligible, ineligible, unknown), before taking any next steps to confirm criteria."
+        ),
+        blank=True,
+        null=True,
+    )
+
+    form_answers_final = models.JSONField(
+        help_text=(
+            "The final set of user responses to the survey form saved on submission, after "
+            "taking any necessary next steps to confirm criteria."
+        ),
+        blank=True,
+        null=True,
+    )
+
+    result_coverage_final: str = models.TextField(
+        help_text=(
+            "The final GCE coverage result taking into account any confirmed "
+            "criteria from next steps."
+        ),
+        choices=COVERAGE.choices,
+        null=True,
+        blank=True,
+    )
+
+    result_criteria_final = models.JSONField(
+        help_text=(
+            "An object with each GCE criteria and the final eligibility determination "
+            "(eligible, ineligible, unknown), after taking any next steps to confirm criteria."
+        ),
+        blank=True,
+        null=True,
+    )

gce/tests/test_views.py

@@ -0,0 +1,212 @@
+import pytest
+import json
+
+from gce.models import GoodCauseEvictionScreenerResponse
+
+
+def base_headers(settings):
+    return {
+        "HTTP_AUTHORIZATION": f"Bearer {settings.GCE_API_TOKEN}",
+        "content_type": "application/json",
+    }
+
+
+DATA_STEP_1 = {
+    "bbl": "1234567890",
+    "house_number": "123",
+    "street_name": "MAIN ST",
+    "borough": "BROOKLYN",
+    "zipcode": "12345",
+}
+
+DATA_STEP_2 = {
+    "address_confirmed": "true",
+    "nycdb_results": {"test": "structure of json not enforced, so not included"},
+}
+
+DATA_STEP_3 = {
+    "form_answers": {
+        "bedrooms": "STUDIO",
+        "rent": 1000.50,
+        "owner_occupied": "YES",
+        "rent_stab": "NO",
+        "subsidy": "UNSURE",
+    }
+}
+
+DATA_STEP_4 = {
+    "result_coverage": "UNKNOWN",
+    "result_criteria": {
+        "rent": "ELIGIBLE",
+        "rent_stab": "INELIGIBLE",
+        "building_class": "ELIGIBLE",
+        "c_of_o": "ELIGIBLE",
+        "subsidy": "ELIGIBLE",
+        "portfolio_size": "UNKNOWN",
+    },
+}
+
+INVALID_DATA = {
+    "bbl": "X",
+    "address_confirmed": "X",
+    "form_answers": {
+        "X": "STUDIO",
+        "rent": 1000.50,
+        "owner_occupied": "NO",
+        "rent_stab": "NO",
+        "subsidy": "X",
+    },
+    "result_coverage": "X",
+}
+
+INVALID_ERRORS = [
+    {
+        "loc": ["bbl"],
+        "msg": "BBL must be 10-digit zero padded string",
+    },
+    {
+        "loc": ["address_confirmed"],
+        "msg": "value could not be parsed to a boolean",
+    },
+    {
+        "loc": ["form_answers", "bedrooms"],
+        "msg": "field required",
+    },
+    {
+        "loc": ["form_answers", "subsidy"],
+        "msg": "unexpected value; permitted: 'NYCHA', 'SUBSIDIZED', 'NONE', 'UNSURE'",
+    },
+    {
+        "loc": ["result_coverage"],
+        "msg": "unexpected value; permitted: 'COVERED', 'NOT_COVERED', 'UNKNOWN'",
+    },
+]
+
+
+def create_new_gce_record():
+    gcer = GoodCauseEvictionScreenerResponse(**DATA_STEP_1)
+    gcer.full_clean()
+    gcer.save()
+    user_id = {"id": gcer.pk}
+    return gcer, user_id
+
+
+def authorized_request(client, settings, post_data, **kawrgs):
+    return client.post(
+        "/gce/upload",
+        json.dumps(post_data),
+        HTTP_ORIGIN=settings.GCE_ORIGIN,
+        **base_headers(settings),
+        **kawrgs,
+    )
+
+
+def get_gcer_by_id(id):
+    return GoodCauseEvictionScreenerResponse.objects.get(id=id)
+
+
+@pytest.mark.django_db
+def test_unauthorized_request_fails(client, settings):
+    res = client.post("/gce/upload", HTTP_ORIGIN=settings.GCE_ORIGIN)
+    assert res.status_code == 401
+    assert res.json()["error"] == "Unauthorized request"
+
+
+@pytest.mark.django_db
+def test_initial_post_creates_record(client, settings):
+    res = authorized_request(client, settings, DATA_STEP_1)
+    assert res.status_code == 200
+    assert res["Content-Type"] == "application/json"
+    data = res.json()
+    gcer = get_gcer_by_id(data["id"])
+    assert gcer.bbl == DATA_STEP_1["bbl"]
+
+
+@pytest.mark.django_db
+def test_subsequent_post_updates_record(client, settings):
+    gcer, user_id = create_new_gce_record()
+
+    res = authorized_request(client, settings, {**DATA_STEP_2, **user_id})
+    assert res.status_code == 200
+    assert res.json() == user_id
+
+    gcer.refresh_from_db()
+    assert gcer.address_confirmed is True
+    assert gcer.updated_at > gcer.created_at
+
+
+@pytest.mark.django_db
+def test_valid_data_works(client, settings):
+    res = authorized_request(client, settings, {**DATA_STEP_1})
+    assert res.status_code == 200
+    user = res.json()
+
+    res = authorized_request(client, settings, {**DATA_STEP_2, **user})
+    assert res.status_code == 200
+
+    res = authorized_request(client, settings, {**DATA_STEP_3, **user})
+    assert res.status_code == 200
+
+    res = authorized_request(client, settings, {**DATA_STEP_4, **user})
+    assert res.status_code == 200
+
+    gcer = get_gcer_by_id(user["id"])
+
+    for field in GoodCauseEvictionScreenerResponse._meta.get_fields():
+        if not field.null and field.name != "id":
+            value = getattr(gcer, field.name)
+            assert value is not None
+
+
+@pytest.mark.django_db
+def test_invalid_data_fails(client, settings):
+    res = authorized_request(client, settings, INVALID_DATA)
+    assert res.status_code == 400
+    data = res.json()
+    assert data["error"] == "Invalid POST data"
+    details = json.loads(data["details"])
+    assert len(details) == len(INVALID_ERRORS)
+    for error, expected in zip(details, INVALID_ERRORS):
+        assert error["loc"] == expected["loc"]
+        assert error["msg"] == expected["msg"]
+
+
+@pytest.mark.django_db
+def test_nonexistant_user_fails(client, settings):
+    post_data = {"id": 100, **DATA_STEP_2}
+    res = authorized_request(client, settings, post_data)
+    assert res.status_code == 500
+    assert res.json()["error"] == "User does not exist"
+
+
+@pytest.mark.django_db
+def test_initial_preserved_final_updates(client, settings):
+    gcer, user_id = create_new_gce_record()
+
+    initial_data = {**user_id, "result_coverage": "UNKNOWN"}
+    final_data = {**user_id, "result_coverage": "COVERED"}
+
+    res_initial = authorized_request(client, settings, initial_data)
+    assert res_initial.status_code == 200
+
+    gcer.refresh_from_db()
+    assert gcer.result_coverage_initial == initial_data["result_coverage"]
+
+    res_final = authorized_request(client, settings, final_data)
+    assert res_final.status_code == 200
+
+    gcer.refresh_from_db()
+    assert gcer.result_coverage_initial == initial_data["result_coverage"]
+    assert gcer.result_coverage_final == final_data["result_coverage"]
+
+
+@pytest.mark.django_db
+def test_invalid_origin_fails(client, settings):
+    res = client.post(
+        "/gce/upload",
+        json.dumps(DATA_STEP_1),
+        **base_headers(settings),
+        HTTP_ORIGIN="https://example.com",
+    )
+    assert res.status_code == 403
+    assert res.json()["error"] == "Invalid origin"

gce/urls.py

@@ -0,0 +1,10 @@
+from django.urls import path
+
+from . import views
+
+
+app_name = "gce"
+
+urlpatterns = [
+    path("upload", views.upload, name="upload"),
+]