Merge pull request #1478 from KentShikama/statewide-mj-form

Use state-wide mj expungement form where applicable

src/backend/expungeservice/form_filling.py

@@ -1,3 +1,4 @@
+import os
 from dataclasses import dataclass, replace
 from os import path
 from pathlib import Path
@@ -18,7 +19,7 @@
 class FormData:
     case_name: str
     case_number: str
-    case_number_with_comments: str # For legacy reasons; same as case_number
+    case_number_with_comments: str  # For legacy reasons; same as case_number
     da_number: str
     full_name: str
     date_of_birth: str
@@ -99,8 +100,8 @@ def build_zip(record_summary: RecordSummary, user_information: Dict[str, str]) -
             )
             pdf_with_warnings = FormFilling._build_pdf_for_case(case_without_deleted_charges, user_information)
             if pdf_with_warnings:
-                pdf, warnings = pdf_with_warnings
-                file_name = f"{case_without_deleted_charges.summary.name}_{case_without_deleted_charges.summary.case_number}.pdf"
+                pdf, internal_file_name, warnings = pdf_with_warnings
+                file_name = f"{case_without_deleted_charges.summary.name}_{case_without_deleted_charges.summary.case_number}_{internal_file_name}"
                 file_path = path.join(temp_dir, file_name)
                 writer = PdfWriter()
                 writer.addpages(pdf.pages)
@@ -124,7 +125,7 @@ def _add_warnings(writer: PdfWriter, warnings: List[str]):
             writer.addpages(blank_pdf.pages)
 
     @staticmethod
-    def _build_pdf_for_case(case: Case, user_information: Dict[str, str]) -> Optional[Tuple[PdfReader, List[str]]]:
+    def _build_pdf_for_case(case: Case, user_information: Dict[str, str]) -> Optional[Tuple[PdfReader, str, List[str]]]:
         eligible_charges, ineligible_charges = Case.partition_by_eligibility(case.charges)
         in_part = ", ".join([charge.ambiguous_charge_id.split("-")[-1] for charge in eligible_charges])
         case_number_with_comments = (
@@ -133,22 +134,22 @@ def _build_pdf_for_case(case: Case, user_information: Dict[str, str]) -> Optiona
             else case.summary.case_number
         )
         if eligible_charges:
-            pdf, warnings = FormFilling._build_pdf_for_eligible_case(
+            pdf, file_name, warnings = FormFilling._build_pdf_for_eligible_case(
                 case, eligible_charges, user_information, case_number_with_comments
             )
             if ineligible_charges:
                 warnings.insert(
                     0,
                     "This form will attempt to expunge a case in part. This is relatively rare, and thus these forms should be reviewed particularly carefully.",
                 )
-            return pdf, warnings
+            return pdf, file_name, warnings
         else:
             return None
 
     @staticmethod
     def _build_pdf_for_eligible_case(
         case: Case, eligible_charges: List[Charge], user_information: Dict[str, str], case_number_with_comments: str
-    ) -> Tuple[PdfReader, List[str]]:
+    ) -> Tuple[PdfReader, str, List[str]]:
         warnings: List[str] = []
         charges = case.charges
         charge_names = [charge.name.title() for charge in charges]
@@ -192,8 +193,14 @@ def _build_pdf_for_eligible_case(
         warning = FormFilling._warn_charge_count_overflow(location, convictions, dismissals)
         if warning:
             warnings.append(warning)
-        pdf_path = FormFilling._build_pdf_path(location, convictions)
-        pdf = PdfReader(pdf_path)
+        if case.qualifying_marijuana_conviction_form_applicable():
+            file_name = "statewide_marijuana_conviction.pdf"
+            pdf_path = path.join(Path(__file__).parent, "files", file_name)
+            pdf = PdfReader(pdf_path)
+        else:
+            pdf_path = FormFilling._build_pdf_path(location, convictions)
+            file_name = os.path.basename(pdf_path)
+            pdf = PdfReader(pdf_path)
         for field in pdf.Root.AcroForm.Fields:
             field_name = field.T.lower().replace(" ", "_").replace("(", "").replace(")", "")
             field_value = getattr(form, field_name)
@@ -205,7 +212,7 @@ def _build_pdf_for_eligible_case(
                 for annotation in annotations:
                     annotation.update(PdfDict(AP=""))
         pdf.Root.AcroForm.update(PdfDict(NeedAppearances=PdfObject("true")))
-        return pdf, warnings
+        return pdf, file_name, warnings
 
     @staticmethod
     def _set_font(field: PdfDict, field_value: str) -> List[str]:

src/backend/expungeservice/models/case.py

@@ -75,6 +75,13 @@ def has_eligible_conviction(self):
         dismissals, convictions = Case.categorize_charges(eligible_charges)
         return len(convictions) > 0
 
+    def qualifying_marijuana_conviction_form_applicable(self):
+        eligible_charges, ineligible_charges = Case.partition_by_eligibility(self.charges)
+        for charge in eligible_charges:
+            if not charge.is_qualifying_mj_conviction():
+                return False
+        return True
+
     @staticmethod
     def partition_by_eligibility(charges: Tuple[Charge, ...]):
         ineligible_charges_generator, eligible_charges_generator = partition(
@@ -138,6 +145,6 @@ def compute_balance_due_in_cents(balance_due_dollar_amount: str):
 
     @staticmethod
     def _balance_to_float(balance: str) -> float:
-        commas_removed = balance.replace(",","")
+        commas_removed = balance.replace(",", "")
         normalized_negative = re.sub("\((?P<balance>.*)\)", "-\g<balance>", commas_removed)
         return float(normalized_negative)

src/backend/expungeservice/models/charge.py

@@ -53,8 +53,8 @@ class ChargeType:
 
     def type_eligibility(self, disposition):
         """If the disposition is present and recognized, this should always return a TypeEligibility.
-It may also return the eligibility without a known disposition (this works for some types).
-If the type eligibility is unknown, the method can return None. """
+        It may also return the eligibility without a known disposition (this works for some types).
+        If the type eligibility is unknown, the method can return None."""
         raise NotImplementedError
 
     def hidden_in_record_summary(self):
@@ -116,3 +116,11 @@ def to_one_line(self) -> str:
         disposition = str(self.disposition.status.name)
         owed = f" - $ owed" if self.balance_due_in_cents > 0 else ""
         return f"{short_name} ({disposition}) Charged {charged_date}{owed}"
+
+    def is_qualifying_mj_conviction(self):
+        # See https://www.oregonlaws.org/ors/475B.401
+        from expungeservice.models.charge_types.marijuana_eligible import MarijuanaViolation
+
+        is_qualifying_type = "Poss LT 1 Oz Marijuana" in self.name or isinstance(self.charge_type, MarijuanaViolation)
+        is_qualifying_date = self.date < date_class(2015, 7, 1)
+        return is_qualifying_type and is_qualifying_date and self.convicted()

src/backend/expungeservice/record_summarizer.py

@@ -27,7 +27,7 @@ def summarize(record, questions: Dict[str, QuestionSummary]) -> RecordSummary:
         )
 
     @staticmethod
-    def _build_county_balances(record: Record):
+    def _build_county_balances(record: Record) -> Tuple[List[CountyFines], List[CountyFilingFee]]:
         def get_location(case: Case):
             return case.summary.location
 
@@ -37,10 +37,14 @@ def get_location(case: Case):
             cases = list(cases_by_county)
             cases_with_fines = filter(lambda case: case.summary.get_balance_due(), cases)
             fines = [CaseFine(case.summary.case_number, case.summary.get_balance_due()) for case in cases_with_fines]
-            cases_with_eligible_convictions = [case for case in cases if case.has_eligible_conviction()]
+            cases_with_conviction_fees = [
+                case
+                for case in cases
+                if case.has_eligible_conviction() and not case.qualifying_marijuana_conviction_form_applicable()
+            ]
             county_fines_list.append(CountyFines(location, fines))
-            if len(cases_with_eligible_convictions) > 0:
-                county_filing_fees.append(CountyFilingFee(location, len(cases_with_eligible_convictions)))
+            if len(cases_with_conviction_fees) > 0:
+                county_filing_fees.append(CountyFilingFee(location, len(cases_with_conviction_fees)))
         return county_fines_list, county_filing_fees
 
     @staticmethod
@@ -94,7 +98,7 @@ def get_label(charge: Charge):
 
     @staticmethod
     def _build_no_fees_reason(charges):
-        reason = "None (no eligible cases)"
+        reason = "None"
         if charges:
             nonconvictions_eligible_now = [
                 c

src/backend/tests/models/test_record_summarizer.py

@@ -31,7 +31,8 @@ def test_record_summarizer_multiple_cases():
         charges=tuple(
             [
                 ChargeFactory.create(
-                    case_number="2", disposition=DispositionCreator.create(ruling="Convicted", date=date(2010, 1, 1)),
+                    case_number="2",
+                    disposition=DispositionCreator.create(ruling="Convicted", date=date(2010, 1, 1)),
                 ),
                 ChargeFactory.create(
                     case_number="2",
@@ -123,7 +124,10 @@ def test_record_summarizer_multiple_cases():
                 case_partially_eligible.charges[1].ambiguous_charge_id,
                 "Theft of services (CONVICTED) Charged Jan 1, 2010",
             ),
-            (case_all_ineligible.charges[0].ambiguous_charge_id, "Theft of services (CONVICTED) Charged Jan 1, 2010",),
+            (
+                case_all_ineligible.charges[0].ambiguous_charge_id,
+                "Theft of services (CONVICTED) Charged Jan 1, 2010",
+            ),
             (
                 case_all_ineligible_2.charges[0].ambiguous_charge_id,
                 "Theft of services (CONVICTED) Charged Jan 1, 2010",
@@ -155,4 +159,4 @@ def test_record_summarizer_no_cases():
     assert record_summary.county_fines == []
     assert record_summary.eligible_charges_by_date == {}
     assert record_summary.county_filing_fees == []
-    assert record_summary.no_fees_reason == "None (no eligible cases)"
+    assert record_summary.no_fees_reason == "None"

src/backend/tests/test_crawler_expunger.py

@@ -5,13 +5,11 @@
 from dateutil.relativedelta import relativedelta
 from expungeservice.expunger import Expunger
 from expungeservice.models.charge_types.marijuana_eligible import MarijuanaUnder21
-from expungeservice.models.charge_types.juvenile_charge import JuvenileCharge
 from expungeservice.models.charge_types.civil_offense import CivilOffense
 from expungeservice.models.charge_types.traffic_violation import TrafficViolation
 from expungeservice.models.charge_types.felony_class_a import FelonyClassA
 
 from expungeservice.models.expungement_result import EligibilityStatus, TimeEligibility
-from tests.models.test_charge import Dispositions
 from tests.factories.crawler_factory import CrawlerFactory
 from tests.fixtures.case_details import CaseDetails
 from tests.fixtures.john_doe import JohnDoe
@@ -212,7 +210,10 @@ def test_probation_revoked_affects_time_eligibility(record_with_revoked_probatio
 @pytest.fixture
 def record_with_odd_event_table_contents():
     return CrawlerFactory.create(
-        record=JohnDoe.SINGLE_CASE_RECORD, cases={"CASEJD1": CaseDetails.CASE_WITH_ODD_EVENT_TABLE_CONTENTS,},
+        record=JohnDoe.SINGLE_CASE_RECORD,
+        cases={
+            "CASEJD1": CaseDetails.CASE_WITH_ODD_EVENT_TABLE_CONTENTS,
+        },
     )
 
 
@@ -235,7 +236,10 @@ def test_expunger_for_record_with_odd_event_table_contents(record_with_odd_event
 @pytest.fixture
 def record_with_mj_under_21():
     return CrawlerFactory.create(
-        record=YoungDoe.SINGLE_CASE_RECORD, cases={"CASEJD1": CaseDetails.CASE_MJ_CONVICTION,},
+        record=YoungDoe.SINGLE_CASE_RECORD,
+        cases={
+            "CASEJD1": CaseDetails.CASE_MJ_CONVICTION,
+        },
     )
 
 
@@ -251,7 +255,12 @@ def test_expunger_for_record_with_mj_under_21(record_with_mj_under_21):
 
 @pytest.fixture
 def record_with_mj_over_21():
-    return CrawlerFactory.create(record=JohnDoe.SINGLE_CASE_RECORD, cases={"CASEJD1": CaseDetails.CASE_MJ_CONVICTION,},)
+    return CrawlerFactory.create(
+        record=JohnDoe.SINGLE_CASE_RECORD,
+        cases={
+            "CASEJD1": CaseDetails.CASE_MJ_CONVICTION,
+        },
+    )
 
 
 def test_expunger_for_record_with_mj_over_21(record_with_mj_over_21):
@@ -266,7 +275,10 @@ def test_expunger_for_record_with_mj_over_21(record_with_mj_over_21):
 @pytest.fixture
 def record_with_mj_under_21_and_traffic_violation():
     return CrawlerFactory.create(
-        record=YoungDoe.SINGLE_CASE_RECORD, cases={"CASEJD1": CaseDetails.CASE_MJ_AND_TRAFFIC_CONVICTION,}
+        record=YoungDoe.SINGLE_CASE_RECORD,
+        cases={
+            "CASEJD1": CaseDetails.CASE_MJ_AND_TRAFFIC_CONVICTION,
+        },
     )
 
 

src/backend/tests/test_form_filling.py

@@ -0,0 +1,57 @@
+import os
+from tempfile import mkdtemp
+from zipfile import ZipFile
+
+from expungeservice.expunger import Expunger
+from expungeservice.form_filling import FormFilling
+from expungeservice.record_merger import RecordMerger
+from expungeservice.record_summarizer import RecordSummarizer
+from tests.factories.crawler_factory import CrawlerFactory
+from tests.fixtures.case_details import CaseDetails
+from tests.fixtures.john_doe import JohnDoe
+from tests.test_crawler_expunger import record_with_mj_over_21
+
+
+def test_normal_conviction_uses_multnomah_conviction_form():
+    record = CrawlerFactory.create(JohnDoe.SINGLE_CASE_RECORD, {"CASEJD1": CaseDetails.CASEJD74})
+    expunger_result = Expunger.run(record)
+    merged_record = RecordMerger.merge([record], [expunger_result], [])
+    record_summary = RecordSummarizer.summarize(merged_record, {})
+    user_information = {
+        "full_name": "",
+        "date_of_birth": "",
+        "mailing_address": "",
+        "phone_number": "",
+        "city": "",
+        "state": "",
+        "zip_code": "",
+    }
+    zip_path, zip_name = FormFilling.build_zip(record_summary, user_information)
+    temp_dir = mkdtemp()
+    with ZipFile(zip_path, "r") as zip_ref:
+        zip_ref.extractall(temp_dir)
+        for _root, _dir, files in os.walk(temp_dir):
+            assert len(files) == 1
+            assert "multnomah_conviction" in files[0]
+
+
+def test_marijuana_violation_uses_statewide_form(record_with_mj_over_21):
+    expunger_result = Expunger.run(record_with_mj_over_21)
+    merged_record = RecordMerger.merge([record_with_mj_over_21], [expunger_result], [])
+    record_summary = RecordSummarizer.summarize(merged_record, {})
+    user_information = {
+        "full_name": "",
+        "date_of_birth": "",
+        "mailing_address": "",
+        "phone_number": "",
+        "city": "",
+        "state": "",
+        "zip_code": "",
+    }
+    zip_path, zip_name = FormFilling.build_zip(record_summary, user_information)
+    temp_dir = mkdtemp()
+    with ZipFile(zip_path, "r") as zip_ref:
+        zip_ref.extractall(temp_dir)
+        for _root, _dir, files in os.walk(temp_dir):
+            assert len(files) == 1
+            assert "statewide_marijuana_conviction" in files[0]