changes based on feedback, added tests

microsetta_private_api/admin/sample_summary.py

@@ -57,6 +57,20 @@ def per_sample(project, barcodes, strip_sampleid):
             sample = diag['sample']
             account = diag['account']
             source = diag['source']
+            first_scans_info = diag['scans_info']
+            last_scans_info = diag['latest_scan']
+            if first_scans_info:
+                first_scan_timestamp = first_scans_info[0]['scan_timestamp']
+                first_scan_status = first_scans_info[0]['sample_status']
+            else:
+                first_scan_timestamp = None
+                first_scan_status = None
+            if last_scans_info:
+                latest_scan_timestamp = last_scans_info['scan_timestamp']
+                latest_scan_status = last_scans_info['sample_status']
+            else:
+                latest_scan_timestamp = None
+                latest_scan_status = None
 
             account_email = None if account is None else account.email
             source_type = None if source is None else source.source_type
@@ -103,15 +117,18 @@ def per_sample(project, barcodes, strip_sampleid):
                     sample.id
                 )
 
-            kit_id_name = sample_repo._get_supplied_kit_id_by_sample(barcode)
-            outbound_fedex_tracking = \
-                admin_repo.get_outbound_tracking_by_barcodes(barcode)
-            inbound_fedex_tracking = \
-                admin_repo.get_inbound_tracking_by_barcodes(barcode)
-            first_scan_timestamp_status = \
-                admin_repo.get_first_scan_timestamp_by_barcodes(barcode)
-            last_scan_timestamp_status = \
-                admin_repo.get_last_scan_timestamp_by_barcodes(barcode)
+            kit_by_barcode = admin_repo.get_kit_by_barcode([barcode])
+
+            if kit_by_barcode and len(kit_by_barcode) > 0:
+                info = kit_by_barcode[0]
+
+                kit_id_name = info['kit_id']
+                outbound_fedex_tracking = info['outbound_tracking']
+                inbound_fedex_tracking = info['inbound_tracking']
+            else:
+                kit_id_name = None
+                outbound_fedex_tracking = None
+                inbound_fedex_tracking = None
 
             summary = {
                 "sampleid": None if strip_sampleid else barcode,
@@ -126,11 +143,13 @@ def per_sample(project, barcodes, strip_sampleid):
                 "ffq-complete": ffq_complete,
                 "sample-status": sample_status,
                 "sample-received": sample_status is not None,
+                "first-scan-timestamp": first_scan_timestamp,
+                "first-scan-status": first_scan_status,
+                "latest-scan-timestamp": latest_scan_timestamp,
+                "latest-scan-status": latest_scan_status,
                 "kit-id": kit_id_name,
                 "outbound-tracking": outbound_fedex_tracking,
                 "inbound-tracking": inbound_fedex_tracking,
-                "first-scan-timestamp-status": first_scan_timestamp_status,
-                "last-scan-timestamp-status": last_scan_timestamp_status
             }
 
             for status in ["sample-is-valid",

microsetta_private_api/admin/tests/test_admin_repo.py

@@ -1487,3 +1487,137 @@ def test_update_perk_fulfillment_state(self):
                 )
                 obs = cur.fetchone()
                 self.assertFalse(obs[0])
+
+    def test_get_barcodes_filter_kit_ids_success(self):
+        with Transaction() as t:
+            setup_sql = """
+                INSERT INTO barcodes.kit (kit_id, box_id)
+                VALUES ('test1', '0001e15f-4170-4b28-b111-191cd567c347');
+
+                INSERT INTO barcodes.barcode (barcode, kit_id)
+                VALUES ('00001234', 'test1');
+            """
+            with t.cursor() as cur:
+                cur.execute(setup_sql)
+
+            admin_repo = AdminRepo(t)
+
+            barcodes = admin_repo.get_barcodes_filter(kit_ids=['test1'])
+            self.assertEqual(barcodes, ['00001234'])
+
+    def test_get_barcodes_filter_kit_ids_failure(self):
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            barcodes = admin_repo.get_barcodes_filter(kit_ids=['notarealkit'])
+            self.assertEqual(barcodes, [])
+
+    def test_get_barcodes_filter_emails_success(self):
+        with Transaction() as t:
+            setup_sql = """
+                INSERT INTO ag.source (id, account_id,
+                                       source_type, source_name)
+                VALUES ('0003ddfd-4949-4105-90a9-1b1530af5352', %s,
+                        'Human', 'Test Source');
+                INSERT INTO barcodes.kit (kit_id, box_id)
+                VALUES ('test1', '0001e15f-4170-4b28-b111-191cd567c347');
+                INSERT INTO barcodes.barcode (barcode, kit_id)
+                VALUES ('00001234', 'test1');
+                INSERT INTO ag.ag_kit_barcodes (barcode, source_id)
+                VALUES ('00001234', '0003ddfd-4949-4105-90a9-1b1530af5352');
+            """
+            with t.cursor() as cur:
+                cur.execute(setup_sql, (STANDARD_ACCT_ID,))
+
+            admin_repo = AdminRepo(t)
+
+            barcodes = admin_repo.get_barcodes_filter(emails=['[email protected]'])
+            self.assertEqual(barcodes, ['00001234'])
+
+    def test_get_barcodes_filter_emails_failure(self):
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            barcodes = admin_repo.get_barcodes_filter(
+                emails=['[email protected]'])
+            self.assertEqual(barcodes, [])
+
+    def test_get_barcodes_filter_outbound_tracking_success(self):
+        with Transaction() as t:
+            setup_sql = """
+                INSERT INTO barcodes.kit (kit_id, box_id,
+                        outbound_fedex_tracking)
+                VALUES ('test1', '0001e15f-4170-4b28-b111-191cd567c347',
+                        '12345');
+                INSERT INTO barcodes.barcode (barcode, kit_id)
+                VALUES ('00001234', 'test1');
+            """
+            with t.cursor() as cur:
+                cur.execute(setup_sql)
+
+            admin_repo = AdminRepo(t)
+
+            barcodes = \
+                admin_repo.get_barcodes_filter(
+                    outbound_tracking_numbers=['12345'])
+            self.assertEqual(barcodes, ['00001234'])
+
+    def test_get_barcodes_filter_outbound_tracking_failure(self):
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            barcodes = admin_repo.get_barcodes_filter(
+                outbound_tracking_numbers=['99999'])
+            self.assertEqual(barcodes, [])
+
+    def test_get_barcodes_filter_inbound_tracking_success(self):
+        with Transaction() as t:
+            setup_sql = """
+                INSERT INTO barcodes.kit (kit_id, box_id,
+                    inbound_fedex_tracking)
+                VALUES ('test1', '0001e15f-4170-4b28-b111-191cd567c347',
+                        '67890');
+                INSERT INTO barcodes.barcode (barcode, kit_id)
+                VALUES ('00001234', 'test1');
+            """
+            with t.cursor() as cur:
+                cur.execute(setup_sql)
+
+            admin_repo = AdminRepo(t)
+
+            barcodes = admin_repo.get_barcodes_filter(
+                inbound_tracking_numbers=['67890'])
+            self.assertEqual(barcodes, ['00001234'])
+
+    def test_get_barcodes_filter_inbound_tracking_failure(self):
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            barcodes = admin_repo.get_barcodes_filter(
+                inbound_tracking_numbers=['99999'])
+            self.assertEqual(barcodes, [])
+
+    def test_get_kit_by_barcode_success(self):
+        with Transaction() as t:
+            setup_sql = """
+                INSERT INTO barcodes.kit (kit_id, box_id)
+                VALUES ('test1', '0001e15f-4170-4b28-b111-191cd567c348');
+
+                INSERT INTO barcodes.barcode (barcode, kit_id)
+                VALUES ('00001234', 'test1');
+            """
+            with t.cursor() as cur:
+                cur.execute(setup_sql)
+
+            admin_repo = AdminRepo(t)
+
+            kit_info = admin_repo.get_kit_by_barcode(['00001234'])
+            expected = [{
+                'barcode': '00001234',
+                'outbound_tracking': None,
+                'inbound_tracking': None,
+                'kit_id': 'test1'
+            }]
+            self.assertEqual(kit_info, expected)
+
+    def test_get_kit_by_barcode_failure(self):
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            kit_info = admin_repo.get_kit_by_barcode(['nonexistent_barcode'])
+            self.assertIsNone(kit_info)

microsetta_private_api/repo/admin_repo.py

@@ -616,24 +616,35 @@ def get_barcodes_filter(self, kit_ids=None, emails=None,
 
         return barcodes
 
-    def get_outbound_tracking_by_barcodes(self, barcodes):
-        """Obtain the outbound tracking numbers associated with a barcode
+    def get_kit_by_barcode(self, barcodes):
+        """Obtain the outbound tracking, inbound tracking numbers,
+        and kit ID associated with a list of barcodes.
 
         Parameters
         ----------
         barcodes : list
-            The list of barcodes to obtain outbound tracking numbers for
+            The list of barcodes to obtain information for.
 
         Returns
         -------
-        list
-            The list of observed outbound tracking numbers
+        list of dict
+            A list of dictionaries with outbound tracking,
+            inbound tracking, and kit ID for each barcode.
         """
         query = """
-            SELECT k.outbound_fedex_tracking
-            FROM barcodes.barcode b
-            JOIN barcodes.kit k ON b.kit_id = k.kit_id
-            WHERE b.barcode IN %s
+            SELECT
+                b.barcode,
+                k.outbound_fedex_tracking,
+                k.inbound_fedex_tracking,
+                k.kit_id
+            FROM
+                barcodes.barcode b
+            JOIN
+                barcodes.kit k
+            ON
+                b.kit_id = k.kit_id
+            WHERE
+                b.barcode IN %s
         """
 
         with self._transaction.cursor() as cur:
@@ -644,112 +655,15 @@ def get_outbound_tracking_by_barcodes(self, barcodes):
             if len(rows) == 0:
                 return None
 
-            return [row[0] for row in rows]
-
-    def get_inbound_tracking_by_barcodes(self, barcodes):
-        """Obtain the inbound tracking numbers associated with a barcode
-
-        Parameters
-        ----------
-        barcodes : list
-            The list of barcodes to obtain inbound tracking numbers for
-
-        Returns
-        -------
-        list
-            The list of observed inbound tracking numbers
-        """
-        query = """
-            SELECT k.inbound_fedex_tracking
-            FROM barcodes.barcode b
-            JOIN barcodes.kit k ON b.kit_id = k.kit_id
-            WHERE b.barcode IN %s
-        """
-
-        with self._transaction.cursor() as cur:
-            cur.execute(query, [tuple(barcodes)])
-
-            rows = cur.fetchall()
-
-            if len(rows) == 0:
-                return None
-
-            return [row[0] for row in rows]
-
-    def get_first_scan_timestamp_by_barcodes(self, barcodes):
-        """Obtain the first scan timestamp and
-           sample status associated with a barcode
-
-        Parameters
-        ----------
-        barcodes : list
-            The list of barcodes to obtain first
-            scan timestamps and sample statuses for
-
-        Returns
-        -------
-        list
-            A list of tuples where each tuple contains
-            the first scan timestamp and sample status
-        """
-
-        if isinstance(barcodes, str):
-            barcodes = [barcodes]
-
-        query = """
-            SELECT MIN(scan_timestamp), sample_status
-            FROM barcodes.barcode_scans
-            WHERE barcode = %s
-            GROUP BY sample_status
-            ORDER BY MIN(scan_timestamp)
-            LIMIT 1
-        """
-
-        with self._transaction.cursor() as cur:
-            for barcode in barcodes:
-                cur.execute(query, [barcode])
-                result = cur.fetchone()
-                if result:
-                    results = result[0], result[1]
-
-                    return results
-
-    def get_last_scan_timestamp_by_barcodes(self, barcodes):
-        """Obtain the last scan timestamp and
-           sample status associated with a barcode
-
-        Parameters
-        ----------
-        barcodes : list
-            The list of barcodes to obtain last
-            scan timestamps and sample statuses for
-
-        Returns
-        -------
-        list
-            A list of tuples where each tuple contains
-            the last scan timestamp and sample status
-        """
-
-        if isinstance(barcodes, str):
-            barcodes = [barcodes]
-
-        query = """
-            SELECT MAX(scan_timestamp), sample_status
-            FROM barcodes.barcode_scans
-            WHERE barcode = %s
-            GROUP BY sample_status
-            ORDER BY MAX(scan_timestamp) DESC
-            LIMIT 1
-        """
-        results = []
-        with self._transaction.cursor() as cur:
-            for barcode in barcodes:
-                cur.execute(query, [barcode])
-                result = cur.fetchone()
-                if result:
-                    results.append((result[0], result[1]))
-                    return results
+            return [
+                {
+                    "barcode": row[0],
+                    "outbound_tracking": row[1],
+                    "inbound_tracking": row[2],
+                    "kit_id": row[3]
+                }
+                for row in rows
+            ]
 
     def create_project(self, project):
         """Create a project entry in the database