'Test assigning Operators to Dags, including deferred assignment'

def test_set_dag(self):

    dag = DAG(u'dag', start_date=DEFAULT_DATE)
    dag2 = DAG(u'dag2', start_date=DEFAULT_DATE)
    op = DummyOperator(task_id=u'op_1', owner=u'test')
    self.assertFalse(op.has_dag())
    self.assertRaises(AirflowException, getattr, op, u'dag')
    with self.assertRaises(TypeError):
        op.dag = 1
    op.dag = dag
    with self.assertRaises(AirflowException):
        op.dag = dag2
    op.dag = dag
    self.assertIs(op.dag, dag)
    self.assertIn(op, dag.tasks)
