~launchpad-pqm/launchpad/devel

# Copyright 2009-2011 Canonical Ltd.  This software is licensed under the

# GNU Affero General Public License version 3 (see the file LICENSE).

"""Content classes for the 'home pages' of the subsystems of Launchpad."""

__all__ = [

    'BazaarApplication',

    'CodeImportSchedulerApplication',

    'FeedsApplication',

    'MailingListApplication',

    'MaloneApplication',

    'PrivateMaloneApplication',

    'RosettaApplication',

    'TestOpenIDApplication',

    ]

__metaclass__ = type

import codecs

import os

from lazr.restful import ServiceRootResource

from lazr.restful.interfaces import ITopLevelEntryLink

from zope.component import getUtility

from zope.interface import implements

from lp.services.config import config

from lp.services.webapp.interfaces import (

    IAPIDocRoot,

    ICanonicalUrlData,

    ILaunchBag,

    )

from lp.bugs.errors import InvalidBugTargetType

from lp.bugs.interfaces.bug import (

    CreateBugParams,

    IBugSet,

    )

from lp.bugs.interfaces.bugtask import (

    BugTaskSearchParams,

    IBugTaskSet,

    )

from lp.bugs.interfaces.bugtracker import IBugTrackerSet

from lp.bugs.interfaces.bugwatch import IBugWatchSet

from lp.bugs.interfaces.malone import (

    IMaloneApplication,

    IPrivateMaloneApplication,

    )

from lp.code.interfaces.codehosting import (

    IBazaarApplication,

    ICodehostingApplication,

    )

from lp.code.interfaces.codeimportscheduler import (

    ICodeImportSchedulerApplication,

    )

from lp.hardwaredb.interfaces.hwdb import (

    IHWDBApplication,

    IHWDeviceSet,

    IHWDriverSet,

    IHWSubmissionDeviceSet,

    IHWSubmissionSet,

    IHWVendorIDSet,

    ParameterError,

    )

from lp.registry.interfaces.distribution import IDistribution

from lp.registry.interfaces.distributionsourcepackage import (

    IDistributionSourcePackage,

    )

from lp.registry.interfaces.distroseries import IDistroSeriesSet

from lp.registry.interfaces.mailinglist import IMailingListApplication

from lp.registry.interfaces.product import (

    IProduct,

    IProductSet,

    )

from lp.services.feeds.interfaces import IFeedsApplication

from lp.services.statistics.interfaces.statistic import ILaunchpadStatisticSet

from lp.services.webservice.interfaces import IWebServiceApplication

from lp.services.worlddata.interfaces.language import ILanguageSet

from lp.testopenid.interfaces.server import ITestOpenIDApplication

from lp.translations.interfaces.translationgroup import ITranslationGroupSet

from lp.translations.interfaces.translations import IRosettaApplication

from lp.translations.interfaces.translationsoverview import (

    ITranslationsOverview,

    )

class CodehostingApplication:

    """Codehosting End-Point."""

    implements(ICodehostingApplication)

    title = "Codehosting API"

class CodeImportSchedulerApplication:

    """CodeImportScheduler End-Point."""

    implements(ICodeImportSchedulerApplication)

    title = "Code Import Scheduler"

class PrivateMaloneApplication:

    """ExternalBugTracker authentication token end-point."""

    implements(IPrivateMaloneApplication)

    title = "Launchpad Bugs."

class MailingListApplication:

    implements(IMailingListApplication)

class FeedsApplication:

    implements(IFeedsApplication)

class MaloneApplication:

    implements(IMaloneApplication)

    def __init__(self):

        self.title = 'Malone: the Launchpad bug tracker'

    def searchTasks(self, search_params, prejoins=[]):

        """See `IMaloneApplication`."""

        return getUtility(IBugTaskSet).search(

            search_params, prejoins=prejoins)

    def createBug(self, owner, title, description, target,

                  security_related=False, private=False, tags=None):

        """See `IMaloneApplication`."""

        params = CreateBugParams(

            title=title, comment=description, owner=owner,

            security_related=security_related, private=private, tags=tags)

        if IProduct.providedBy(target):

            params.setBugTarget(product=target)

        elif IDistribution.providedBy(target):

            params.setBugTarget(distribution=target)

        elif IDistributionSourcePackage.providedBy(target):

            params.setBugTarget(distribution=target.distribution,

                                sourcepackagename=target.sourcepackagename)

        else:

            raise InvalidBugTargetType(

                "A bug target must be a Project, a Distribution or a "

                "DistributionSourcePackage. Got %r." % target)

        return getUtility(IBugSet).createBug(params)

    @property

    def bug_count(self):

        user = getUtility(ILaunchBag).user

        return getUtility(IBugSet).searchAsUser(user=user).count()

    @property

    def bugwatch_count(self):

        return getUtility(IBugWatchSet).search().count()

    @property

    def bugtask_count(self):

        user = getUtility(ILaunchBag).user

        search_params = BugTaskSearchParams(user=user)

        return getUtility(IBugTaskSet).search(search_params).count()

    @property

    def bugtracker_count(self):

        return getUtility(IBugTrackerSet).count

    @property

    def projects_with_bugs_count(self):

        return getUtility(ILaunchpadStatisticSet).value('projects_with_bugs')

    @property

    def shared_bug_count(self):

        return getUtility(ILaunchpadStatisticSet).value('shared_bug_count')

    @property

    def top_bugtrackers(self):

        return getUtility(IBugTrackerSet).getMostActiveBugTrackers(limit=5)

    @property

    def latest_bugs(self):

        user = getUtility(ILaunchBag).user

        return getUtility(IBugSet).searchAsUser(

            user=user, orderBy=['-datecreated'], limit=5)

    def default_bug_list(self, user=None):

        return getUtility(IBugSet).searchAsUser(user)

class BazaarApplication:

    implements(IBazaarApplication)

    def __init__(self):

        self.title = 'The Open Source Bazaar'

class RosettaApplication:

    implements(IRosettaApplication)

    def __init__(self):

        self.title = 'Rosetta: Translations in the Launchpad'

        self.name = 'Rosetta'

    @property

    def languages(self):

        """See `IRosettaApplication`."""

        return getUtility(ILanguageSet)

    @property

    def language_count(self):

        """See `IRosettaApplication`."""

        stats = getUtility(ILaunchpadStatisticSet)

        return stats.value('language_count')

    @property

    def statsdate(self):

        stats = getUtility(ILaunchpadStatisticSet)

        return stats.dateupdated('potemplate_count')

    @property

    def translation_groups(self):

        """See `IRosettaApplication`."""

        return getUtility(ITranslationGroupSet)

    def translatable_products(self):

        """See `IRosettaApplication`."""

        products = getUtility(IProductSet)

        return products.getTranslatables()

    def featured_products(self):

        """See `IRosettaApplication`."""

        projects = getUtility(ITranslationsOverview)

        for project in projects.getMostTranslatedPillars():

            yield {

                'pillar': project['pillar'],

                'font_size': project['weight'] * 10,

                }

    def translatable_distroseriess(self):

        """See `IRosettaApplication`."""

        distroseriess = getUtility(IDistroSeriesSet)

        return distroseriess.translatables()

    def potemplate_count(self):

        """See `IRosettaApplication`."""

        stats = getUtility(ILaunchpadStatisticSet)

        return stats.value('potemplate_count')

    def pofile_count(self):

        """See `IRosettaApplication`."""

        stats = getUtility(ILaunchpadStatisticSet)

        return stats.value('pofile_count')

    def pomsgid_count(self):

        """See `IRosettaApplication`."""

        stats = getUtility(ILaunchpadStatisticSet)

        return stats.value('pomsgid_count')

    def translator_count(self):

        """See `IRosettaApplication`."""

        stats = getUtility(ILaunchpadStatisticSet)

        return stats.value('translator_count')

class HWDBApplication:

    """See `IHWDBApplication`."""

    implements(IHWDBApplication, ITopLevelEntryLink)

    link_name = 'hwdb'

    entry_type = IHWDBApplication

    def devices(self, bus, vendor_id, product_id=None):

        """See `IHWDBApplication`."""

        return getUtility(IHWDeviceSet).search(bus, vendor_id, product_id)

    def drivers(self, package_name=None, name=None):

        """See `IHWDBApplication`."""

        return getUtility(IHWDriverSet).search(package_name, name)

    def vendorIDs(self, bus):

        """See `IHWDBApplication`."""

        return getUtility(IHWVendorIDSet).idsForBus(bus)

    @property

    def driver_names(self):

        """See `IHWDBApplication`."""

        return getUtility(IHWDriverSet).all_driver_names()

    @property

    def package_names(self):

        """See `IHWDBApplication`."""

        return getUtility(IHWDriverSet).all_package_names()

    def search(self, user=None, device=None, driver=None, distribution=None,

               distroseries=None, architecture=None, owner=None,

               created_before=None, created_after=None,

               submitted_before=None, submitted_after=None):

        """See `IHWDBApplication`."""

        return getUtility(IHWSubmissionSet).search(

            user=user, device=device, driver=driver,

            distribution=distribution, distroseries=distroseries,

            architecture=architecture, owner=owner,

            created_before=created_before, created_after=created_after,

            submitted_before=submitted_before,

            submitted_after=submitted_after)

    def getDistroTarget(self, distribution, distroseries, distroarchseries):

        distro_targets = [

            target for target in (

                distribution, distroseries, distroarchseries)

            if target is not None]

        if len(distro_targets) == 0:

            return None

        elif len(distro_targets) == 1:

            return distro_targets[0]

        else:

            raise ParameterError(

                'Only one of `distribution`, `distroseries` or '

                '`distroarchseries` can be present.')

    def numSubmissionsWithDevice(

        self, bus=None, vendor_id=None, product_id=None, driver_name=None,

        package_name=None, distribution=None, distroseries=None,

        distroarchseries=None):

        """See `IHWDBApplication`."""

        submissions_with_device, all_submissions = (

            getUtility(IHWSubmissionSet).numSubmissionsWithDevice(

                bus, vendor_id, product_id, driver_name, package_name,

                distro_target=self.getDistroTarget(

                    distribution, distroseries, distroarchseries)))

        return {

            'submissions_with_device': submissions_with_device,

            'all_submissions': all_submissions,

            }

    def numOwnersOfDevice(

        self, bus=None, vendor_id=None, product_id=None, driver_name=None,

        package_name=None, distribution=None, distroseries=None,

        distroarchseries=None):

        """See `IHWDBApplication`."""

        owners, all_submitters = (

            getUtility(IHWSubmissionSet).numOwnersOfDevice(

                bus, vendor_id, product_id, driver_name, package_name,

                distro_target=self.getDistroTarget(

                    distribution, distroseries, distroarchseries)))

        return {

            'owners': owners,

            'all_submitters': all_submitters,

            }

    def numDevicesInSubmissions(

        self, bus=None, vendor_id=None, product_id=None, driver_name=None,

        package_name=None, distribution=None, distroseries=None,

        distroarchseries=None):

        """See `IHWDBApplication`."""

        return getUtility(IHWSubmissionDeviceSet).numDevicesInSubmissions(

                bus, vendor_id, product_id, driver_name, package_name,

                distro_target=self.getDistroTarget(

                    distribution, distroseries, distroarchseries))

    def deviceDriverOwnersAffectedByBugs(

        self, bus=None, vendor_id=None, product_id=None, driver_name=None,

        package_name=None, bug_ids=None, bug_tags=None, affected_by_bug=False,

        subscribed_to_bug=False, user=None):

        """See `IHWDBApplication`."""

        return getUtility(IHWSubmissionSet).deviceDriverOwnersAffectedByBugs(

            bus, vendor_id, product_id, driver_name, package_name, bug_ids,

            bug_tags, affected_by_bug, subscribed_to_bug, user)

    def hwInfoByBugRelatedUsers(

        self, bug_ids=None, bug_tags=None, affected_by_bug=False,

        subscribed_to_bug=False, user=None):

        """See `IHWDBApplication`."""

        return getUtility(IHWSubmissionSet).hwInfoByBugRelatedUsers(

            bug_ids, bug_tags, affected_by_bug, subscribed_to_bug, user)

class WebServiceApplication(ServiceRootResource):

    """See `IWebServiceApplication`.

    This implementation adds a 'cached_wadl' attribute, which starts

    out as an empty dict and is populated as needed.

    """

    implements(IWebServiceApplication, ICanonicalUrlData)

    inside = None

    path = ''

    rootsite = None

    cached_wadl = {}

    # This should only be used by devel instances: production serves root

    # WADL (and JSON) from the filesystem.

    @classmethod

    def cachedWADLPath(cls, instance_name, version):

        """Helper method to calculate the path to a cached WADL file."""

        return os.path.join(

            config.root, 'lib', 'canonical', 'launchpad',

            'apidoc', version, '%s.wadl' % (instance_name,))

    def toWADL(self):

        """See `IWebServiceApplication`.

        Look for a cached WADL file for the request version at the

        location used by the script

        utilities/create-launchpad-wadl.py. If the file is present,

        load the file and cache its contents rather than generating

        new WADL. Otherwise, generate new WADL and cache it.

        """

        version = self.request.version

        if self.__class__.cached_wadl is None:

            # The cache has been disabled for testing

            # purposes. Generate the WADL.

            return super(WebServiceApplication, self).toWADL()

        if  version not in self.__class__.cached_wadl:

            # It's not cached. Look for it on disk.

            _wadl_filename = self.cachedWADLPath(

                config.instance_name, version)

            _wadl_fd = None

            try:

                _wadl_fd = codecs.open(_wadl_filename, encoding='UTF-8')

                try:

                    wadl = _wadl_fd.read()

                finally:

                    _wadl_fd.close()

            except IOError:

                # It's not on disk; generate it.

                wadl = super(WebServiceApplication, self).toWADL()

            del _wadl_fd

            self.__class__.cached_wadl[version] = wadl

        return self.__class__.cached_wadl[version]

class TestOpenIDApplication:

    implements(ITestOpenIDApplication)

class APIDocRoot:

    implements(IAPIDocRoot)

    __parent__ = None

    __name__ = None

apidocroot = APIDocRoot()