# SPDX-License-Identifier: AGPL-3.0-or-later
#
# Cata-Log - the central hub for digital flyers
# Copyright (C) 2026 David Aderbauer & The Cata-Log Contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from datetime import UTC, datetime
from hashlib import sha256
import pytest
import sqlalchemy.exc
from PIL import Image
from sqlalchemy.sql import text
from cata_log_hub import database, exceptions
from cata_log_hub.constants import StatusEnum
from cata_log_hub.settings import get_settings
from test.cata_log_hub.conftest import SideEffects
[docs]
def test_sqlite_pragmas(LocalSession):
with LocalSession() as db_session:
assert db_session.execute(text("PRAGMA foreign_keys;")).scalar() == 1
assert db_session.execute(text("PRAGMA busy_timeout;")).scalar() == 5000
assert db_session.execute(text("PRAGMA synchronous;")).scalar() == 1
[docs]
def test_Provider_insertion(LocalSession, started_scheduler, provider_test_class):
with LocalSession() as db_session:
provider = database.Provider(
class_uid=provider_test_class.uid,
configuration=dict(provider_test_class.default_configuration),
)
db_session.add(provider)
db_session.commit()
db_session.refresh(provider)
assert provider.id
assert provider.class_uid == provider_test_class.uid
assert provider.configuration == provider_test_class.default_configuration
assert provider.job_id
job = started_scheduler.get_job(provider.job_id)
assert job
assert job.func.__name__ == "fetch_provider"
assert len(job.args) == 1
assert 1 in job.args
assert provider.created_at
assert provider.created_at.tzinfo == UTC
assert provider.updated_at
assert provider.updated_at.tzinfo == UTC
[docs]
def test_Provider_insertion__bad_class_uid(LocalSession):
with LocalSession() as db_session:
provider = database.Provider(
class_uid="unknown1234",
configuration={},
)
db_session.add(provider)
db_session.commit()
db_session.refresh(provider)
assert provider.id
assert provider.class_uid == "unknown1234"
assert provider.configuration == {}
assert not provider.job_id
assert provider.created_at
assert provider.updated_at
[docs]
def test_Provider_deletion(db_session, full_database, fake_provider):
pass
[docs]
@pytest.mark.parametrize(
"side_effect",
[
SideEffects.DONOTHING,
],
)
def test_Provider_fetch_catalog__success(
db_session, fake_provider, provider_test_class, side_effect
):
fake_provider.configuration["side_effect"] = side_effect
db_session.commit()
fake_provider.fetch_catalog(db_session)
db_session.refresh(fake_provider)
assert fake_provider.status == StatusEnum.HEALTHY
assert len(fake_provider.catalogs) == 1
assert fake_provider.catalogs[
0
].valid_until == provider_test_class.const_valid_since.replace(
tzinfo=provider_test_class.region.timezone
)
assert len(fake_provider.catalogs[0].pages) == 10
for page in fake_provider.catalogs[0].pages:
assert page.file.sha256 == sha256(page.file.path.read_bytes()).hexdigest()
with Image.open(page.file.path) as page_image:
assert page_image.height == page.file.height
assert page_image.width == page.file.width
assert page_image.format == "WEBP"
[docs]
@pytest.mark.parametrize(
"side_effect",
[
SideEffects.TRANSPORTERROR,
],
)
def test_Provider_fetch_catalog__networkerror(db_session, fake_provider, side_effect):
fake_provider.configuration = {
**fake_provider.configuration,
"side_effect": side_effect,
}
db_session.commit()
with pytest.raises(exceptions.NetworkError):
fake_provider.fetch_catalog(db_session)
[docs]
@pytest.mark.parametrize(
("side_effect", "expected_warning"),
[
(SideEffects.HTTP_400, exceptions.ProviderMisconfiguredOrBrokenWarning),
(SideEffects.HTTP_500, exceptions.ProviderMisconfiguredOrBrokenWarning),
(SideEffects.HTTP_404, exceptions.ProviderMisconfiguredOrBrokenWarning),
(SideEffects.KEYERROR, exceptions.ProviderBrokenWarning),
(SideEffects.VALUEERROR, exceptions.ProviderBrokenWarning),
(SideEffects.EXCEPTION, exceptions.ProviderBrokenWarning),
(SideEffects.CATALOGUNAVAILABLE, exceptions.CatalogUnavailableWarning),
],
)
def test_Provider_fetch_catalog__warning(
db_session, fake_provider, side_effect, expected_warning
):
fake_provider.configuration = {
**fake_provider.configuration,
"side_effect": side_effect,
}
db_session.commit()
fake_provider.fetch_catalog(db_session)
db_session.refresh(fake_provider)
assert fake_provider.status == expected_warning.provider_status
assert not fake_provider.catalogs
[docs]
def test_Catalog_insertion(LocalSession, fake_provider):
with LocalSession() as db_session:
catalog = database.Catalog(
valid_since=datetime.now(tz=UTC),
valid_until=datetime.now(tz=UTC),
provider_id=fake_provider.id,
)
db_session.add(catalog)
db_session.commit()
db_session.refresh(catalog)
assert catalog.id
assert catalog.valid_since
assert catalog.valid_since.tzinfo == UTC
assert catalog.valid_until
assert catalog.valid_until.tzinfo == UTC
assert catalog.provider.id == fake_provider.id
assert catalog.provider.class_uid == fake_provider.class_uid
assert catalog.provider.configuration == fake_provider.configuration
assert len(fake_provider.catalogs) == 1
assert fake_provider.catalogs[0].id == catalog.id
assert fake_provider.catalogs[0].valid_since == catalog.valid_since
assert catalog.created_at
assert catalog.created_at.tzinfo == UTC
assert catalog.updated_at
assert catalog.updated_at.tzinfo == UTC
[docs]
def test_Catalog_deletion(db_session, fake_catalog, full_database):
assert len(db_session.query(database.Catalog).all()) == 3
db_session.delete(fake_catalog)
db_session.commit()
assert len(db_session.query(database.Catalog).all()) == 2
assert not db_session.query(database.Page).all()
[docs]
def test_PageFile_insertion(faker, LocalSession):
with LocalSession() as db_session:
pagefile = database.PageFile(
path=get_settings().storage_path / "testfile.jpg",
sha256=faker.sha256(),
original_sha256=faker.sha256(),
size=1,
width=1,
height=1,
)
db_session.add(pagefile)
db_session.commit()
db_session.refresh(pagefile)
assert pagefile.id
assert pagefile.sha256
assert pagefile.original_sha256
assert pagefile.width
assert pagefile.height
assert pagefile.path == get_settings().storage_path / "testfile.jpg"
assert pagefile.created_at
assert pagefile.created_at.tzinfo == UTC
assert pagefile.updated_at
assert pagefile.updated_at.tzinfo == UTC
[docs]
@pytest.mark.parametrize(
("bad_key", "bad_value"),
[
("size", -101),
("width", -52),
("height", -67),
("size", 0),
("width", 0),
("height", 0),
],
)
def test_PageFile_insertion__bad_values(faker, bad_key, bad_value):
values = {
"path": get_settings().storage_path / "testfile.jpg",
"sha256": faker.sha256(),
"original_sha256": faker.sha256(),
"size": faker.random.randint(1, 2000),
"width": faker.random.randint(100, 1000),
"height": faker.random.randint(100, 1000),
}
values[bad_key] = bad_value
with pytest.raises(ValueError, match=rf"{bad_value}"):
database.PageFile(**values)
[docs]
def test_PageFile_deletion(db_session, fake_file, fake_pagefile):
assert fake_file.exists()
db_session.delete(fake_pagefile)
db_session.commit()
assert not fake_file.exists()
[docs]
def test_PageFile_deletion__restricted(db_session, fake_file, fake_pagefile, fake_page):
assert fake_file.exists()
assert len(fake_pagefile.pages)
db_session.delete(fake_pagefile)
with pytest.raises(
sqlalchemy.exc.IntegrityError, match="FOREIGN KEY constraint failed"
):
db_session.commit()
assert fake_file.exists()
[docs]
def test_Page_insertion(LocalSession, fake_catalog, fake_pagefile):
with LocalSession() as db_session:
page = database.Page(
number=0,
catalog_id=fake_catalog.id,
file_id=fake_pagefile.id,
)
db_session.add(page)
db_session.commit()
db_session.refresh(page)
assert page.id
assert page.number == 0
assert page.file.id == fake_pagefile.id
assert len(fake_pagefile.pages) == 1
assert fake_pagefile.pages[0].id == page.id
assert page.catalog.id == fake_catalog.id
assert page.catalog.valid_since == fake_catalog.valid_since
assert len(fake_catalog.pages) == 1
assert fake_catalog.pages[0].id == page.id
assert fake_catalog.pages[0].number == page.number
assert page.created_at
assert page.created_at.tzinfo == UTC
assert page.updated_at
assert page.updated_at.tzinfo == UTC
[docs]
@pytest.mark.parametrize(
("bad_key", "bad_value"),
[
("number", -101),
("number", -1),
],
)
def test_Page_insertion__bad_values(fake_catalog, fake_pagefile, bad_key, bad_value):
values = {
"number": 4,
"catalog_id": fake_catalog.id,
"file_id": fake_pagefile.id,
}
values[bad_key] = bad_value
with pytest.raises(ValueError, match=rf"{bad_value}"):
database.Page(**values)
[docs]
def test_Page_unique_together_constraint(LocalSession, fake_pagefile, fake_page):
with LocalSession() as db_session:
db_session.add(
database.Page(
number=fake_page.number,
catalog_id=fake_page.catalog_id,
file_id=fake_pagefile.id,
)
)
with pytest.raises(sqlalchemy.exc.IntegrityError):
db_session.flush()
[docs]
def test_Page_deletion(db_session, fake_pagefile, fake_page):
pagefile_id = fake_pagefile.id
assert len(fake_pagefile.pages) == 1
db_session.delete(fake_page)
db_session.commit()
assert not db_session.get(database.PageFile, pagefile_id)
[docs]
def test_Page_deletion__remaining_pagefile_pages(
faker, db_session, fake_pagefile, fake_page
):
other_page = database.Page(
number=faker.random.randint(0, 10),
catalog_id=fake_page.catalog_id,
file_id=fake_pagefile.id,
)
db_session.add(other_page)
db_session.commit()
pagefile_id = fake_pagefile.id
assert len(fake_pagefile.pages) == 2
db_session.delete(fake_page)
db_session.commit()
assert db_session.get(database.PageFile, pagefile_id)
assert len(fake_pagefile.pages) == 1
