def skip_last_updated(model_cls):
    skip_last_updated = getattr(session(model_cls)._get(), 'skip_last_updated', False)
    session(model_cls)._get().skip_last_updated = True
    try:
        yield
    finally:
        session(model_cls)._get().skip_last_updated = skip_last_updated

 def test_send_no_configured_webhooks(self, send_webhook):
     self.wh.delete()
     session(self.wh).flush(self.wh)
     sender = RepoPushWebhookSender()
     with h.push_config(c, app=self.git):
         sender.send(dict(arg1=1, arg2=2))
     assert_equal(send_webhook.post.call_count, 0)

    def test_edit_validation(self):
        invalid = M.Webhook(
            type='invalid type',
            app_config_id=None,
            hook_url='http://httpbin.org/post',
            secret='secret')
        session(invalid).flush(invalid)
        self.app.get(self.url + '/repo-push/%s' % invalid._id, status=404)

        data = {'url': u'http://httpbin.org/post',
                'secret': u'secret'}
        self.create_webhook(data).follow()
        wh = M.Webhook.query.get(hook_url=data['url'], type='repo-push')

        # invalid id in hidden field, just in case
        r = self.app.get(self.url + '/repo-push/%s' % wh._id)
        data = {k: v[0].value for (k, v) in r.forms[0].fields.items()}
        data['webhook'] = unicode(invalid._id)
        self.app.post(self.url + '/repo-push/edit', data, status=404)

        # empty values
        data = {'url': '', 'secret': '', 'webhook': str(wh._id)}
        r = self.app.post(self.url + '/repo-push/edit', data)
        self.find_error(r, 'url', 'Please enter a value', 'edit')

        data = {'url': 'qwe', 'secret': 'qwe', 'webhook': str(wh._id)}
        r = self.app.post(self.url + '/repo-push/edit', data)
        self.find_error(r, 'url',
                        'You must provide a full domain name (like qwe.com)', 'edit')

 def create(self, **kwargs):
     try:
         entry = self.model(**request.json)
     except Invalid as exc:
         abort(400, exc.message)
     session(entry).flush()
     return entry

 def execute(cls, options):
     models = [M.Post, ForumPost]
     app_config_id = cls.get_tool_id(options.tool)
     # Find all posts that have parent_id, but does not have actual parent
     # and create fake parent for them
     for model in models:
         q = {'parent_id': {'$ne': None},
              'app_config_id': app_config_id}
         for chunk in chunked_find(model, q):
             for post in chunk:
                 if not post.parent:
                     log.info('Creating deleted parent for %s %s',
                              model.__mongometa__.name, post._id)
                     c.project = post.app_config.project
                     slug = post.slug.rsplit('/', 1)[0]
                     full_slug = post.full_slug.rsplit('/', 1)[0]
                     author = c.project.admins()[0]
                     deleted_post = model(
                         _id=post.parent_id,
                         deleted=True,
                         text="Automatically created in place of deleted post",
                         app_id=post.app_id,
                         app_config_id=post.app_config_id,
                         discussion_id=post.discussion_id,
                         thread_id=post.thread_id,
                         author_id=author._id,
                         slug=slug,
                         full_slug=full_slug,
                     )
                     if options.dry_run:
                         session(deleted_post).expunge(deleted_post)
                     else:
                         session(deleted_post).flush(deleted_post)

 def test_update_limit(self, dt_mock):
     _now = dt.datetime(2015, 02, 02, 13, 39)
     dt_mock.datetime.utcnow.return_value = _now
     assert_equal(self.wh.last_sent, None)
     self.wh.update_limit()
     session(self.wh).expunge(self.wh)
     assert_equal(M.Webhook.query.get(_id=self.wh._id).last_sent, _now)

 def test_project_is_deleted(self):
     p = M.Project.query.get(shortname=self.p_shortname)
     assert p is not None, 'Can not find project to delete'
     self.run_script(['p/{}'.format(p.shortname)])
     session(p).expunge(p)
     p = M.Project.query.get(shortname=p.shortname)
     assert p is None, 'Project is not deleted'

 def add_webhooks(suffix, n):
     for i in range(n):
         webhook = M.Webhook(
             type='repo-push',
             app_config_id=self.git.config._id,
             hook_url='http://httpbin.org/{}/{}'.format(suffix, i),
             secret='secret')
         session(webhook).flush(webhook)

 def verify_and_remove_code(self, user, code):
     self.enforce_rate_limit(user)
     rc = RecoveryCode.query.get(user_id=user._id, code=code)
     if rc:
         rc.query.delete()
         session(rc).flush(rc)
         return True
     else:
         raise InvalidRecoveryCode

 def test_subproject_is_deleted(self):
     p = M.Project.query.get(shortname='test/sub1')
     assert p is not None, 'Can not find subproject to delete'
     self.run_script(['p/test/sub1'])
     session(p).expunge(p)
     p = M.Project.query.get(shortname='test/sub1')
     assert p is None, 'Project is not deleted'
     p = M.Project.query.get(shortname='test')
     assert p is not None, 'Parent project should not be deleted'

def merge(merge_request_id):
    from allura import model as M
    mr = M.MergeRequest.query.get(_id=merge_request_id)
    mr.app.repo.merge(mr)
    mr.add_meta_post(changes={'Status': [mr.status, 'merged']})
    mr.status = 'merged'
    g.director.create_activity(c.user, 'merged', mr,
                               related_nodes=[c.project], tags=['merge-request'])
    session(mr).flush(mr)

 def test_delete_with_reason(self, log, post_event):
     p = M.Project.query.get(shortname=self.p_shortname)
     pid = p._id
     assert p is not None, 'Can not find project to delete'
     self.run_script(['-r', 'The Reason', 'p/{}'.format(p.shortname)])
     session(p).expunge(p)
     p = M.Project.query.get(shortname=p.shortname)
     assert p is None, 'Project is not deleted'
     log.info.assert_called_once_with('Purging %s Reason: %s', '/p/test-delete/', 'The Reason')
     post_event.assert_called_once_with('project_deleted', project_id=pid, reason='The Reason')

 def subscribe(self, **kw):
     threads = kw.pop('threads', [])
     for t in threads:
         thread = self.M.Thread.query.get(_id=t['_id'])
         if t.get('subscription'):
             thread.subscribe()
         else:
             thread.unsubscribe()
         session(self.M.Thread)._get().skip_mod_date = True
         session(self.M.Thread)._get().skip_last_updated = True
     redirect(request.referer)

 def installable_tools_for(project):
     tools = []
     for name, App in g.entry_points['tool'].iteritems():
         cfg = M.AppConfig(project_id=project._id, tool_name=name)
         app = App(project, cfg)
         if app.installable:
             tools.append(dict(name=name, app=App))
         session(cfg).expunge(cfg)  # prevent from saving temporary config to db
     tools.sort(key=lambda t: (t['app'].status_int(), t['app'].ordinal))
     return [t for t in tools
         if t['app'].status in project.allowed_tool_status]

 def test_deleted_post(self):
     r = self._make_post('This is a post')
     reply_form = r.html.find(
         'div', {'class': 'edit_post_form reply'}).find('form')
     post_link = str(reply_form['action']).rstrip('/')
     _, slug = post_link.rsplit('/', 1)
     r = self.app.get(post_link, status=200)
     post = M.Post.query.get(slug=slug)
     post.deleted = True
     session(post).flush(post)
     r = self.app.get(post_link, status=404)

 def update_webhook(self, wh, url, secret=None):
     if not secret:
         secret = self.gen_secret()
     wh.hook_url = url
     wh.secret = secret
     try:
         session(wh).flush(wh)
     except DuplicateKeyError:
         session(wh).expunge(wh)
         msg = u'_the_form: "{}" webhook already exists for {} {}'.format(
             wh.type, self.app.config.options.mount_label, url)
         raise Invalid(msg, None, None)

 def setUp(self):
     setup_basic_test()
     self.patches = self.monkey_patch()
     for p in self.patches:
         p.start()
     self.setup_with_tools()
     self.project = M.Project.query.get(shortname=test_project_with_repo)
     self.git = self.project.app_instance('src')
     self.wh = M.Webhook(
         type='repo-push',
         app_config_id=self.git.config._id,
         hook_url='http://httpbin.org/post',
         secret='secret')
     session(self.wh).flush(self.wh)

    def cached_convert(self, artifact, field_name):
        """Convert ``artifact.field_name`` markdown source to html, caching
        the result if the render time is greater than the defined threshold.

        """
        source_text = getattr(artifact, field_name)
        # Check if contents macro and never cache
        if "[[" in source_text:
            return self.convert(source_text)
        cache_field_name = field_name + '_cache'
        cache = getattr(artifact, cache_field_name, None)
        if not cache:
            log.warn(
                'Skipping Markdown caching - Missing cache field "%s" on class %s',
                field_name, artifact.__class__.__name__)
            return self.convert(source_text)

        bugfix_rev = 4  # increment this if we need all caches to invalidated (e.g. xss in markdown rendering fixed)
        md5 = None
        # If a cached version exists and it is valid, return it.
        if cache.md5 is not None:
            md5 = hashlib.md5(source_text.encode('utf-8')).hexdigest()
            if cache.md5 == md5 and getattr(cache, 'fix7528', False) == bugfix_rev:
                return h.html.literal(cache.html)

        # Convert the markdown and time the result.
        start = time.time()
        html = self.convert(source_text, render_limit=False)
        render_time = time.time() - start

        threshold = config.get('markdown_cache_threshold')
        try:
            threshold = float(threshold) if threshold else None
        except ValueError:
            threshold = None
            log.warn('Skipping Markdown caching - The value for config param '
                     '"markdown_cache_threshold" must be a float.')

        if threshold is not None and render_time > threshold:
            # Save the cache
            if md5 is None:
                md5 = hashlib.md5(source_text.encode('utf-8')).hexdigest()
            cache.md5, cache.html, cache.render_time = md5, html, render_time
            cache.fix7528 = bugfix_rev  # flag to indicate good caches created after [#7528] and other critical bugs were fixed.

            with utils.skip_mod_date(artifact.__class__), \
                 utils.skip_last_updated(artifact.__class__):
                session(artifact).flush(artifact)
        return html

    def test_delete_validation(self):
        invalid = M.Webhook(
            type='invalid type',
            app_config_id=None,
            hook_url='http://httpbin.org/post',
            secret='secret')
        session(invalid).flush(invalid)
        assert_equal(M.Webhook.query.find().count(), 1)

        data = {'webhook': ''}
        self.app.post(self.url + '/repo-push/delete', data, status=404)

        data = {'webhook': unicode(invalid._id)}
        self.app.post(self.url + '/repo-push/delete', data, status=404)
        assert_equal(M.Webhook.query.find().count(), 1)

 def __call__(self):
     '''Call the task function with its arguments.'''
     self.time_start = datetime.utcnow()
     session(self).flush(self)
     try:
         func = self.function
         self.result = func(*self.task.args, **self.task.kwargs)
         self.state = 'complete'
         return self.result
     except Exception:
         self.state = 'error'
         self.result = traceback.format_exc()
         raise
     finally:
         self.time_stop = datetime.utcnow()
         session(self).flush(self)