subjects of a path

test_subjects_under_ctx.py

# -*- coding: utf-8 -*-
"""Setup tests for this package."""
from bda.aaf.site.testing import BDA_AAF_SITE_FUNCTIONAL_TESTING
from plone import api
from plone.app.testing import setRoles
from plone.app.testing import TEST_USER_ID

import unittest


class TestLocalSubjects(unittest.TestCase):
    """Test that bda.aaf.site is properly installed."""

    layer = BDA_AAF_SITE_FUNCTIONAL_TESTING

    def setUp(self):
        """Custom shared utility setup for tests."""
        self.portal = self.layer['portal']
        self._make('Subsite', self.portal)
        self.subsite_1 = self._make('Subsite', self.portal, 1)
        self.subsite_2 = self._make('Subsite', self.portal, 2)
        self.channel_1 = self._make('Channel', self.subsite_1, 12)
        self.channel_2 = self._make('Channel', self.subsite_2, 21)

    def _make(self, ptype, container, count=None, **kwargs):
        setRoles(self.portal, TEST_USER_ID, ['Manager'])
        content = api.content.create(
            container,
            type=ptype,
            id='my{0}{1}'.format(
                ptype.lower().replace(' ', '_'),
                str(count) if count else '',
            ),
            title='My {0}'.format(ptype),
            **kwargs
        )
        content.reindexObject()
        return content

    def test_util_subjects(self):
        subjects_1_base = ['Berlin', 'Wien', 'Paris']
        subjects_2_base = ['Montreal', 'Washington', 'Brasilia']
        self.channel_1.subject = subjects_1_base
        self.channel_1.reindexObject()
        self.channel_2.subject = subjects_2_base
        self.channel_2.reindexObject()

        from bda.aaf.site.utils import subjects_under_context

        subjects_1 = subjects_under_context(self.channel_1)
        self.assertEqual(sorted(subjects_1_base), subjects_1)

        subjects_2 = subjects_under_context(self.channel_2)
        self.assertEqual(sorted(subjects_2_base), subjects_2)

utils.py

from BTrees.IIBTree import intersection
from plone import api


def subjects_under_context(context, indexname='Subject'):
    """valid subjects under the given context
    """
    pcat = api.portal.get_tool('portal_catalog')
    cat = pcat._catalog
    path_idx = cat.indexes['path']
    tags_idx = cat.indexes[indexname]
    result = []
    # query all oids of path - low level
    pquery = {
        'path': {
            'query': '/'.join(context.getPhysicalPath()),
            'depth': -1,
        }
    }
    path_result, info = path_idx._apply_index(pquery)
    for tag in tags_idx.uniqueValues():
        tquery = {indexname: tag}
        tags_result, info = tags_idx._apply_index(tquery)
        if intersection(path_result, tags_result):
            result.append(tag)
    return result