Divided unit test file into smaller test files

.github/workflows/unit-tests.yaml

@@ -22,4 +22,4 @@ jobs:
 
       - name: Run unit tests
         run: |
-          python -m unittest process_report/tests/unit_tests.py
+          python -m unittest

process_report/tests/unit/invoices/test_base_invoice.py

@@ -0,0 +1,63 @@
+from unittest import TestCase, mock
+import pandas
+
+from process_report.tests import util as test_utils
+
+
+class TestBaseInvoice(TestCase):
+    def test_filter_exported_columns(self):
+        test_invoice = pandas.DataFrame(columns=["C1", "C2", "C3", "C4", "C5"])
+        answer_invoice = pandas.DataFrame(columns=["C1", "C3R", "C5R"])
+        inv = test_utils.new_base_invoice(data=test_invoice)
+        inv.export_columns_list = ["C1", "C3", "C5"]
+        inv.exported_columns_map = {"C3": "C3R", "C5": "C5R"}
+        result_invoice = inv._filter_columns()
+
+        self.assertTrue(result_invoice.equals(answer_invoice))
+
+
+class TestUploadToS3(TestCase):
+    @mock.patch("process_report.util.get_invoice_bucket")
+    @mock.patch("process_report.util.get_iso8601_time")
+    def test_upload_to_s3(self, mock_get_time, mock_get_bucket):
+        mock_bucket = mock.MagicMock()
+        mock_get_bucket.return_value = mock_bucket
+        mock_get_time.return_value = "0"
+
+        invoice_month = "2024-03"
+        filenames = ["test-test", "test2.test", "test3"]
+        sample_base_invoice = test_utils.new_base_invoice(invoice_month=invoice_month)
+
+        answers = [
+            (
+                f"test-test {invoice_month}.csv",
+                f"Invoices/{invoice_month}/test-test {invoice_month}.csv",
+            ),
+            (
+                f"test-test {invoice_month}.csv",
+                f"Invoices/{invoice_month}/Archive/test-test {invoice_month} 0.csv",
+            ),
+            (
+                f"test2.test {invoice_month}.csv",
+                f"Invoices/{invoice_month}/test2.test {invoice_month}.csv",
+            ),
+            (
+                f"test2.test {invoice_month}.csv",
+                f"Invoices/{invoice_month}/Archive/test2.test {invoice_month} 0.csv",
+            ),
+            (
+                f"test3 {invoice_month}.csv",
+                f"Invoices/{invoice_month}/test3 {invoice_month}.csv",
+            ),
+            (
+                f"test3 {invoice_month}.csv",
+                f"Invoices/{invoice_month}/Archive/test3 {invoice_month} 0.csv",
+            ),
+        ]
+
+        for filename in filenames:
+            sample_base_invoice.name = filename
+            sample_base_invoice.export_s3(mock_bucket)
+
+        for i, call_args in enumerate(mock_bucket.upload_file.call_args_list):
+            self.assertTrue(answers[i] in call_args)

process_report/tests/unit/invoices/test_pi_specific_invoice.py

@@ -0,0 +1,68 @@
+from unittest import TestCase, mock
+import tempfile
+import pandas
+import os
+
+from process_report.tests import util as test_utils
+
+
+class TestExportPICSV(TestCase):
+    def setUp(self):
+        data = {
+            "Invoice Month": ["2023-01", "2023-01", "2023-01", "2023-01", "2023-01"],
+            "Manager (PI)": ["PI1", "PI1", "PI1", "PI2", "PI2"],
+            "Institution": ["BU", "BU", "BU", "HU", "HU"],
+            "Project - Allocation": [
+                "ProjectA",
+                "ProjectB",
+                "ProjectC",
+                "ProjectD",
+                "ProjectE",
+            ],
+            "Untouch Data Column": ["DataA", "DataB", "DataC", "DataD", "DataE"],
+            "Is Billable": [True, True, True, True, True],
+            "Missing PI": [False, False, False, False, False],
+        }
+        self.dataframe = pandas.DataFrame(data)
+        self.invoice_month = data["Invoice Month"][0]
+
+    @mock.patch("process_report.invoices.invoice.Invoice._filter_columns")
+    def test_export_pi(self, mock_filter_cols):
+        mock_filter_cols.return_value = self.dataframe
+
+        output_dir = tempfile.TemporaryDirectory()
+        pi_inv = test_utils.new_pi_specific_invoice(
+            output_dir.name, invoice_month=self.invoice_month, data=self.dataframe
+        )
+        pi_inv.process()
+        pi_inv.export()
+        pi_csv_1 = f'{self.dataframe["Institution"][0]}_{self.dataframe["Manager (PI)"][0]} {self.dataframe["Invoice Month"][0]}.csv'
+        pi_csv_2 = f'{self.dataframe["Institution"][3]}_{self.dataframe["Manager (PI)"][3]} {self.dataframe["Invoice Month"][3]}.csv'
+        self.assertIn(pi_csv_1, os.listdir(output_dir.name))
+        self.assertIn(pi_csv_2, os.listdir(output_dir.name))
+        self.assertEqual(
+            len(os.listdir(output_dir.name)),
+            len(self.dataframe["Manager (PI)"].unique()),
+        )
+
+        pi_df = pandas.read_csv(output_dir.name + "/" + pi_csv_1)
+        self.assertEqual(len(pi_df["Manager (PI)"].unique()), 1)
+        self.assertEqual(
+            pi_df["Manager (PI)"].unique()[0], self.dataframe["Manager (PI)"][0]
+        )
+
+        self.assertIn("ProjectA", pi_df["Project - Allocation"].tolist())
+        self.assertIn("ProjectB", pi_df["Project - Allocation"].tolist())
+        self.assertIn("ProjectC", pi_df["Project - Allocation"].tolist())
+
+        pi_df = pandas.read_csv(output_dir.name + "/" + pi_csv_2)
+        self.assertEqual(len(pi_df["Manager (PI)"].unique()), 1)
+        self.assertEqual(
+            pi_df["Manager (PI)"].unique()[0], self.dataframe["Manager (PI)"][3]
+        )
+
+        self.assertIn("ProjectD", pi_df["Project - Allocation"].tolist())
+        self.assertIn("ProjectE", pi_df["Project - Allocation"].tolist())
+        self.assertNotIn("ProjectA", pi_df["Project - Allocation"].tolist())
+        self.assertNotIn("ProjectB", pi_df["Project - Allocation"].tolist())
+        self.assertNotIn("ProjectC", pi_df["Project - Allocation"].tolist())

process_report/tests/unit/processors/test_add_institution_processor.py

@@ -0,0 +1,41 @@
+from unittest import TestCase
+
+from process_report.tests import util as test_utils
+
+
+class TestAddInstituteProcessor(TestCase):
+    def test_get_pi_institution(self):
+        institute_map = {
+            "harvard.edu": "Harvard University",
+            "bu.edu": "Boston University",
+            "bentley.edu": "Bentley",
+            "mclean.harvard.edu": "McLean Hospital",
+            "northeastern.edu": "Northeastern University",
+            "childrens.harvard.edu": "Boston Children's Hospital",
+            "meei.harvard.edu": "Massachusetts Eye & Ear",
+            "dfci.harvard.edu": "Dana-Farber Cancer Institute",
+            "bwh.harvard.edu": "Brigham and Women's Hospital",
+            "bidmc.harvard.edu": "Beth Israel Deaconess Medical Center",
+        }
+
+        answers = {
+            "[email protected]": "Boston University",
+            "[email protected]": "McLean Hospital",
+            "[email protected]": "Harvard University",
+            "e@edu": "",
+            "[email protected]": "Northeastern University",
+            "[email protected]": "Harvard University",
+            "[email protected]": "Boston Children's Hospital",
+            "[email protected]": "Massachusetts Eye & Ear",
+            "[email protected]": "",
+            "[email protected]": "Brigham and Women's Hospital",
+            "[email protected]": "Beth Israel Deaconess Medical Center",
+        }
+
+        add_institute_proc = test_utils.new_add_institution_processor()
+
+        for pi_email, answer in answers.items():
+            self.assertEqual(
+                add_institute_proc._get_institution_from_pi(institute_map, pi_email),
+                answer,
+            )

process_report/tests/unit/processors/test_bu_subsidy_processor.py

@@ -0,0 +1,177 @@
+from unittest import TestCase
+import pandas
+
+from process_report.tests import util as test_utils
+
+
+class TestBUSubsidyProcessor(TestCase):
+    def _assert_result_invoice(
+        self,
+        subsidy_amount,
+        test_invoice,
+        answer_invoice,
+        invoice_month="0000-00",
+    ):
+        new_bu_subsidy_proc = test_utils.new_bu_subsidy_processor(
+            invoice_month=invoice_month,
+            data=test_invoice,
+            subsidy_amount=subsidy_amount,
+        )
+        new_bu_subsidy_proc.process()
+        output_invoice = new_bu_subsidy_proc.data
+        answer_invoice = answer_invoice.astype(output_invoice.dtypes)
+
+        self.assertTrue(output_invoice.equals(answer_invoice))
+
+    def _get_test_invoice(
+        self,
+        pi,
+        pi_balances,
+        balances=None,
+        project_names=None,
+        institution=None,
+        is_billable=None,
+        missing_pi=None,
+    ):
+        if not balances:
+            balances = pi_balances
+
+        if not project_names:
+            project_names = ["Project" for _ in range(len(pi))]
+
+        if not institution:
+            institution = ["Boston University" for _ in range(len(pi))]
+
+        if not is_billable:
+            is_billable = [True for _ in range(len(pi))]
+
+        if not missing_pi:
+            missing_pi = [False for _ in range(len(pi))]
+
+        return pandas.DataFrame(
+            {
+                "Manager (PI)": pi,
+                "Project - Allocation": project_names,
+                "PI Balance": pi_balances,
+                "Balance": balances,
+                "Institution": institution,
+                "Is Billable": is_billable,
+                "Missing PI": missing_pi,
+            }
+        )
+
+    def test_exclude_non_BU_pi(self):
+        """Are only BU PIs given the subsidy?"""
+
+        subsidy_amount = 100
+        test_invoice = self._get_test_invoice(
+            [str(i) for i in range(5)],
+            pi_balances=[subsidy_amount for _ in range(5)],
+            institution=[
+                "Boston University",
+                "Boston University",
+                "boston university",
+                "Harvard University",
+                "BU",
+            ],
+        )
+
+        answer_invoice = test_invoice.copy()
+        answer_invoice["Project"] = answer_invoice["Project - Allocation"]
+        answer_invoice["Subsidy"] = [subsidy_amount, subsidy_amount, 0, 0, 0]
+        answer_invoice["PI Balance"] = [
+            0,
+            0,
+            subsidy_amount,
+            subsidy_amount,
+            subsidy_amount,
+        ]
+
+        self._assert_result_invoice(subsidy_amount, test_invoice, answer_invoice)
+
+    def test_exclude_nonbillables(self):
+        """Are nonbillables excluded from the subsidy?"""
+        subsidy_amount = 100
+        test_invoice = self._get_test_invoice(
+            [str(i) for i in range(6)],
+            pi_balances=[subsidy_amount for _ in range(6)],
+            is_billable=[True, True, False, False, True, True],
+            missing_pi=[True, True, False, False, False, False],
+        )
+
+        answer_invoice = test_invoice.copy()
+        answer_invoice["Project"] = answer_invoice["Project - Allocation"]
+        answer_invoice["Subsidy"] = [0, 0, 0, 0, subsidy_amount, subsidy_amount]
+        answer_invoice["PI Balance"] = [
+            subsidy_amount,
+            subsidy_amount,
+            subsidy_amount,
+            subsidy_amount,
+            0,
+            0,
+        ]
+
+        self._assert_result_invoice(subsidy_amount, test_invoice, answer_invoice)
+
+    def test_one_pi_many_allocations(self):
+        """Is subsidy applied properly to BU PI with many allocations?"""
+
+        # Two projects, one allocation each
+        subsidy_amount = 100
+        test_invoice = self._get_test_invoice(
+            ["PI" for i in range(2)],
+            pi_balances=[60, 60],
+            project_names=["P1", "P2"],
+        )
+
+        answer_invoice = test_invoice.copy()
+        answer_invoice["Project"] = answer_invoice["Project - Allocation"]
+        answer_invoice["Subsidy"] = [60, 40]
+        answer_invoice["PI Balance"] = [0, 20]
+
+        self._assert_result_invoice(subsidy_amount, test_invoice, answer_invoice)
+
+        # Two projects, two allocations each
+        test_invoice = self._get_test_invoice(
+            ["PI" for i in range(4)],
+            pi_balances=[40, 40, 40, 40],
+            project_names=["P1-A1", "P1-A1-test", "P2", "P2-"],
+        )
+
+        answer_invoice = test_invoice.copy()
+        answer_invoice["Project"] = ["P1", "P1-A1", "P2", "P2"]
+        answer_invoice["Subsidy"] = [40, 40, 20, 0]
+        answer_invoice["PI Balance"] = [0, 0, 20, 40]
+
+        self._assert_result_invoice(subsidy_amount, test_invoice, answer_invoice)
+
+        # Two allocations, one where PI balance != NERC balance
+        test_invoice = self._get_test_invoice(
+            ["PI" for i in range(2)],
+            pi_balances=[80, 80],
+            project_names=["P1", "P2"],
+            balances=[100, 80],
+        )
+
+        answer_invoice = test_invoice.copy()
+        answer_invoice["Project"] = answer_invoice["Project - Allocation"]
+        answer_invoice["Subsidy"] = [80, 20]
+        answer_invoice["PI Balance"] = [0, 60]
+
+        self._assert_result_invoice(subsidy_amount, test_invoice, answer_invoice)
+
+    def test_two_pi(self):
+        """Is subsidy applied to more than one PI?"""
+        # Each PI has two allocations
+        subsidy_amount = 100
+        test_invoice = self._get_test_invoice(
+            ["PI1", "PI1", "PI2", "PI2"],
+            pi_balances=[80, 80, 40, 40],
+        )
+
+        answer_invoice = test_invoice.copy()
+        answer_invoice["Project"] = answer_invoice["Project - Allocation"]
+        answer_invoice["Subsidy"] = [80, 20, 40, 40]
+        answer_invoice["PI Balance"] = [0, 60, 0, 0]
+
+        self._assert_result_invoice(subsidy_amount, test_invoice, answer_invoice)

process_report/tests/unit/processors/test_lenovo_processor.py

@@ -0,0 +1,22 @@
+from unittest import TestCase
+import pandas
+
+from process_report.tests import util as test_utils
+
+
+class TestLenovoProcessor(TestCase):
+    def test_process_lenovo(self):
+        test_invoice = pandas.DataFrame(
+            {
+                "SU Hours (GBhr or SUhr)": [1, 10, 100, 4, 432, 10],
+            }
+        )
+        answer_invoice = test_invoice.copy()
+        answer_invoice["SU Charge"] = 1
+        answer_invoice["Charge"] = (
+            answer_invoice["SU Hours (GBhr or SUhr)"] * answer_invoice["SU Charge"]
+        )
+
+        lenovo_proc = test_utils.new_lenovo_processor(data=test_invoice)
+        lenovo_proc.process()
+        self.assertTrue(lenovo_proc.data.equals(answer_invoice))