diff --git a/odp/db/models/archive.py b/odp/db/models/archive.py index 16d6cb6..e2f90f1 100644 --- a/odp/db/models/archive.py +++ b/odp/db/models/archive.py @@ -31,8 +31,8 @@ class ArchiveResource(Base): archive_id = Column(String, ForeignKey('archive.id', ondelete='RESTRICT'), primary_key=True) resource_id = Column(String, ForeignKey('resource.id', ondelete='CASCADE'), primary_key=True) - archive = relationship('Archive', viewonly=True) - resource = relationship('Resource', viewonly=True) + archive = relationship('Archive') + resource = relationship('Resource') path = Column(String, nullable=False) timestamp = Column(TIMESTAMP(timezone=True), nullable=False) diff --git a/test/api/test_archive.py b/test/api/test_archive.py new file mode 100644 index 0000000..ad9703a --- /dev/null +++ b/test/api/test_archive.py @@ -0,0 +1,77 @@ +from random import randint + +import pytest +from sqlalchemy import select + +from odp.const import ODPScope +from odp.db.models import Archive +from test import TestSession +from test.api import assert_forbidden, assert_not_found +from test.factories import ArchiveFactory, ArchiveResourceFactory + + +@pytest.fixture +def archive_batch(): + """Create and commit a batch of Archive instances.""" + archives = [ + ArchiveFactory() + for _ in range(randint(3, 5)) + ] + for archive in archives: + archive.resource_count = randint(0, 3) + ArchiveResourceFactory.create_batch(archive.resource_count, archive=archive) + return archives + + +def assert_db_state(archives): + """Verify that the DB archive table contains the given archive batch.""" + result = TestSession.execute(select(Archive)).scalars().all() + assert set((row.id, row.url) for row in result) \ + == set((archive.id, archive.url) for archive in archives) + + +def assert_json_result(response, json, archive): + """Verify that the API result matches the given archive object.""" + assert response.status_code == 200 + assert json['id'] == archive.id + assert json['url'] == archive.url + assert json['resource_count'] == archive.resource_count + + +def assert_json_results(response, json, archives): + """Verify that the API result list matches the given archive batch.""" + items = json['items'] + assert json['total'] == len(items) == len(archives) + items.sort(key=lambda i: i['id']) + archives.sort(key=lambda a: a.id) + for n, archive in enumerate(archives): + assert_json_result(response, items[n], archive) + + +@pytest.mark.require_scope(ODPScope.ARCHIVE_READ) +def test_list_archives(api, archive_batch, scopes): + authorized = ODPScope.ARCHIVE_READ in scopes + r = api(scopes).get('/archive/') + if authorized: + assert_json_results(r, r.json(), archive_batch) + else: + assert_forbidden(r) + assert_db_state(archive_batch) + + +@pytest.mark.require_scope(ODPScope.ARCHIVE_READ) +def test_get_archive(api, archive_batch, scopes): + authorized = ODPScope.ARCHIVE_READ in scopes + r = api(scopes).get(f'/archive/{archive_batch[2].id}') + if authorized: + assert_json_result(r, r.json(), archive_batch[2]) + else: + assert_forbidden(r) + assert_db_state(archive_batch) + + +def test_get_archive_not_found(api, archive_batch): + scopes = [ODPScope.ARCHIVE_READ] + r = api(scopes).get('/archive/foo') + assert_not_found(r) + assert_db_state(archive_batch) diff --git a/test/factories.py b/test/factories.py index 10c1c32..f094045 100644 --- a/test/factories.py +++ b/test/factories.py @@ -11,6 +11,7 @@ import odp.db from odp.db.models import ( Archive, + ArchiveResource, Catalog, Client, Collection, @@ -408,3 +409,13 @@ class Meta: id = factory.Sequence(lambda n: f'{fake.slug()}.{n}') url = factory.Faker('url') + + +class ArchiveResourceFactory(ODPModelFactory): + class Meta: + model = ArchiveResource + + archive = factory.SubFactory(ArchiveFactory) + resource = factory.SubFactory(ResourceFactory) + path = factory.LazyAttribute(lambda a: f'/{fake.uri_path(deep=randint(1, 5))}/{a.resource.filename}') + timestamp = factory.LazyFunction(lambda: datetime.now(timezone.utc))