def _duplicate_versions(self, old_id, new_doc):
        """
        Duplicates the version history of the article identified by old_id. Each version identifiers are changed
        to have the identifiers of new_doc.

        :param old_id: identifier to fetch version history
        :param new_doc: identifiers from this doc will be used to create version history for the duplicated item.
        """
        resource_def = app.config['DOMAIN']['archive']
        version_id = versioned_id_field(resource_def)
        old_versions = get_resource_service('archive_versions').get(req=None, lookup={'guid': old_id})

        new_versions = []
        for old_version in old_versions:
            old_version[version_id] = new_doc[config.ID_FIELD]
            del old_version[config.ID_FIELD]

            old_version['guid'] = new_doc['guid']
            old_version['unique_name'] = new_doc['unique_name']
            old_version['unique_id'] = new_doc['unique_id']
            old_version['versioncreated'] = utcnow()
            if old_version[VERSION] == new_doc[VERSION]:
                old_version[ITEM_OPERATION] = new_doc[ITEM_OPERATION]
            new_versions.append(old_version)
        last_version = deepcopy(new_doc)
        last_version['_id_document'] = new_doc['_id']
        del last_version['_id']
        new_versions.append(last_version)
        if new_versions:
            get_resource_service('archive_versions').post(new_versions)

    def test_expiring_with_content(self):
        provider_name = 'reuters'
        guid = 'tag_reuters.com_2014_newsml_KBN0FL0NM:10'
        provider = get_resource_service('ingest_providers').find_one(name=provider_name, req=None)
        provider_service = self._get_provider_service(provider)
        provider_service.provider = provider
        provider_service.URL = provider.get('config', {}).get('url')
        items = provider_service.fetch_ingest(guid)
        now = utcnow()
        for i, item in enumerate(items):
            item['ingest_provider'] = provider['_id']
            expiry_time = now - timedelta(hours=11)
            if i > 4:
                expiry_time = now + timedelta(minutes=11)

            item['expiry'] = item['versioncreated'] = expiry_time

        service = get_resource_service('ingest')
        service.post(items)

        # ingest the items and expire them
        before = service.get(req=None, lookup={})
        self.assertEqual(6, before.count())

        remove = RemoveExpiredContent()
        remove.run(provider.get('type'))

        # only one left in ingest
        after = service.get(req=None, lookup={})
        self.assertEqual(1, after.count())

    def test_products(self, article):
        req = ParsedRequest()
        results = []
        products = list(get_resource_service('products').get(req=req, lookup=None))
        for product in products:
            result = {}
            result['product_id'] = product['_id']
            result['matched'] = True

            reason = ''
            if not EnqueueService().conforms_product_targets(product, article):
                # Here it fails to match due to geo restriction
                # story has target_region and product has geo restriction
                result['matched'] = False

                if BasePublishService().is_targeted(article, 'target_regions'):
                    reason = 'Story has target_region'

                if product.get('geo_restrictions'):
                    reason = '{} {}'.format(reason, 'Product has target_region')

            if not EnqueueService().conforms_content_filter(product, article):
                # Here it fails to match due to content filter
                content_filter = product.get('content_filter')
                filter = get_resource_service('content_filters').find_one(req=None, _id=content_filter['filter_id'])
                result['matched'] = False
                reason = 'Story does not match the filter: {}'.format(filter.get('name'))

            result['reason'] = reason
            results.append(result)
        return results

 def enhance_document_with_user_privileges(self, session_doc, user_doc):
     role_doc = get_resource_service('users').get_role(user_doc)
     get_resource_service('users').set_privileges(user_doc, role_doc)
     session_doc[_privileges_key] = user_doc.get(_privileges_key, {})
     # set last_updated to max for session/user/role so that client will fetch changes
     # after a change to any of those
     session_doc[app.config['LAST_UPDATED']] = last_updated(session_doc, user_doc, role_doc)

 def run(self, republish):
     # update themes
     theme_service = get_resource_service('themes')
     created, updated = theme_service.update_registered_theme_with_local_files()
     print('\n* %d themes updated from local files\n' % (len(created) + len(updated)))
     # retrieves all opened blogs
     blogs_service = get_resource_service('blogs')
     blogs = blogs_service.get(req=None, lookup=dict(blog_status='open'))
     print('* Update the theme for every blog\n')
     for blog in blogs:
         theme = blogs_service.get_theme_snapshot(blog['blog_preferences']['theme'])
         try:
             blogs_service.system_update(ObjectId(blog['_id']), {'theme': theme}, blog)
         except eve.io.base.DataLayer.OriginalChangedError:
             print(u'! an error occured during saving blog "%s".' % (blog['title']),
                   'Can be a broken relationship (with user for instance)')
         else:
             print('- Blog "%s"\'s theme was updated to %s %s' % (
                 blog['title'], theme['name'], theme['version']))
     # republish on s3
     if republish:
         print('\n* Republishing blogs:\n')
         for blog in blogs:
             url = publish_blog_embed_on_s3(blog_id=str(blog['_id']), safe=False)
             print('  - Blog "%s" republished: %s' % (blog['title'], url))

    def update(self, id, updates, original):
        original_state = original[config.CONTENT_STATE]
        if not is_workflow_state_transition_valid("spike", original_state):
            raise InvalidStateTransitionError()

        package_service = PackageService()
        user = get_user(required=True)

        item = get_resource_service(ARCHIVE).find_one(req=None, _id=id)
        expiry_minutes = app.settings["SPIKE_EXPIRY_MINUTES"]

        # check if item is in a desk. If it's then use the desks spike_expiry
        if is_assigned_to_a_desk(item):
            desk = get_resource_service("desks").find_one(_id=item["task"]["desk"], req=None)
            expiry_minutes = desk.get("spike_expiry", expiry_minutes)

        updates[EXPIRY] = get_expiry_date(expiry_minutes)
        updates[REVERT_STATE] = item.get(app.config["CONTENT_STATE"], None)

        if original.get("rewrite_of"):
            updates["rewrite_of"] = None

        item = self.backend.update(self.datasource, id, updates, original)
        push_notification("item:spike", item=str(item.get("_id")), user=str(user))
        package_service.remove_spiked_refs_from_package(id)
        return item

def transmit_items(queue_items, subscriber, destination, output_channels):
    failed_items = []

    for queue_item in queue_items:
        # Check if output channel is active
        if not (output_channels.get(str(queue_item['output_channel_id']), {})).get('is_active', False):
            continue

        try:
            if not is_on_time(queue_item, destination):
                continue

            # update the status of the item to in-progress
            queue_update = {'state': 'in-progress', 'transmit_started_at': utcnow()}
            superdesk.get_resource_service('publish_queue').patch(queue_item.get('_id'), queue_update)

            # get the formatted item
            formatted_item = superdesk.get_resource_service('formatted_item').\
                find_one(req=None, _id=queue_item['formatted_item_id'])

            transmitter = superdesk.publish.transmitters[destination.get('delivery_type')]
            transmitter.transmit(queue_item, formatted_item, subscriber, destination)
            update_content_state(queue_item)
        except:
            failed_items.append(queue_item)

    if len(failed_items) > 0:
        logger.error('Failed to publish the following items: %s', str(failed_items))

def get_expiry(desk_id=None, stage_id=None, desk_or_stage_doc=None):
    """
    Calculates the expiry for a content from fetching the expiry duration from one of the below
        1. desk identified by desk_id
        2. stage identified by stage_id. This will ignore desk_id if specified
        3. desk doc or stage doc identified by doc_or_stage_doc. This will ignore desk_id and stage_id if specified

    :param desk_id: desk identifier
    :param stage_id: stage identifier
    :param desk_or_stage_doc: doc from either desks collection or stages collection
    :return: when the doc will expire
    """

    stage = None

    if desk_or_stage_doc is None and desk_id:
        desk = superdesk.get_resource_service('desks').find_one(req=None, _id=desk_id)

        if not desk:
            raise SuperdeskApiError.notFoundError('Invalid desk identifier %s' % desk_id)

        if not stage_id:
            stage = get_resource_service('stages').find_one(req=None, _id=desk['incoming_stage'])

            if not stage:
                raise SuperdeskApiError.notFoundError('Invalid stage identifier %s' % stage_id)

    if desk_or_stage_doc is None and stage_id:
        stage = get_resource_service('stages').find_one(req=None, _id=stage_id)

        if not stage:
                raise SuperdeskApiError.notFoundError('Invalid stage identifier %s' % stage_id)

    return get_item_expiry(app=app, stage=desk_or_stage_doc or stage)

    def data_scaffolding_test(self):
        with self.app.app_context():
            command = AppInitializeWithDataCommand()
            result = command.run()
            self.assertEquals(result, 0)

            service = get_resource_service('text_archive')
            docs = [{
                'type': 'text',
                'abstract': 'test abstract {}'.format(x),
                'headline': 'test headline {}'.format(x),
                'body_html': 'test long story body {}'.format(x)
            } for x in range(0, 40)]
            service.post(docs)

            stories_per_desk = 2
            existing_desks = 1
            command = AppScaffoldDataCommand()
            result = command.run(stories_per_desk)
            self.assertEquals(result, 0)

            cursor = get_resource_service('desks').get_from_mongo(None, {})
            self.assertEquals(cursor.count(), existing_desks)

            cursor = get_resource_service('archive').get_from_mongo(None, {})
            self.assertEquals(cursor.count(), existing_desks * stories_per_desk)

    def _validate_disable(self, updates, original):
        """
        Checks the templates and desks that are referencing the given
        content profile if the profile is being disabled
        """
        if 'enabled' in updates and updates.get('enabled') is False and original.get('enabled') is True:
            templates = list(superdesk.get_resource_service('content_templates').
                             get_templates_by_profile_id(original.get('_id')))

            if len(templates) > 0:
                template_names = ', '.join([t.get('template_name') for t in templates])
                raise SuperdeskApiError.badRequestError(
                    message='Cannot disable content profile as following templates are referencing: {}'.
                    format(template_names))

            req = ParsedRequest()
            all_desks = list(superdesk.get_resource_service('desks').get(req=req, lookup={}))
            profile_desks = [desk for desk in all_desks if
                             desk.get('default_content_profile') == str(original.get('_id'))]

            if len(profile_desks) > 0:
                profile_desk_names = ', '.join([d.get('name') for d in profile_desks])
                raise SuperdeskApiError.badRequestError(
                    message='Cannot disable content profile as following desks are referencing: {}'.
                    format(profile_desk_names))

    def queue_transmission(self, doc, subscribers):
        """
        Method formats and then queues the article for transmission to the passed subscribers.
        ::Important Note:: Format Type across Subscribers can repeat. But we can't have formatted item generated once
        based on the format_types configured across for all the subscribers as the formatted item must have a published
        sequence number generated by Subscriber.
        :param dict doc: document to queue for transmission
        :param list subscribers: List of subscriber dict.
        :return : (list, bool) tuple of list of missing formatters and boolean flag. True if queued else False
        """

        try:
            queued = False
            no_formatters = []
            for subscriber in subscribers:
                try:
                    if doc[ITEM_TYPE] not in [CONTENT_TYPE.TEXT, CONTENT_TYPE.PREFORMATTED] and \
                            subscriber.get('subscriber_type', '') == SUBSCRIBER_TYPES.WIRE:
                        # wire subscribers can get only text and preformatted stories
                        continue

                    for destination in subscriber['destinations']:
                        # Step 2(a)
                        formatter = get_formatter(destination['format'], doc)

                        if not formatter:  # if formatter not found then record it
                            no_formatters.append(destination['format'])
                            continue

                        formatted_docs = formatter.format(doc, subscriber)

                        for pub_seq_num, formatted_doc in formatted_docs:
                            publish_queue_item = dict()
                            publish_queue_item['item_id'] = doc['_id']
                            publish_queue_item['item_version'] = doc[config.VERSION]
                            publish_queue_item['formatted_item'] = formatted_doc
                            publish_queue_item['subscriber_id'] = subscriber['_id']
                            publish_queue_item['destination'] = destination
                            publish_queue_item['published_seq_num'] = pub_seq_num
                            publish_queue_item['publish_schedule'] = doc.get('publish_schedule', None)
                            publish_queue_item['unique_name'] = doc.get('unique_name', None)
                            publish_queue_item['content_type'] = doc.get('type', None)
                            publish_queue_item['headline'] = doc.get('headline', None)

                            self.set_state(doc, publish_queue_item)
                            if publish_queue_item.get(ITEM_STATE):
                                publish_queue_item['publishing_action'] = publish_queue_item.get(ITEM_STATE)
                                del publish_queue_item[ITEM_STATE]
                            else:
                                publish_queue_item['publishing_action'] = self.published_state

                            get_resource_service('publish_queue').post([publish_queue_item])
                            queued = True
                except:
                    logger.exception("Failed to queue item for id {} with headline {} for subscriber {}."
                                     .format(doc.get(config.ID_FIELD), doc.get('headline'), subscriber.get('name')))

            return no_formatters, queued
        except:
            raise

    def on_delete(self, doc):
        """
        Checks if deleting the stage would not violate data integrity, raises an exception if it does.

            1/ Can't delete the default incoming stage
            2/ The stage must have no documents (spiked or unspiked)
            3/ The stage can not be referred to by a ingest routing rule

        :param doc:
        :return:
        """
        if doc['default_incoming'] is True:
            desk_id = doc.get('desk', None)
            if desk_id and superdesk.get_resource_service('desks').find_one(req=None, _id=desk_id):
                raise SuperdeskApiError.preconditionFailedError(message='Cannot delete a default stage.')

        archive_versions_query = {'task.stage': str(doc[config.ID_FIELD])}
        items = superdesk.get_resource_service('archive_versions').get(req=None, lookup=archive_versions_query)
        if items and items.count():
            raise SuperdeskApiError.preconditionFailedError(
                message='Cannot delete stage as it has article(s) or referenced by versions of the article(s).')

        # check if the stage is referred to in a ingest routing rule
        rules = self._stage_in_rule(doc[config.ID_FIELD])
        if rules.count() > 0:
            rule_names = ', '.join(rule.get('name') for rule in rules)
            raise SuperdeskApiError.preconditionFailedError(
                message='Stage is referred by Ingest Routing Schemes : {}'.format(rule_names))

    def test_subject_to_anpa_category_derived_ingest_ignores_inactive_map_entries(self):
        vocab = [{'_id': 'iptc_category_map',
                  'items': [{'name': 'Finance', 'category': 'f', 'subject': '04000000', 'is_active': False}]},
                 {'_id': 'categories',
                  'items': [{'is_active': True, 'name': 'Australian Weather', 'qcode': 'b', 'subject': '17000000'}]}]

        self.app.data.insert('vocabularies', vocab)

        provider_name = 'AAP'
        provider = get_resource_service('ingest_providers').find_one(name=provider_name, req=None)
        file_path = os.path.join(provider.get('config', {}).get('path', ''), 'nitf-fishing.xml')
        provider_service = self._get_provider_service(provider)
        feeding_parser = provider_service.get_feed_parser(provider)
        with open(file_path, 'r') as f:
            xml_string = etree.etree.fromstring(f.read())
            items = [feeding_parser.parse(xml_string, provider)]
            for item in items:
                item['ingest_provider'] = provider['_id']
                item['expiry'] = utcnow() + timedelta(hours=11)

            service = get_resource_service('ingest')
            service.post(items)

            # ingest the items and check the subject code has been derived
            self.ingest_items(items, provider)
            self.assertNotIn('anpa_category', items[0])

    def on_deleted(self, doc):
        """
        Overriding clean up reset password tokens:
        """

        super().on_deleted(doc)
        get_resource_service('reset_user_password').remove_all_tokens_for_email(doc.get('email'))

    def lock(self, item_filter, user_id, session_id, etag):
        item_model = get_model(ItemModel)
        item = item_model.find_one(item_filter)

        if not item:
            raise SuperdeskApiError.notFoundError()

        can_user_lock, error_message = self.can_lock(item, user_id, session_id)

        if can_user_lock:
            self.app.on_item_lock(item, user_id)
            updates = {LOCK_USER: user_id, LOCK_SESSION: session_id, 'lock_time': utcnow()}
            item_model.update(item_filter, updates)

            if item.get(TASK):
                item[TASK]['user'] = user_id
            else:
                item[TASK] = {'user': user_id}

            superdesk.get_resource_service('tasks').assign_user(item[config.ID_FIELD], item[TASK])
            self.app.on_item_locked(item, user_id)
            push_notification('item:lock',
                              item=str(item.get(config.ID_FIELD)),
                              item_version=str(item.get(config.VERSION)),
                              user=str(user_id), lock_time=updates['lock_time'],
                              lock_session=str(session_id))
        else:
            raise SuperdeskApiError.forbiddenError(message=error_message)

        item = item_model.find_one(item_filter)
        return item

def get_expiry(desk_id, stage_id, offset=None):
    """
    Calculates the expiry for a content from fetching the expiry duration from one of the below
        1. desk identified by desk_id
        2. stage identified by stage_id
    :param desk_id: desk identifier
    :param stage_id: stage identifier
    :return: when the doc will expire
    """
    stage = None
    desk = None

    if desk_id:
        desk = superdesk.get_resource_service('desks').find_one(req=None, _id=desk_id)

        if not desk:
            raise SuperdeskApiError.notFoundError('Invalid desk identifier %s' % desk_id)

    if stage_id:
        stage = get_resource_service('stages').find_one(req=None, _id=stage_id)

        if not stage:
                raise SuperdeskApiError.notFoundError('Invalid stage identifier %s' % stage_id)

    return get_item_expiry(desk, stage, offset)

    def unlock(self, item_filter, user_id, session_id, etag):
        item_model = get_model(ItemModel)
        item = item_model.find_one(item_filter)

        if not item:
            raise SuperdeskApiError.notFoundError()

        if not item.get(LOCK_USER):
            raise SuperdeskApiError.badRequestError(message="Item is not locked.")

        can_user_unlock, error_message = self.can_unlock(item, user_id)

        if can_user_unlock:
            self.app.on_item_unlock(item, user_id)

            # delete the item if nothing is saved so far
            # version 0 created on lock item
            if item.get(config.VERSION, 0) == 0 and item[ITEM_STATE] == CONTENT_STATE.DRAFT:
                superdesk.get_resource_service('archive').delete_action(lookup={'_id': item['_id']})
                push_content_notification([item])
            else:
                updates = {LOCK_USER: None, LOCK_SESSION: None, 'lock_time': None, 'force_unlock': True}
                item_model.update(item_filter, updates)
                self.app.on_item_unlocked(item, user_id)

            push_notification('item:unlock',
                              item=str(item_filter.get(config.ID_FIELD)),
                              item_version=str(item.get(config.VERSION)),
                              state=item.get(ITEM_STATE),
                              user=str(user_id), lock_session=str(session_id))
        else:
            raise SuperdeskApiError.forbiddenError(message=error_message)

        item = item_model.find_one(item_filter)
        return item

    def test_expiring_content_with_files(self):
        provider_name = 'reuters'
        guid = 'tag_reuters.com_2014_newsml_KBN0FL0NM'
        provider = get_resource_service('ingest_providers').find_one(name=provider_name, req=None)
        provider_service = self.provider_services[provider.get('type')]
        provider_service.provider = provider

        items = provider_service.fetch_ingest(guid)
        for item in items:
            item['ingest_provider'] = provider['_id']

        now = utcnow()
        items[0]['expiry'] = now - timedelta(hours=11)
        items[1]['expiry'] = now - timedelta(hours=11)
        items[2]['expiry'] = now + timedelta(hours=11)
        items[5]['versioncreated'] = now + timedelta(minutes=11)

        service = get_resource_service('ingest')
        service.post(items)

        # ingest the items and expire them
        self.ingest_items(items, provider)

        # four files in grid fs
        current_files = self.app.media.fs('upload').find()
        self.assertEqual(4, current_files.count())

        remove = RemoveExpiredContent()
        remove.run(provider.get('type'))

        # all gone
        current_files = self.app.media.fs('upload').find()
        self.assertEqual(0, current_files.count())

 def find_one(self, req, **lookup):
     session = get_resource_service('sessions').find_one(req=None, _id=lookup['_id'])
     _id = session['user'] if session else lookup['_id']
     doc = get_resource_service('users').find_one(req, _id=_id)
     if doc:
         doc['_id'] = session['_id'] if session else _id
     return doc

    def test_files_dont_duplicate_ingest(self):
        provider_name = 'reuters'
        guid = 'tag_reuters.com_2014_newsml_KBN0FL0NM'
        provider = get_resource_service('ingest_providers').find_one(name=provider_name, req=None)
        provider_service = self.provider_services[provider.get('type')]
        provider_service.provider = provider

        items = provider_service.fetch_ingest(guid)
        for item in items:
            item['ingest_provider'] = provider['_id']
            item['expiry'] = utcnow() + timedelta(hours=11)

        service = get_resource_service('ingest')
        service.post(items)

        # ingest the items
        self.ingest_items(items, provider)

        items = provider_service.fetch_ingest(guid)
        for item in items:
            item['ingest_provider'] = provider['_id']
            item['expiry'] = utcnow() + timedelta(hours=11)

        # ingest them again
        self.ingest_items(items, provider)

        # 12 files in grid fs
        current_files = self.app.media.fs('upload').find()
        self.assertEqual(12, current_files.count())