'Correctly sets/dirties/cleans rows of DagStat table'

def test_dag_stats(self):

    session = settings.Session()
    session.query(models.DagRun).delete()
    session.query(models.DagStat).delete()
    session.commit()
    models.DagStat.update([], session=session)
    run1 = self.dag_bash.create_dagrun(run_id='run1', execution_date=DEFAULT_DATE, state=State.RUNNING)
    models.DagStat.update([self.dag_bash.dag_id], session=session)
    qry = session.query(models.DagStat).all()
    self.assertEqual(3, len(qry))
    self.assertEqual(self.dag_bash.dag_id, qry[0].dag_id)
    for stats in qry:
        if (stats.state == State.RUNNING):
            self.assertEqual(stats.count, 1)
        else:
            self.assertEqual(stats.count, 0)
        self.assertFalse(stats.dirty)
    run2 = self.dag_bash.create_dagrun(run_id='run2', execution_date=(DEFAULT_DATE + timedelta(days=1)), state=State.RUNNING)
    models.DagStat.update([self.dag_bash.dag_id], session=session)
    qry = session.query(models.DagStat).all()
    self.assertEqual(3, len(qry))
    self.assertEqual(self.dag_bash.dag_id, qry[0].dag_id)
    for stats in qry:
        if (stats.state == State.RUNNING):
            self.assertEqual(stats.count, 2)
        else:
            self.assertEqual(stats.count, 0)
        self.assertFalse(stats.dirty)
    session.query(models.DagRun).first().state = State.SUCCESS
    session.commit()
    models.DagStat.update([self.dag_bash.dag_id], session=session)
    qry = session.query(models.DagStat).filter((models.DagStat.state == State.SUCCESS)).all()
    self.assertEqual(1, len(qry))
    self.assertEqual(self.dag_bash.dag_id, qry[0].dag_id)
    self.assertEqual(State.SUCCESS, qry[0].state)
    self.assertEqual(1, qry[0].count)
    self.assertFalse(qry[0].dirty)
    qry = session.query(models.DagStat).filter((models.DagStat.state == State.RUNNING)).all()
    self.assertEqual(1, len(qry))
    self.assertEqual(self.dag_bash.dag_id, qry[0].dag_id)
    self.assertEqual(State.RUNNING, qry[0].state)
    self.assertEqual(1, qry[0].count)
    self.assertFalse(qry[0].dirty)
    session.query(models.DagRun).delete()
    session.query(models.DagStat).delete()
    session.commit()
    session.close()
