def test_db_session_always_fresh(self):
     s = db.get_session()
     try:
         r = ReportStore()
         s.add(r)
         s.commit()
         r.id = None
         s.commit()
     except:
         pass
     
     # if the session is not cleaned up properly, this will throw an exception
     s = db.get_session()
     s.execute('select 1').fetchall()
     
     s = db.get_mw_session(mediawiki_project)
     try:
         u = MediawikiUser()
         s.add(u)
         s.commit()
         u.user_id = None
         s.commit()
     except:
         pass
     
     # if the session is not cleaned up properly, this will throw an exception
     s = db.get_mw_session(mediawiki_project)
     s.execute('select 1').fetchall()

 def get_session_and_leave_open(*args, **kwargs):
     from wikimetrics.configurables import db
     from wikimetrics.models import ReportStore, RunReport
     session = db.get_session()
     session2 = db.get_session()
     session2.query(ReportStore).first()
     session.query(ReportStore).first()

 def add(task_type, task_id, message, traceback):
     db_session = db.get_session()
     existing = TaskErrorStore.get(db_session, task_type, task_id)
     if existing:
         TaskErrorStore.update(db_session, existing, message, traceback)
     else:
         TaskErrorStore.create(db_session, task_type, task_id, message, traceback)

    def post_process(self, results):
        """
         If the report is public and this task went well,
         it will create a file on disk asynchronously.

         Results are of this form:

         Parameters:
            results : data to write to disk, in this form:
                {'5cab8d55-da19-436f-b675-1d2a3fca3481':
                    {'Sum': {'pages_created': Decimal('0.0000')}}
                }
        """

        if self.public is False:
            return

        try:
            session = db.get_session()
            db_report = session.query(ReportStore).get(self.persistent_id)
        finally:
            session.close()

        data = db_report.get_json_result(results)

        # code below schedules an async task on celery to write the file
        if self.recurrent_parent_id is not None:
            write_report_task.delay(self.recurrent_parent_id, self.created, data)
        else:
            # report is public and does not have a recurrent_parent_id, it's
            # the parent report, call the first run of the report
            self._run_child_report()

def get_usernames_for_task_result(task_result):
    """
    Parameters
        task_result : the result dictionary from Celery
    Returns
         user_names : dictionary of user names (keyed by WikiUserKey)
                      empty if results are not detailed by user

    TODO: this function should move outside the controller,
          at the time of writing the function we are
          consolidating code that wasduplicated
    """
    user_names = {}
    if Aggregation.IND in task_result:
        session = db.get_session()
        # cohort should be the same for all users
        # get cohort from first key
        cohort_id = None

        for wiki_user_key_str, row in task_result[Aggregation.IND].iteritems():
            wiki_user_key = WikiUserKey.fromstr(wiki_user_key_str)
            cohort_id = wiki_user_key.cohort_id
            break

        user_names = g.cohort_service.get_wikiusernames_for_cohort(cohort_id, session)

    return user_names

def cohort_detail(name_or_id):
    """
    Returns a JSON object of the form:
    {id: 2, name: 'Berlin Beekeeping Society', description: '', wikiusers: [
        {mediawiki_username: 'Andrea', mediawiki_userid: 5, project: 'dewiki'},
        {mediawiki_username: 'Dennis', mediawiki_userid: 6, project: 'dewiki'},
        {mediawiki_username: 'Florian', mediawiki_userid: 7, project: 'dewiki'},
        {mediawiki_username: 'Gabriele', mediawiki_userid: 8, project: 'dewiki'},
    ]}
    """
    cohort = None
    db_session = db.get_session()
    try:
        kargs = dict()
        if str(name_or_id).isdigit():
            kargs['by_id'] = int(name_or_id)
        else:
            kargs['by_name'] = name_or_id
        cohort = g.cohort_service.get_for_display(db_session, current_user.id, **kargs)

        cohort_dict = cohort.__dict__
        cohort_dict['tags'] = populate_cohort_tags(cohort.id, db_session)

        cohort_dict['validation'] =\
            populate_cohort_validation_status(cohort, db_session, cohort.size)

    # don't need to roll back session because it's just a query
    except Unauthorized:
        return 'You are not allowed to access this Cohort', 401
    except NoResultFound:
        return 'Could not find this Cohort', 404

    return json_response(cohort_dict)

 def report_result(self, results, child_results=None):
     """
     Creates a unique identifier for this ReportNode, and returns a one element
     dictionary with that identifier as the key and its results as the value.
     This allows ReportNode results to be merged as the tree of ReportNodes is
     evaluated.
     
     Parameters
         results         : Anything that the ReportNode compiles in its finish step
         child_results   : The results from a child Report(s) if they should be
                           preserved.  ReportLeaf results and any ReportNode results
                           that are copied should not be preserved.
     """
     if child_results is None:
         child_results = []
     
     self.result_key = str(uuid4())
     db_session = db.get_session()
     try:
         pj = db_session.query(PersistentReport).get(self.persistent_id)
         pj.result_key = self.result_key
         db_session.add(pj)
         db_session.commit()
     finally:
         db_session.close()
     
     merged = {self.result_key: results}
     for child_result in child_results:
         merged.update(child_result)
     return merged

 def __init__(self,
              user_id=None,
              status=celery.states.PENDING,
              name=None,
              result_key=None,
              children=[]):
     
     self.user_id = None
     try:
         if current_user.is_authenticated():
             self.user_id = current_user.id
     except RuntimeError:
         # nothing to worry about, just using current_user outside
         # of a web context.  This should only happen during testing
         pass
     
     self.status = status
     self.name = name
     self.result_key = result_key
     self.children = children
     
     # create PersistentJob and store id
     # note that result_key is always empty at this stage
     pj = PersistentJob(user_id=self.user_id,
                        status=self.status,
                        name=self.name,
                        show_in_ui=self.show_in_ui)
     db_session = db.get_session()
     db_session.add(pj)
     db_session.commit()
     self.persistent_id = pj.id

    def group_by_project(self):
        """
        mimics the interface of itertools.groupby, with the
        exception that the grouped items are simply user_ids
        rather than complete user records

        Returns:
            iterable of tuples of the form:
                (project, <iterable_of_usernames>)

        this is useful for turning a project-heterogenous cohort
        into a set of project-homogenous cohorts, which can be
        analyzed using a single database connection
        """
        db_session = db.get_session()
        user_id_projects = self.filter_wikiuser_query(
            db_session.query(WikiUserStore.mediawiki_userid, WikiUserStore.project)
        ).order_by(WikiUserStore.project).all()

        if not len(user_id_projects):
            return [(self.default_project, None)]

        groups = itertools.groupby(user_id_projects, key=itemgetter(1))

        return (
            (project or self.default_project, (r[0] for r in users))
            for project, users in groups
        )

def cohort_detail(name_or_id):
    """
    Returns a JSON object of the form:
    {id: 2, name: 'Berlin Beekeeping Society', description: '', wikiusers: [
        {mediawiki_username: 'Andrea', mediawiki_userid: 5, project: 'dewiki'},
        {mediawiki_username: 'Dennis', mediawiki_userid: 6, project: 'dewiki'},
        {mediawiki_username: 'Florian', mediawiki_userid: 7, project: 'dewiki'},
        {mediawiki_username: 'Gabriele', mediawiki_userid: 8, project: 'dewiki'},
    ]}
    """
    full_detail = request.args.get('full_detail', 0)
    
    cohort = None
    db_session = db.get_session()
    try:
        kargs = dict()
        if str(name_or_id).isdigit():
            kargs['by_id'] = int(name_or_id)
        else:
            kargs['by_name'] = name_or_id
        cohort = g.cohort_service.get_for_display(db_session, current_user.id, **kargs)
    except Unauthorized:
        return 'You are not allowed to access this Cohort', 401
    except NoResultFound:
        return 'Could not find this Cohort', 404
    finally:
        db_session.close()
    
    limit = 200 if full_detail == 'true' else 3
    cohort_with_wikiusers = populate_cohort_wikiusers(cohort, limit)
    cohort_with_tags = populate_cohort_tags(cohort_with_wikiusers, cohort.id)
    cohort_with_status = populate_cohort_validation_status(cohort_with_tags)
    return json_response(cohort_with_status)

 def login_for_testing_only():
     if app.config['DEBUG']:
         db_session = db.get_session()
         user = db_session.query(UserStore).filter_by(email='[email protected]').one()
         user.login(db_session)
         login_user(user)
         return ''

 def run(self):
     """
     This will get executed if the instance is added into a Report node hierarchy
     It outputs failure messages due to any invalid configuration.  None of these
     failures should happen unless the user tries to hack the system.
     """
     self.set_status(celery.states.STARTED, task_id=current_task.request.id)
     session = db.get_session()
     try:
         from wikimetrics.models.storage import ReportStore
         pj = session.query(ReportStore).get(self.persistent_id)
         pj.name = '{0} - {1} (failed validation)'.format(
             self.metric_label,
             self.cohort_name,
         )
         pj.status = celery.states.FAILURE
         session.commit()
     finally:
         session.close()
     
     message = ''
     if not self.metric_valid:
         message += '{0} was incorrectly configured\n'.format(
             self.metric_label,
         )
     return {'FAILURE': message or 'False'}

 def group_by_project(self):
     """
     mimics the interface of itertools.groupby, with the
     exception that the grouped items are simply user_ids
     rather than complete user records
     
     Returns:
         iterable of tuples of the form:
             (project, <iterable_of_usernames>)
     
     this is useful for turning a project-heterogenous cohort
     into a set of project-homogenous cohorts, which can be
     analyzed using a single database connection
     """
     db_session = db.get_session()
     try:
         user_id_projects = self.filter_wikiuser_query(
             db_session.query(WikiUser.mediawiki_userid, WikiUser.project)
         ).order_by(WikiUser.project).all()
     finally:
         db_session.close()
     # TODO: push this logic into sqlalchemy.  The solution
     # includes subquery(), but I can't seem to get anything working
     groups = itertools.groupby(user_id_projects, key=itemgetter(1))
     
     return (
         (project or self.default_project, (r[0] for r in users))
         for project, users in groups
     )

 def __init__(self,
              user_id=None,
              status=celery.states.PENDING,
              name=None,
              queue_result_key=None,
              children=[],
              parameters='{}'):
     
     self.user_id = user_id
     if not self.user_id:
         try:
             if current_user.is_authenticated():
                 self.user_id = current_user.id
         except RuntimeError:
             # nothing to worry about, just using current_user outside
             # of a web context.  This should only happen during testing
             pass
     
     self.status = status
     self.name = name
     self.queue_result_key = queue_result_key
     self.children = children
     
     # store report to database
     # note that queue_result_key is always empty at this stage
     pj = PersistentReport(user_id=self.user_id,
                           status=self.status,
                           name=self.name,
                           show_in_ui=self.show_in_ui,
                           parameters=parameters)
     db_session = db.get_session()
     db_session.add(pj)
     db_session.commit()
     self.persistent_id = pj.id
     db_session.close()

    def report_result(self, results, child_results=None):
        """
        NOTE: child_results is currently not used.  This function will still work
        as originally implemented, but child_results should go under evaluation.
        
        Creates a unique identifier for this ReportNode, and returns a one element
        dictionary with that identifier as the key and its results as the value.
        This allows ReportNode results to be merged as the tree of ReportNodes is
        evaluated.
        
        Parameters
            results         : Anything that the ReportNode compiles in its finish step
            child_results   : The results from a child Report(s) if they should be
                              preserved.  ReportLeaf results and any ReportNode results
                              that are copied should not be preserved.
        """
        if child_results is None:
            child_results = []
        
        self.result_key = str(uuid4())

        if self.store:
            db_session = db.get_session()
            pj = db_session.query(ReportStore).get(self.persistent_id)
            pj.result_key = self.result_key
            db_session.add(pj)
            db_session.commit()
        
        merged = {self.result_key: results}
        for child_result in child_results:
            merged.update(child_result)
        return merged

def delete_cohort(cohort_id):
    """
    Deletes a cohort and all its associated links if it belongs to only current_user
    Removes the relationship between current_user and this cohort if it belongs
    to a user other than current_user
    """
    session = db.get_session()
    try:
        owner_and_viewers = num_users(session, cohort_id)
        role = get_role(session, cohort_id)

        # Owner wants to delete, no other viewers or
        # Owner wants to delete, have other viewers, delete from other viewer's lists too
        if owner_and_viewers >= 1 and role == CohortUserRole.OWNER:
            delete_owner_cohort(session, cohort_id)
            session.commit()
            return json_redirect(url_for('cohorts_index'))

        # Viewer wants to delete cohort from their list, doesn't delete cohort from db;l,
        elif owner_and_viewers > 1 and role == CohortUserRole.VIEWER:
            delete_viewer_cohort(session, cohort_id)
            session.commit()
            return json_redirect(url_for('cohorts_index'))

        # None of the other cases fit.
        else:
            session.rollback()
            return json_error('This Cohort can not be deleted.')
    except DatabaseError as e:
        session.rollback()
        return json_error(e.message)
    finally:
        session.close()

def get_celery_task(result_key):
    """
    From a unique identifier, gets the celery task and database records associated.

    Parameters
        result_key  : The unique identifier found in the report database table
                        This parameter is required and should not be None

    Returns
        A tuple of the form (celery_task_object, database_report_object)
    """
    if not result_key:
        return (None, None)

    try:
        db_session = db.get_session()
        try:
            pj = db_session.query(ReportStore)\
                .filter(ReportStore.result_key == result_key)\
                .one()

            celery_task = Report.task.AsyncResult(pj.queue_result_key)
        finally:
            db_session.close()
        return (celery_task, pj)
    except NoResultFound:
        return (None, None)

    def set_public_report_state(report_id, owner_id, file_manager, public=True, data=''):
        """
        Internal method that sets a report public/private status.
        If we are making a report private that
        was public before will remove files from disk.

        If a new report is made public it will save report to disk.

        Validation that user can update this report has already happened before
        we reach this method.

        The UI calls this method on a per report basis
        but updates can be done for a set of reports.

        TODO: This method should not have http level code and
            should be part of an API,
            not be on the controller.
        TODO: We should not open & close a session here, I think session should be
            open/closed at the beginning/end of the request
            using flask request scoped functions

        Parameters:
            report_id   : id of PersistentReport to update
            owner_id    : the User purporting to own this report
            public      : True | False if True data must be present
            data        : String, report data to write out to filepath
            file_manager: PublicReportFileManager to manage io interactions

        Returns:
            Nothing

        Throws:
            Exception if there are issues making the report public or private

        A private report is has public=False
        """
        # NOTE: update_reports checks ownership and raises an exception if needed
        PersistentReport.update_reports([report_id], owner_id, public=public)

        # good no exception
        try:
            db_session = db.get_session()
            path = file_manager.get_public_report_path(report_id)
            if public:
                file_manager.write_data(path, data)

            else:
                file_manager.remove_file(path)

        except (PublicReportIOError, SQLAlchemyError) as e:
            app.logger.exception(str(e))
            # if there was an IO error rollback prior changes
            # this issues a new query as now our session scope and
            # transaction scope are now the same
            PersistentReport.update_reports([report_id], owner_id, public=not public)
            raise e

        finally:
            db_session.close()

def cohorts_index():
    """
    Renders a page with a list cohorts belonging to the currently logged in user.
    If the user is an admin, she has the option of seeing other users' cohorts.
    """
    session = db.get_session()
    tags = g.tag_service.get_all_tags(session)
    return render_template('cohorts.html', tags=json.dumps(tags))

 def run(self):
     session = db.get_session()
     try:
         cohort = session.query(Cohort).get(self.cohort_id)
         cohort.validation_queue_key = current_task.request.id
         session.commit()
         self.validate_records(session, cohort)
     finally:
         session.close()