Merge pull request #249 from wasade/override_associate_lockout

Admin override for associate sample

microsetta_private_api/api/_sample.py

@@ -28,12 +28,15 @@ def read_sample_associations(account_id, source_id, token_info):
 def associate_sample(account_id, source_id, body, token_info):
     _validate_account_access(token_info, account_id)
 
+    is_admin = token_grants_admin_access(token_info)
     with Transaction() as t:
         sample_repo = SampleRepo(t)
         sample_repo.associate_sample(account_id,
                                      source_id,
-                                     body['sample_id'])
+                                     body['sample_id'],
+                                     override_locked=is_admin)
         t.commit()
+
     response = flask.Response()
     response.status_code = 201
     response.headers['Location'] = '/api/accounts/%s/sources/%s/samples/%s' % \

microsetta_private_api/api/tests/test_api.py

@@ -434,7 +434,7 @@ def create_dummy_answered_survey(dummy_acct_id, dummy_source_id,
     return survey_answers_id
 
 
-def create_dummy_kit(account_id=None, source_id=None):
+def create_dummy_kit(account_id=None, source_id=None, associate_sample=True):
     with Transaction() as t:
         _create_mock_kit(t, barcodes=[BARCODE],
                          mock_sample_ids=[MOCK_SAMPLE_ID])
@@ -445,8 +445,11 @@ def create_dummy_kit(account_id=None, source_id=None):
 
             sample_info, _ = create_dummy_sample_objects(True)
             sample_repo = SampleRepo(t)
-            sample_repo.associate_sample(account_id, source_id, MOCK_SAMPLE_ID)
-            sample_repo.update_info(account_id, source_id, sample_info)
+
+            if associate_sample:
+                sample_repo.associate_sample(account_id, source_id,
+                                             MOCK_SAMPLE_ID)
+                sample_repo.update_info(account_id, source_id, sample_info)
 
         t.commit()
 
@@ -1351,6 +1354,58 @@ def test_associate_answered_survey_to_sample_success(self):
         #  that fail with with a 401 for answered survey not found and
         #  a 401 for sample not found
 
+    def test_associate_sample_locked(self):
+        dummy_acct_id, dummy_source_id = create_dummy_source(
+            "Bo", Source.SOURCE_TYPE_HUMAN, DUMMY_HUMAN_SOURCE,
+            create_dummy_1=True)
+
+        create_dummy_kit(dummy_acct_id, dummy_source_id,
+                         associate_sample=False)
+        _ = create_dummy_answered_survey(
+            dummy_acct_id, dummy_source_id)
+
+        base_url = '/api/accounts/{0}/sources/{1}/samples'.format(
+            dummy_acct_id, dummy_source_id)
+
+        # "scan" the sample in
+        _ = create_dummy_acct(create_dummy_1=True,
+                              iss=ACCT_MOCK_ISS_3,
+                              sub=ACCT_MOCK_SUB_3,
+                              dummy_is_admin=True)
+        post_resp = self.client.post('/api/admin/scan/%s' % BARCODE,
+                                     json={'sample_status': 'sample-is-valid',
+                                           'technician_notes': "foobar"},
+                                     headers=make_headers(FAKE_TOKEN_ADMIN))
+        self.assertEqual(201, post_resp.status_code)
+
+        # attempt to associate as a regular user
+        post_resp = self.client.post(
+            '%s?%s' % (base_url, self.default_lang_querystring),
+            content_type='application/json',
+            data=json.dumps(
+                {
+                    'sample_id': MOCK_SAMPLE_ID,
+                }),
+            headers=self.dummy_auth
+        )
+
+        # check response code
+        self.assertEqual(422, post_resp.status_code)
+
+        # associate as admin user
+        post_resp = self.client.post(
+            '%s?%s' % (base_url, self.default_lang_querystring),
+            content_type='application/json',
+            data=json.dumps(
+                {
+                    'sample_id': MOCK_SAMPLE_ID,
+                }),
+            headers=make_headers(FAKE_TOKEN_ADMIN)
+        )
+
+        # check response code
+        self.assertEqual(201, post_resp.status_code)
+
     def test_dissociate_sample_from_source_locked(self):
         dummy_acct_id, dummy_source_id = create_dummy_source(
             "Bo", Source.SOURCE_TYPE_HUMAN, DUMMY_HUMAN_SOURCE,

microsetta_private_api/repo/sample_repo.py

@@ -176,7 +176,8 @@ def update_info(self, account_id, source_id, sample_info,
                             sample_info.id
                         ))
 
-    def associate_sample(self, account_id, source_id, sample_id):
+    def associate_sample(self, account_id, source_id, sample_id,
+                         override_locked=False):
         with self._transaction.cursor() as cur:
             cur.execute("SELECT "
                         "ag_kit_barcode_id, "
@@ -193,7 +194,6 @@ def associate_sample(self, account_id, source_id, sample_id):
             if row is None:
                 raise werkzeug.exceptions.NotFound("No sample ID: %s" %
                                                    sample_id)
-
             if row[2] is not None:
                 if row[1] != account_id:
                     # This is the case where the sample is already assigned in
@@ -205,9 +205,8 @@ def associate_sample(self, account_id, source_id, sample_id):
                     raise RepoException("Sample is already assigned")
             else:
                 # This is the case where the sample is not yet assigned
-                # NOTE: we are not passing override_locked here as a sample
-                # that is not yet assigned cannot be locked.
-                self._update_sample_association(sample_id, source_id)
+                self._update_sample_association(sample_id, source_id,
+                                                override_locked=override_locked)  # noqa
 
     def dissociate_sample(self, account_id, source_id, sample_id,
                           override_locked=False):