Adding Code Coverage webserver code and Fix TypeError and AttributeError in Views.py

tests/models/test_xcom_arg_map.py

@@ -20,6 +20,9 @@
 import pytest
 
 from airflow.exceptions import AirflowSkipException
+from airflow.models.taskinstance import TaskInstance
+from airflow.models.taskmap import TaskMap, TaskMapVariant
+from airflow.operators.empty import EmptyOperator
 from airflow.utils.state import TaskInstanceState
 from airflow.utils.trigger_rule import TriggerRule
 
@@ -188,6 +191,56 @@ def does_not_work_with_c(v):
     ]
 
 
+def test_task_map_from_task_instance_xcom():
+    task = EmptyOperator(task_id="test_task")
+    ti = TaskInstance(task=task, run_id="test_run", map_index=0)
+    ti.dag_id = "test_dag"
+    value = {"key1": "value1", "key2": "value2"}
+
+    # Test case where run_id is not None
+    task_map = TaskMap.from_task_instance_xcom(ti, value)
+    assert task_map.dag_id == ti.dag_id
+    assert task_map.task_id == ti.task_id
+    assert task_map.run_id == ti.run_id
+    assert task_map.map_index == ti.map_index
+    assert task_map.length == len(value)
+    assert task_map.keys == list(value)
+
+    # Test case where run_id is None
+    ti.run_id = None
+    with pytest.raises(ValueError, match="cannot record task map for unrun task instance"):
+        TaskMap.from_task_instance_xcom(ti, value)
+
+
+def test_task_map_with_invalid_task_instance():
+    task = EmptyOperator(task_id="test_task")
+    ti = TaskInstance(task=task, run_id=None, map_index=0)
+    ti.dag_id = "test_dag"
+
+    # Define some arbitrary XCom-like value data
+    value = {"example_key": "example_value"}
+
+    with pytest.raises(ValueError, match="cannot record task map for unrun task instance"):
+        TaskMap.from_task_instance_xcom(ti, value)
+
+
+def test_task_map_variant():
+    # Test case where keys is None
+    task_map = TaskMap(
+        dag_id="test_dag",
+        task_id="test_task",
+        run_id="test_run",
+        map_index=0,
+        length=3,
+        keys=None,
+    )
+    assert task_map.variant == TaskMapVariant.LIST
+
+    # Test case where keys is not None
+    task_map.keys = ["key1", "key2"]
+    assert task_map.variant == TaskMapVariant.DICT
+
+
 def test_xcom_map_raise_to_skip(dag_maker, session):
     result = None
 

tests/www/test_utils.py

@@ -28,8 +28,10 @@
 import pytest
 from bs4 import BeautifulSoup
 from flask_appbuilder.models.sqla.filters import get_field_setup_query, set_value_to_type
+from flask_wtf import FlaskForm
 from markupsafe import Markup
 from sqlalchemy.orm import Query
+from wtforms.fields import StringField, TextAreaField
 
 from airflow.models import DagRun
 from airflow.utils import json as utils_json
@@ -41,6 +43,7 @@
     json_f,
     wrapped_markdown,
 )
+from airflow.www.widgets import AirflowDateTimePickerROWidget, BS3TextAreaROWidget, BS3TextFieldROWidget
 from tests.test_utils.config import conf_vars
 
 
@@ -703,3 +706,52 @@ def test_dag_run_custom_sqla_interface_delete_no_collateral_damage(dag_maker, se
     assert len(dag_runs) == 6
     assert len(set(x.dag_id for x in dag_runs)) == 3
     assert len(set(x.run_id for x in dag_runs)) == 3
+
+
+@pytest.fixture
+def app():
+    from flask import Flask
+
+    app = Flask(__name__)
+    app.config["WTF_CSRF_ENABLED"] = False
+    app.config["SECRET_KEY"] = "secret"
+    with app.app_context():
+        yield app
+
+
+class TestWidgets:
+    def test_airflow_datetime_picker_ro_widget(self, app):
+        class TestForm(FlaskForm):
+            datetime_field = StringField(widget=AirflowDateTimePickerROWidget())
+
+        form = TestForm()
+        field = form.datetime_field
+
+        html_output = field()
+
+        assert 'readonly="true"' in html_output
+        assert "input-group datetime datetimepicker" in html_output
+
+    def test_bs3_text_field_ro_widget(self, app):
+        class TestForm(FlaskForm):
+            text_field = StringField(widget=BS3TextFieldROWidget())
+
+        form = TestForm()
+        field = form.text_field
+
+        html_output = field()
+
+        assert 'readonly="true"' in html_output
+        assert "form-control" in html_output
+
+    def test_bs3_text_area_ro_widget(self, app):
+        class TestForm(FlaskForm):
+            textarea_field = TextAreaField(widget=BS3TextAreaROWidget())
+
+        form = TestForm()
+        field = form.textarea_field
+
+        html_output = field()
+
+        assert 'readonly="true"' in html_output
+        assert "form-control" in html_output

tests/www/views/test_views_tasks.py

@@ -43,7 +43,7 @@
 from airflow.utils.session import create_session
 from airflow.utils.state import DagRunState, State
 from airflow.utils.types import DagRunType
-from airflow.www.views import TaskInstanceModelView
+from airflow.www.views import TaskInstanceModelView, _safe_parse_datetime
 from tests.test_utils.api_connexion_utils import create_user, delete_roles, delete_user
 from tests.test_utils.config import conf_vars
 from tests.test_utils.db import clear_db_runs, clear_db_xcom
@@ -66,7 +66,7 @@ def reset_dagruns():
 
 
 @pytest.fixture(autouse=True)
-def init_dagruns(app, reset_dagruns):
+def init_dagruns(app):
     with time_machine.travel(DEFAULT_DATE, tick=False):
         app.dag_bag.get_dag("example_bash_operator").create_dagrun(
             run_id=DEFAULT_DAGRUN,
@@ -1050,6 +1050,35 @@ def test_graph_view_doesnt_fail_on_recursion_error(app, dag_maker, admin_client)
         assert resp.status_code == 200
 
 
+def test_get_date_time_num_runs_dag_runs_form_data_graph_view(app, dag_maker, admin_client):
+    """Test the get_date_time_num_runs_dag_runs_form_data function."""
+    from airflow.www.views import get_date_time_num_runs_dag_runs_form_data
+
+    execution_date = pendulum.now(tz="UTC")
+    with dag_maker(
+        dag_id="test_get_date_time_num_runs_dag_runs_form_data",
+        start_date=execution_date,
+    ) as dag:
+        BashOperator(task_id="task_1", bash_command="echo test")
+
+    with unittest.mock.patch.object(app, "dag_bag") as mocked_dag_bag:
+        mocked_dag_bag.get_dag.return_value = dag
+        url = f"/dags/{dag.dag_id}/graph"
+        resp = admin_client.get(url, follow_redirects=True)
+        assert resp.status_code == 200
+
+    with create_session() as session:
+        data = get_date_time_num_runs_dag_runs_form_data(resp.request, session, dag)
+
+        dttm = pendulum.parse(data["dttm"].isoformat())
+        base_date = pendulum.parse(data["base_date"].isoformat())
+
+        assert dttm.date() == execution_date.date()
+        assert dttm.time().hour == _safe_parse_datetime(execution_date.time().isoformat()).time().hour
+        assert dttm.time().minute == _safe_parse_datetime(execution_date.time().isoformat()).time().minute
+        assert base_date.date() == execution_date.date()
+
+
 def test_task_instances(admin_client):
     """Test task_instances view."""
     resp = admin_client.get(