test: add tests for external_pids

Signed-off-by: Nicolas Harraudeau <[email protected]>

tests/b2share_unit_tests/deposit/test_deposit_api.py

@@ -218,3 +218,39 @@ def test_change_deposit_schema_fails(app, draft_deposits):
         del deposit['$schema']
         with pytest.raises(AlteredRecordError):
             deposit.commit()
+
+
+def test_create_deposit_with_external_pids(app, deposit_with_external_pids):
+    expected_files = deposit_with_external_pids.data['_deposit']['external_pids']
+    expected_uris = { exp['ePIC_PID'] for exp in expected_files }
+    expected_keys = { exp['key'] for exp in expected_files }
+    with app.app_context():
+        files = [f for f in deposit_with_external_pids.get_deposit().files]
+        assert len(files) == len(expected_files)
+        # Check that the URIs match
+        uris = {file.obj.file.uri for file in files}
+        assert uris == expected_uris
+        # Check that the keys match
+        keys = {file.obj.key for file in files}
+        assert keys == expected_keys
+        # Check that all files are redirect files
+        assert all([file.obj.file.storage_class == 'B' for file in files])
+        # Check that the files are deduplicated in the DB
+        file_ids = {file.obj.file_id for file in files}
+        assert len(file_ids) == len(uris)
+
+
+def test_modify_record_adding_b2safe_files(app):
+    # create an empty record
+    # modify it to add 2 b2safe files
+    # modify it to remove 1 and add 1
+    # modify it to remove all
+    pass
+
+
+def test_record_with_both_types_of_files(app):
+    # create a record with a b2safe file
+    # upload a normal record
+    # modify to change the b2safe file
+    # upload a second normal file
+    pass

tests/b2share_unit_tests/files/test_b2share_storage.py

@@ -39,65 +39,26 @@
 from invenio_db import db
 
 
-def test_deposit_create(app, tmp_location, login_user, test_users):
-    """Test deposit creation."""
-    with app.app_context():
-        tmp_location = Location.query.first()
-        with db.session.begin_nested():
-            b1 = Bucket.create(tmp_location, storage_class='B')
-            f1 = FileInstance.create()
-            f1.set_uri('http://hdl.handle.net/11304/74c66f0b-f814-4202-9dcb-4889ba9b1047',
-                       1, 0, storage_class='B')
-            # f2 = FileInstance(uri="f2", size=1,
-            #                   checksum="mychecksum", storage_class='S')
-            # f2.create
-            # f2.set_uri('http://hdl.handle.net/11304/74c66f0b-f814-4202-9dcb-4889ba9b1047',
-                       # 1, 0, storage_class='B2SAFE')
-            ObjectVersion.create(b1, 'test.txt', f1.id)
-            # ObjectVersion.create(b1, 'test2.txt', f2.id)
-        db.session.commit()
-
-        with app.test_client() as client:
-            login_user(test_users['normal'], client)
-            resp = client.get(
-                url_for('invenio_files_rest.object_api',
-                        bucket_id=b1.id,
-                        key='test.txt',
-                        follow_redirects=True)
-            )
-            print(resp.data)
-
-
-# move these to the functional tests
-def test_create_record_with_b2safe_files(app):
-    # create a record which has a file_pids field in the metadata
-    pass
-
-
-def test_modify_record_adding_b2safe_files(app):
-    # create an empty record
-    # modify it to add 2 b2safe files
-    # modify it to remove 1 and add 1
-    # modify it to remove all
-    pass
-
-
-def test_record_with_both_types_of_files(app):
-    # create a record with a b2safe file
-    # upload a normal record
-    # modify to change the b2safe file
-    # upload a second normal file
-    pass
-
-
-def test_getting_a_b2safe_file(app, tmp_location):
-    # create a b2safe file
-    # query to get the contents of the b2safe file
-    # check that the redirect works
+def test_b2share_storage_with_pid(app, tmp_location, login_user, test_users):
+    """Check that the storage class will redirect pid files."""
+    pid = 'http://hdl.handle.net/11304/74c66f0b-f814-4202-9dcb-4889ba9b1047'
+    # Disable access control for this test
+    app.config.update({
+        'FILES_REST_PERMISSION_FACTORY': 'invenio_records_rest.utils:allow_all'
+    })
     with app.app_context():
         tmp_location = Location.query.first()
         with db.session.begin_nested():
             bucket = Bucket.create(tmp_location, storage_class='B')
-            b2safe_file = FileInstance.create()
-            b2safe_file.set_uri('http://hdl.handle.net/11304/74c66f0b-f814-4202-9dcb-4889ba9b1047',
-                                1, 0, storage_class='B')
+            pid_file = FileInstance.create()
+            pid_file.set_uri(pid, 1, 0, storage_class='B')
+            ObjectVersion.create(bucket, 'test.txt', pid_file.id)
+        db.session.commit()
+        url = url_for('invenio_files_rest.object_api',
+                        bucket_id=bucket.id,
+                        key='test.txt')
+    # Check that accessing the file redirects to the PID
+    with app.test_client() as client:
+        resp = client.get(url)
+        assert resp.headers['Location'] == pid
+        assert resp.status_code == 302

tests/conftest.py

@@ -344,6 +344,40 @@ def test_records_data(app, test_communities):
         ]
 
 
+@pytest.fixture(scope='function')
+def deposit_with_external_pids(app, test_communities, test_users):
+    """Create a deposit with external pids."""
+    record_data = json.dumps({
+        "external_pids":[
+            {
+                "key":"file1.txt",
+                "ePIC_PID": "http://hdl.handle.net/11304/0d8dbdec-74e4-4774-954e-1a98e5c0cfa3"
+            }, {
+                "key":"file1_copy.txt",
+                "ePIC_PID": "http://hdl.handle.net/11304/0d8dbdec-74e4-4774-954e-1a98e5c0cfa3"
+            }, {
+                "key":"file2.txt",
+                "ePIC_PID": "http://hdl.handle.net/11304/50fafc50-4227-4464-bacc-2d85295c18a7"
+            }
+        ],
+        'titles': [{'title':'BBMRI dataset 6'}],
+        'community': '$COMMUNITY_ID[MyTestCommunity2]',
+        'open_access': True,
+        'community_specific': {
+            '$BLOCK_SCHEMA_ID[MyTestSchema]': {
+                'study_design': ['Case-control']
+            }
+        }
+    })
+    with app.app_context():
+        data = json.loads(resolve_block_schema_id(resolve_community_id(
+            record_data)))
+        result = create_deposits(app, [data],
+                                 test_users['deposits_creator'])[0]
+        db.session.commit()
+        return result
+
+
 @pytest.fixture(scope='function')
 def test_incomplete_records_data(app, test_communities):
     """Create incomplete record data and the corresponding patch."""
@@ -400,7 +434,13 @@ def test_incomplete_records_data(app, test_communities):
 
 def create_deposits(app, test_records_data, creator):
     """Create test deposits."""
-    DepositInfo = namedtuple('DepositInfo', ['deposit_id', 'data', 'deposit'])
+    DepositInfo = namedtuple(
+        'DepositInfo', [
+            'deposit_id',
+            'data',
+            'deposit', # FIXME: replaced by get_deposit, remove it later
+            'get_deposit'
+        ])
 
     deposits = []
     with authenticated_user(creator):
@@ -409,7 +449,10 @@ def create_deposits(app, test_records_data, creator):
             # Create persistent identifier
             b2share_deposit_uuid_minter(record_uuid, data=data)
             deposits.append(B2ShareDeposit.create(data=data, id_=record_uuid))
-    return [DepositInfo(dep.id, dep.dumps(), dep) for dep in deposits]
+    return [DepositInfo(
+        dep.id, dep.dumps(), dep,
+        (lambda id: lambda: B2ShareDeposit.get_record(id))(dep.id)
+    ) for dep in deposits]
 
 
 @pytest.fixture(scope='function')