diff --git a/src/onegov/town6/api.py b/src/onegov/town6/api.py index 3dca312a8f..ee6f53dea6 100644 --- a/src/onegov/town6/api.py +++ b/src/onegov/town6/api.py @@ -1,29 +1,55 @@ from __future__ import annotations from functools import cached_property +from uuid import UUID + from onegov.api.models import ApiEndpoint, ApiEndpointItem +from onegov.core.collection import Pagination from onegov.event.collections import OccurrenceCollection +from onegov.form import FormCollection +from onegov.form.models import FormDefinition from onegov.gis import Coordinates from onegov.org.models.directory import ( ExtendedDirectory, ExtendedDirectoryEntry, ExtendedDirectoryEntryCollection) +from onegov.org.models.external_link import ( + ExternalFormLink, ExternalLinkCollection, ExternalResourceLink) +from onegov.org.models.meeting import Meeting, MeetingCollection from onegov.org.models.page import News, NewsCollection, Topic, TopicCollection +from onegov.org.models.parliament import ( + RISCommission, RISCommissionCollection, + RISParliamentarian, RISParliamentarianCollection, + RISParliamentaryGroup, RISParliamentaryGroupCollection) +from onegov.org.models.political_business import ( + PoliticalBusiness, PoliticalBusinessCollection) +from onegov.people import Person +from onegov.people.collections import PersonCollection +from onegov.reservation.collection import ResourceCollection +from onegov.reservation.models import Resource from onegov.town6 import _ +from sqlalchemy import or_ from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.inspection import inspect as sa_inspect -from typing import Any, Self +from typing import Any, Generic, Self, TypeAlias from typing import TYPE_CHECKING if TYPE_CHECKING: + from collections.abc import Callable, Sequence from onegov.town6.app import TownApp from onegov.town6.request import TownRequest from onegov.event.models import Occurrence from onegov.core.orm.mixins import ContentMixin from onegov.core.orm.mixins import TimestampMixin - from typing import TypeVar + from typing import Protocol, TypeVar from onegov.core.collection import PKType + from sqlalchemy.orm import Query T = TypeVar('T') + class SupportsQueryAndById(Protocol[T]): + def query(self) -> Query[T]: ... + def by_id(self, id: PKType) -> T | None: ... + def get_geo_location(item: ContentMixin) -> dict[str, Any]: geo = item.content.get('coordinates', Coordinates()) or Coordinates() @@ -40,6 +66,221 @@ def get_modified_iso_format(item: TimestampMixin) -> str: return item.last_change.isoformat() +API_BATCH_SIZE = 25 +MANAGER_ROLES = {'admin', 'editor'} +LISTING_ACCESSES = { + 'member': ('member', 'mtan', 'public'), +} + + +def role_for_request(request: TownRequest) -> str: + return getattr(request.identity, 'role', 'anonymous') + + +def available_accesses(request: TownRequest) -> tuple[str, ...]: + role = role_for_request(request) + if role in MANAGER_ROLES: + return () + return LISTING_ACCESSES.get(role, ('mtan', 'public')) + + +def apply_visibility_filters( + request: TownRequest, + query: Query[T], + model_class: Any, +) -> Query[T]: + accesses = available_accesses(request) + if accesses and hasattr(model_class, 'access'): + query = query.filter(or_( + *( + model_class.meta['access'].astext == access + for access in accesses + ), + model_class.meta['access'].is_(None) + )) + + if ( + role_for_request(request) not in MANAGER_ROLES + and hasattr(model_class, 'publication_started') + and hasattr(model_class, 'publication_ended') + ): + query = query.filter( + model_class.publication_started == True, + model_class.publication_ended == False + ) + + return query + + +class FilteredCollection(Generic[T]): + + def __init__( + self, + request: TownRequest, + collection: Any, + model_class: type[T], + query_transform: Callable[[Query[T]], Query[T]], + ) -> None: + self.request = request + self.collection = collection + self.model_class = model_class + self.query_transform = query_transform + + def query(self) -> Query[T]: + return self.query_transform(self.collection.query()) + + def by_id(self, id: PKType) -> T | None: + try: + if callable(getattr(self.collection, 'by_id', None)): + item = self.collection.by_id(id) + if item is not None: + return item if self.request.is_visible(item) else None + + primary_key = sa_inspect(self.model_class).primary_key[0] + return self.query().filter( + primary_key == self._normalize_id(id) + ).first() + except SQLAlchemyError: + return None + + @staticmethod + def _normalize_id(id: PKType) -> PKType: + if isinstance(id, str): + try: + return UUID(id) + except ValueError: + return id + return id + + +class PaginatedCollection(Pagination[T], Generic[T]): + + def __init__( + self, + collection: SupportsQueryAndById[T], + page: int = 0, + batch_size: int = API_BATCH_SIZE, + ) -> None: + self.collection = collection + self.page = page + self.batch_size = batch_size + + def __eq__(self, other: object) -> bool: + return ( + isinstance(other, self.__class__) + and self.collection is other.collection + and self.page == other.page + ) + + def by_id(self, id: PKType) -> T | None: + return self.collection.by_id(id) + + def subset(self) -> Query[T]: + return self.collection.query() + + def page_by_index(self, index: int) -> PaginatedCollection[T]: + return self.__class__( + self.collection, + page=index, + batch_size=self.batch_size, + ) + + @property + def page_index(self) -> int: + return self.page + + +class PaginatedSumCollection(Pagination[T], Generic[T]): + + def __init__( + self, + collections: Sequence[SupportsQueryAndById[T]], + page: int = 0, + batch_size: int = API_BATCH_SIZE, + ) -> None: + self.collections = tuple(collections) + self.page = page + self.batch_size = batch_size + + def __eq__(self, other: object) -> bool: + return ( + isinstance(other, self.__class__) + and self.collections == other.collections + and self.page == other.page + ) + + def by_id(self, id: PKType) -> T | None: + for collection in self.collections: + item = collection.by_id(id) + if item is not None: + return item + return None + + def subset(self) -> Query[T]: + raise NotImplementedError( + 'PaginatedSumCollection does not expose a single subset query' + ) + + @cached_property + def counts(self) -> tuple[int, ...]: + return tuple( + collection.query().order_by(None).count() + for collection in self.collections + ) + + @cached_property + def subset_count(self) -> int: + return sum(self.counts) + + @cached_property + def batch(self) -> tuple[T, ...]: + offset = self.offset + remaining = self.batch_size + items: list[T] = [] + + for collection, count in zip(self.collections, self.counts): + if remaining <= 0: + break + if offset >= count: + offset -= count + continue + + query = collection.query().offset(offset).limit(remaining) + batch = tuple(query) + items.extend(batch) + remaining -= len(batch) + offset = 0 + + return tuple(items) + + def page_by_index(self, index: int) -> PaginatedSumCollection[T]: + return self.__class__( + self.collections, + page=index, + batch_size=self.batch_size, + ) + + @property + def page_index(self) -> int: + return self.page + + +def visible_collection( + request: TownRequest, + collection: Any, + model_class: type[T], + *, + order_by: Any | None = None, +) -> FilteredCollection[T]: + def transform(query: Query[T]) -> Query[T]: + query = apply_visibility_filters(request, query, model_class) + if order_by is not None: + query = query.order_by(order_by) + return query + + return FilteredCollection(request, collection, model_class, transform) + + class EventApiEndpoint(ApiEndpoint['Occurrence']): app: TownApp endpoint = 'events' @@ -300,3 +541,349 @@ def item_links(self, item: ExtendedDirectoryEntry) -> dict[str, Any]: data['html'] = item # type: ignore return data + + +FormOrExternalLink: TypeAlias = FormDefinition | ExternalFormLink +ResourceOrExternalLink: TypeAlias = Resource | ExternalResourceLink + + +class FormApiEndpoint(ApiEndpoint[FormOrExternalLink]): + app: TownApp + endpoint = 'forms' + + @property + def title(self) -> str: + return self.request.translate(_('Forms')) + + @property + def collection(self) -> Any: + return PaginatedSumCollection( + ( + visible_collection( + self.request, + FormCollection(self.session).definitions, + FormDefinition, + order_by=FormDefinition.order, + ), + visible_collection( + self.request, + ExternalLinkCollection.for_model( + self.session, FormCollection + ), + ExternalFormLink, + order_by=ExternalFormLink.order, + ), + ), + page=self.page or 0, + batch_size=API_BATCH_SIZE, + ) + + def by_id(self, id: PKType) -> FormOrExternalLink | None: + return self.collection.by_id(id) + + def item_data(self, item: FormOrExternalLink) -> dict[str, Any]: + if isinstance(item, ExternalFormLink): + return { + 'title': item.title, + 'lead': item.lead, + 'group': item.group, + 'url': item.url, + 'type': 'external', + } + return { + 'title': item.title, + 'lead': item.lead, + 'text': str(item.text) if item.text else None, + 'group': item.group, + 'type': 'internal', + } + + def item_links(self, item: FormOrExternalLink) -> dict[str, Any]: + if isinstance(item, ExternalFormLink): + return {'html': item.url} + return {'html': item} + + +class ResourceApiEndpoint(ApiEndpoint[ResourceOrExternalLink]): + app: TownApp + endpoint = 'resources' + + @property + def title(self) -> str: + return self.request.translate(_('Resources')) + + @property + def collection(self) -> Any: + return PaginatedSumCollection( + ( + visible_collection( + self.request, + ResourceCollection(self.app.libres_context), + Resource, + order_by=Resource.title, + ), + visible_collection( + self.request, + ExternalLinkCollection.for_model( + self.session, ResourceCollection + ), + ExternalResourceLink, + order_by=ExternalResourceLink.order, + ), + ), + page=self.page or 0, + batch_size=API_BATCH_SIZE, + ) + + def by_id(self, id: PKType) -> ResourceOrExternalLink | None: + return self.collection.by_id(id) + + def item_data( + self, item: ResourceOrExternalLink + ) -> dict[str, Any]: + if isinstance(item, ExternalResourceLink): + return { + 'title': item.title, + 'lead': item.lead, + 'group': item.group, + 'url': item.url, + 'kind': 'external', + } + return { + 'title': item.title, + 'lead': getattr(item, 'lead', None), + 'group': item.group, + 'type': item.type, + 'kind': 'internal', + } + + def item_links( + self, item: ResourceOrExternalLink + ) -> dict[str, Any]: + if isinstance(item, ExternalResourceLink): + return {'html': item.url} + return {'html': item} + + +class PersonApiEndpoint(ApiEndpoint[Person]): + app: TownApp + endpoint = 'people' + + _public_fields: tuple[str, ...] = ( + 'academic_title', + 'born', + 'email', + 'first_name', + 'function', + 'last_name', + 'location_address', + 'location_code_city', + 'notes', + 'organisation', + 'parliamentary_group', + 'phone', + 'phone_direct', + 'political_party', + 'postal_address', + 'postal_code_city', + 'profession', + 'salutation', + 'title', + 'website', + ) + + @property + def title(self) -> str: + return self.request.translate(_('People')) + + @property + def collection(self) -> Any: + return PaginatedCollection( + visible_collection( + self.request, + PersonCollection(self.session), + Person, + ), + page=self.page or 0, + batch_size=API_BATCH_SIZE, + ) + + def item_data(self, item: Person) -> dict[str, Any]: + hidden = self.app.org.hidden_people_fields + data = { + attr: getattr(item, attr, None) + for attr in self._public_fields + if attr not in hidden + } + data['modified'] = get_modified_iso_format(item) + return data + + def item_links(self, item: Person) -> dict[str, Any]: + hidden = self.app.org.hidden_people_fields + result: dict[str, Any] = {'html': item} + if 'picture_url' not in hidden: + result['picture_url'] = item.picture_url + if 'website' not in hidden: + result['website'] = item.website + return result + + +class MeetingApiEndpoint(ApiEndpoint[Meeting]): + app: TownApp + endpoint = 'meetings' + + @property + def title(self) -> str: + return self.request.translate(_('Meetings')) + + @property + def collection(self) -> Any: + return PaginatedCollection( + visible_collection( + self.request, + MeetingCollection(self.session), + Meeting, + ), + page=self.page or 0, + batch_size=API_BATCH_SIZE, + ) + + def item_data(self, item: Meeting) -> dict[str, Any]: + return { + 'title': item.title, + 'start_datetime': ( + item.start_datetime.isoformat() + if item.start_datetime else None + ), + 'end_datetime': ( + item.end_datetime.isoformat() + if item.end_datetime else None + ), + 'address': str(item.address) if item.address else None, + 'description': str(item.description) if item.description else None, + 'audio_link': item.audio_link or None, + 'video_link': item.video_link or None, + } + + def item_links(self, item: Meeting) -> dict[str, Any]: + return {'html': item} + + +class PoliticalBusinessApiEndpoint(ApiEndpoint[PoliticalBusiness]): + app: TownApp + endpoint = 'political_businesses' + + @property + def title(self) -> str: + return self.request.translate(_('Political Businesses')) + + @property + def collection(self) -> Any: + return PaginatedCollection( + visible_collection( + self.request, + PoliticalBusinessCollection(self.request), + PoliticalBusiness, + ), + page=self.page or 0, + batch_size=API_BATCH_SIZE, + ) + + def item_data(self, item: PoliticalBusiness) -> dict[str, Any]: + return { + 'title': item.title, + 'number': item.number, + 'political_business_type': item.political_business_type, + 'status': item.status, + 'entry_date': ( + item.entry_date.isoformat() if item.entry_date else None + ), + 'display_name': item.display_name, + } + + def item_links(self, item: PoliticalBusiness) -> dict[str, Any]: + return {'html': item} + + +class RISParliamentarianApiEndpoint(ApiEndpoint[RISParliamentarian]): + app: TownApp + endpoint = 'parliamentarians' + + @property + def title(self) -> str: + return self.request.translate(_('Parliamentarians')) + + @property + def collection(self) -> Any: + return PaginatedCollection( + RISParliamentarianCollection(self.session), + page=self.page or 0, + batch_size=API_BATCH_SIZE, + ) + + def item_data(self, item: RISParliamentarian) -> dict[str, Any]: + return { + 'first_name': item.first_name, + 'last_name': item.last_name, + 'title': item.title, + 'party': item.party, + 'active': item.active, + } + + def item_links(self, item: RISParliamentarian) -> dict[str, Any]: + return { + 'html': item, + 'picture': item.picture, + } + + +class RISCommissionApiEndpoint(ApiEndpoint[RISCommission]): + app: TownApp + endpoint = 'commissions' + + @property + def title(self) -> str: + return self.request.translate(_('Commissions')) + + @property + def collection(self) -> Any: + return PaginatedCollection( + RISCommissionCollection(self.session), + page=self.page or 0, + batch_size=API_BATCH_SIZE, + ) + + def item_data(self, item: RISCommission) -> dict[str, Any]: + return { + 'name': item.name, + 'description': str(item.description) if item.description else None, + } + + def item_links(self, item: RISCommission) -> dict[str, Any]: + return {'html': item} + + +class RISParliamentaryGroupApiEndpoint(ApiEndpoint[RISParliamentaryGroup]): + app: TownApp + endpoint = 'parliamentary_groups' + + @property + def title(self) -> str: + return self.request.translate(_('Parliamentary groups')) + + @property + def collection(self) -> Any: + return PaginatedCollection( + RISParliamentaryGroupCollection(self.session), + page=self.page or 0, + batch_size=API_BATCH_SIZE, + ) + + def item_data(self, item: RISParliamentaryGroup) -> dict[str, Any]: + return { + 'name': item.name, + 'description': str(item.description) if item.description else None, + } + + def item_links(self, item: RISParliamentaryGroup) -> dict[str, Any]: + return {'html': item} diff --git a/src/onegov/town6/app.py b/src/onegov/town6/app.py index 7d4ad65f4f..c722031fca 100644 --- a/src/onegov/town6/app.py +++ b/src/onegov/town6/app.py @@ -16,7 +16,10 @@ from onegov.org.models.directory import ExtendedDirectory from onegov.town6.api import ( EventApiEndpoint, NewsApiEndpoint, TopicApiEndpoint, - DirectoryEntryApiEndpoint) + DirectoryEntryApiEndpoint, FormApiEndpoint, ResourceApiEndpoint, + PersonApiEndpoint, MeetingApiEndpoint, PoliticalBusinessApiEndpoint, + RISParliamentarianApiEndpoint, RISCommissionApiEndpoint, + RISParliamentaryGroupApiEndpoint) from onegov.town6.custom import get_global_tools from onegov.town6.initial_content import create_new_organisation from onegov.town6.theme import TownTheme @@ -237,6 +240,14 @@ def get_api_endpoints( yield EventApiEndpoint(request, extra_parameters, page) yield NewsApiEndpoint(request, extra_parameters, page) yield TopicApiEndpoint(request, extra_parameters, page) + yield FormApiEndpoint(request, extra_parameters, page) + yield ResourceApiEndpoint(request, extra_parameters, page) + yield PersonApiEndpoint(request, extra_parameters, page) + yield MeetingApiEndpoint(request, extra_parameters, page) + yield PoliticalBusinessApiEndpoint(request, extra_parameters, page) + yield RISParliamentarianApiEndpoint(request, extra_parameters, page) + yield RISCommissionApiEndpoint(request, extra_parameters, page) + yield RISParliamentaryGroupApiEndpoint(request, extra_parameters, page) directories = request.exclude_invisible( request.session.query(ExtendedDirectory).all()) for directory in directories: diff --git a/tests/onegov/town6/test_api.py b/tests/onegov/town6/test_api.py new file mode 100644 index 0000000000..4ecd7710e3 --- /dev/null +++ b/tests/onegov/town6/test_api.py @@ -0,0 +1,473 @@ +from __future__ import annotations + +from datetime import timedelta + +import transaction + +from onegov.form import FormCollection +from onegov.org.models.external_link import ( + ExternalFormLink, ExternalResourceLink) +from onegov.org.models.meeting import Meeting +from onegov.org.models.parliament import ( + RISCommission, RISParliamentarian, RISParliamentaryGroup) +from onegov.org.models.political_business import PoliticalBusiness +from onegov.people import Person +from onegov.reservation import ResourceCollection +from sedate import utcnow + +from typing import Any, TYPE_CHECKING +if TYPE_CHECKING: + from .conftest import Client, TestTownApp + + +def api_item_data(item: dict[str, Any]) -> dict[str, Any]: + return { + entry['name']: entry['value'] + for entry in item['data'] + } + + +def api_items( + client: Client, + url: str, + page: int | None = None, +) -> list[dict[str, Any]]: + page_param = '' if page is None else f'?page={page}' + response = client.get(f'{url}{page_param}') + assert response.status_code == 200 + return response.json['collection']['items'] + + +def test_api_lists_new_endpoints(client: Client) -> None: + response = client.get('/api') + assert response.status_code == 200 + + data = response.json + queries = data['collection']['queries'] + endpoint_names = [q['data'][0]['value'] for q in queries] + + assert 'forms' in endpoint_names + assert 'resources' in endpoint_names + assert 'people' in endpoint_names + assert 'meetings' in endpoint_names + assert 'political_businesses' in endpoint_names + assert 'parliamentarians' in endpoint_names + assert 'commissions' in endpoint_names + assert 'parliamentary_groups' in endpoint_names + + +def test_api_people_endpoint(client: Client, town_app: TestTownApp) -> None: + session = town_app.session() + session.add(Person( + first_name='Hans', + last_name='Muster', + function='Mayor', + email='hans@example.org', + phone='0791234567', + phone_direct='0791234568', + organisation='Town Hall', + academic_title='Dr.', + profession='Politician', + salutation='Herr', + political_party='FDP', + )) + transaction.commit() + + response = client.get('/api/people') + assert response.status_code == 200 + data = response.json + items = data['collection']['items'] + assert len(items) == 1 + + item_data = api_item_data(items[0]) + assert item_data['first_name'] == 'Hans' + assert item_data['last_name'] == 'Muster' + assert item_data['function'] == 'Mayor' + assert item_data['email'] == 'hans@example.org' + assert item_data['phone'] == '0791234567' + assert item_data['phone_direct'] == '0791234568' + assert item_data['organisation'] == 'Town Hall' + assert item_data['academic_title'] == 'Dr.' + assert item_data['profession'] == 'Politician' + assert item_data['salutation'] == 'Herr' + assert item_data['political_party'] == 'FDP' + assert 'modified' in item_data + + +def test_api_meetings_endpoint( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + now = utcnow() + session.add(Meeting( + title='Council Meeting', + start_datetime=now, + address='Town Hall, Main Street 1', + )) + transaction.commit() + + response = client.get('/api/meetings') + assert response.status_code == 200 + data = response.json + items = data['collection']['items'] + assert len(items) == 1 + + item_data = api_item_data(items[0]) + assert item_data['title'] == 'Council Meeting' + assert item_data['start_datetime'] is not None + assert 'Town Hall' in (item_data['address'] or '') + + +def test_api_political_businesses_endpoint( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + session.add(PoliticalBusiness( + title='Motion about parks', + number='2024.001', + political_business_type='motion', + status='pendent_legislative', + )) + transaction.commit() + + response = client.get('/api/political_businesses') + assert response.status_code == 200 + data = response.json + items = data['collection']['items'] + assert len(items) == 1 + + item_data = api_item_data(items[0]) + assert item_data['title'] == 'Motion about parks' + assert item_data['number'] == '2024.001' + assert item_data['political_business_type'] == 'motion' + assert item_data['status'] == 'pendent_legislative' + assert item_data['display_name'] == '2024.001 Motion about parks' + + +def test_api_parliamentarians_endpoint( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + session.add(RISParliamentarian( + first_name='Anna', + last_name='Mueller', + party='SP', + )) + transaction.commit() + + response = client.get('/api/parliamentarians') + assert response.status_code == 200 + data = response.json + items = data['collection']['items'] + assert len(items) == 1 + + item_data = api_item_data(items[0]) + assert item_data['first_name'] == 'Anna' + assert item_data['last_name'] == 'Mueller' + assert item_data['party'] == 'SP' + + +def test_api_commissions_endpoint( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + session.add(RISCommission(name='Finance Commission')) + transaction.commit() + + response = client.get('/api/commissions') + assert response.status_code == 200 + data = response.json + items = data['collection']['items'] + assert len(items) == 1 + + item_data = api_item_data(items[0]) + assert item_data['name'] == 'Finance Commission' + + +def test_api_parliamentary_groups_endpoint( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + session.add(RISParliamentaryGroup(name='Green Party Group')) + transaction.commit() + + response = client.get('/api/parliamentary_groups') + assert response.status_code == 200 + data = response.json + items = data['collection']['items'] + assert len(items) == 1 + + item_data = {d['name']: d['value'] for d in items[0]['data']} + assert item_data['name'] == 'Green Party Group' + + +def test_api_forms_endpoint( + client: Client, town_app: TestTownApp +) -> None: + response = client.get('/api/forms') + assert response.status_code == 200 + data = response.json + items = data['collection']['items'] + assert len(items) >= 1 + + item_data = api_item_data(items[0]) + assert item_data['type'] == 'internal' + assert 'title' in item_data + assert 'text' in item_data + + +def test_api_forms_with_external_links( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + session.add(ExternalFormLink( + title='External Survey', + url='https://example.org/survey', + group='Surveys', + )) + transaction.commit() + + response = client.get('/api/forms') + assert response.status_code == 200 + data = response.json + items = data['collection']['items'] + + external_items = [ + item for item in items + if any(d['name'] == 'type' and d['value'] == 'external' + for d in item['data']) + ] + assert len(external_items) == 1 + + ext_data = api_item_data(external_items[0]) + assert ext_data['title'] == 'External Survey' + assert ext_data['url'] == 'https://example.org/survey' + assert ext_data['group'] == 'Surveys' + + +def test_api_resources_endpoint( + client: Client, town_app: TestTownApp +) -> None: + response = client.get('/api/resources') + assert response.status_code == 200 + data = response.json + assert 'items' in data['collection'] + + +def test_api_resources_with_external_links( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + session.add(ExternalResourceLink( + title='External Room', + url='https://example.org/room', + group='Rooms', + )) + transaction.commit() + + response = client.get('/api/resources') + assert response.status_code == 200 + data = response.json + items = data['collection']['items'] + + external_items = [ + item for item in items + if any(d['name'] == 'kind' and d['value'] == 'external' + for d in item['data']) + ] + assert len(external_items) == 1 + + ext_data = api_item_data(external_items[0]) + assert ext_data['title'] == 'External Room' + assert ext_data['url'] == 'https://example.org/room' + + +def test_api_forms_paginates_external_links_and_hides_invisible_items( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + forms = FormCollection(session) + + for ix in range(25): + form = forms.definitions.add( + f'API Form {ix:02d}', + 'E-Mail *= @@@', + type='custom', + ) + form.access = 'public' # type:ignore[attr-defined] + + hidden_form = forms.definitions.add( + 'API Form Hidden', + 'E-Mail *= @@@', + type='custom', + ) + hidden_form.access = 'private' # type:ignore[attr-defined] + + session.add(ExternalFormLink( + title='API External Form', + url='https://example.org/forms/public', + group='API', + )) + hidden_external = ExternalFormLink( + title='API External Hidden', + url='https://example.org/forms/private', + group='API', + ) + hidden_external.access = 'private' + session.add(hidden_external) + transaction.commit() + + pages = [api_items(client, '/api/forms', page) for page in range(5)] + relevant = [ + api_item_data(item)['title'] + for items in pages + for item in items + if str(api_item_data(item)['title']).startswith('API ') + ] + + assert all(len(items) <= 25 for items in pages) + assert relevant.count('API External Form') == 1 + assert 'API External Hidden' not in relevant + assert 'API Form Hidden' not in relevant + assert len(set(relevant)) == 26 + + client.get(f'/api/forms/{hidden_form.id}', status=404) + client.get(f'/api/forms/{hidden_external.id.hex}', status=404) + + +def test_api_resources_paginates_external_links_and_hides_invisible_items( + client: Client, town_app: TestTownApp +) -> None: + resources = ResourceCollection(town_app.libres_context) + + for ix in range(25): + resources.add( + title=f'API Resource {ix:02d}', + timezone='Europe/Zurich', + type='room', + meta={'access': 'public'}, + ) + + hidden_resource = resources.add( + title='API Resource Hidden', + timezone='Europe/Zurich', + type='room', + meta={'access': 'private'}, + ) + + session = town_app.session() + session.add(ExternalResourceLink( + title='API External Resource', + url='https://example.org/resources/public', + group='API', + )) + hidden_external = ExternalResourceLink( + title='API External Resource Hidden', + url='https://example.org/resources/private', + group='API', + ) + hidden_external.access = 'private' + session.add(hidden_external) + transaction.commit() + + pages = [api_items(client, '/api/resources', page) for page in range(5)] + relevant = [ + api_item_data(item)['title'] + for items in pages + for item in items + if str(api_item_data(item)['title']).startswith('API ') + ] + + assert all(len(items) <= 25 for items in pages) + assert relevant.count('API External Resource') == 1 + assert 'API External Resource Hidden' not in relevant + assert 'API Resource Hidden' not in relevant + assert len(set(relevant)) == 26 + + client.get(f'/api/resources/{hidden_resource.id.hex}', status=404) + client.get(f'/api/resources/{hidden_external.id.hex}', status=404) + + +def test_api_people_endpoint_hides_unpublished_entries( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + session.add(Person( + first_name='Public', + last_name='Person', + )) + hidden_person = Person( + first_name='Hidden', + last_name='Person', + publication_start=utcnow() + timedelta(days=1), + ) + session.add(hidden_person) + transaction.commit() + + items = api_items(client, '/api/people') + titles = {api_item_data(item)['title'] for item in items} + + assert 'Person Public' in titles + assert 'Person Hidden' not in titles + + client.get(f'/api/people/{hidden_person.id.hex}', status=404) + + +def test_api_meetings_endpoint_hides_private_entries( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + public_meeting = Meeting( + title='Public Meeting', + start_datetime=utcnow(), + address='Town Hall', + ) + hidden_meeting = Meeting( + title='Hidden Meeting', + start_datetime=utcnow(), + address='Secret Hall', + ) + hidden_meeting.access = 'private' + session.add(public_meeting) + session.add(hidden_meeting) + transaction.commit() + + items = api_items(client, '/api/meetings') + titles = {api_item_data(item)['title'] for item in items} + + assert 'Public Meeting' in titles + assert 'Hidden Meeting' not in titles + + client.get(f'/api/meetings/{hidden_meeting.id.hex}', status=404) + + +def test_api_political_businesses_endpoint_hides_private_entries( + client: Client, town_app: TestTownApp +) -> None: + session = town_app.session() + session.add(PoliticalBusiness( + title='Public Business', + number='2024.010', + political_business_type='motion', + status='pendent_legislative', + )) + hidden_business = PoliticalBusiness( + title='Hidden Business', + number='2024.011', + political_business_type='motion', + status='pendent_legislative', + ) + hidden_business.access = 'private' + session.add(hidden_business) + transaction.commit() + + items = api_items(client, '/api/political_businesses') + titles = {api_item_data(item)['title'] for item in items} + + assert 'Public Business' in titles + assert 'Hidden Business' not in titles + + client.get( + f'/api/political_businesses/{hidden_business.id.hex}', + status=404, + )