code: def test_correct_counts_update(
      self, admin_client, session, dag_maker, app, monkeypatch
  ):
      with monkeypatch.context() as m:
          datasets = [
              Dataset(uri=f"s3://bucket/key/{i}")
              for i in [1, 2, 3, 4, 5]
          ]
  
          # DAG that produces dataset #1
          with dag_maker(
              dag_id="upstream",
              schedule=None,
              serialized=True,
              session=session,
          ):
              EmptyOperator(task_id="task1", outlets=[datasets[0]])
  
          # DAG that is consumes only datasets #1 and #2
          with dag_maker(
              dag_id="downstream",
              schedule=datasets[:2],
              serialized=True,
              session=session,
          ):
              EmptyOperator(task_id="task1")
  
          # We create multiple dataset-producing and dataset-consuming DAGs because the query requires
          # COUNT(DISTINCT ...) for total_updates, or else it returns a multiple of the correct number due
          # to the outer joins with DagScheduleDatasetReference and TaskOutletDatasetReference
          # Two independent DAGs that produce dataset #3
          with dag_maker(
              dag_id="independent_producer_1",
              serialized=True,
              session=session,
          ):
              EmptyOperator(task_id="task1", outlets=[datasets[2]])
          with dag_maker(
              dag_id="independent_producer_2",
              serialized=True,
              session=session,
          ):
              EmptyOperator(task_id="task1", outlets=[datasets[2]])
          # Two independent DAGs that consume dataset #4
          with dag_maker(
              dag_id="independent_consumer_1",
              schedule=[datasets[3]],
              serialized=True,
              session=session,
          ):
              EmptyOperator(task_id="task1")
          with dag_maker(
              dag_id="independent_consumer_2",
              schedule=[datasets[3]],
              serialized=True,
              session=session,
          ):
              EmptyOperator(task_id="task1")
  
          # Independent DAG that is produces and consumes the same dataset, #5
          with dag_maker(
              dag_id="independent_producer_self_consumer",
              schedule=[datasets[4]],
              serialized=True,
              session=session,
          ):
              EmptyOperator(task_id="task1", outlets=[datasets[4]])
  
          m.setattr(app, "dag_bag", dag_maker.dagbag)
  
          ds1_id = (
              session.query(DatasetModel.id)
              .filter_by(uri=datasets[0].uri)
              .scalar()
          )
          ds2_id = (
              session.query(DatasetModel.id)
              .filter_by(uri=datasets[1].uri)
              .scalar()
          )
          ds3_id = (
              session.query(DatasetModel.id)
              .filter_by(uri=datasets[2].uri)
              .scalar()
          )
          ds4_id = (
              session.query(DatasetModel.id)
              .filter_by(uri=datasets[3].uri)
              .scalar()
          )
          ds5_id = (
              session.query(DatasetModel.id)
              .filter_by(uri=datasets[4].uri)
              .scalar()
          )
  
          # dataset 1 events
          session.add_all(
              [
                  DatasetEvent(
                      dataset_id=ds1_id,
                      timestamp=pendulum.DateTime(
                          2022, 8, 1, i, tzinfo=UTC
                      ),
                  )
                  for i in range(3)
              ]
          )
          # dataset 3 events
          session.add_all(
              [
                  DatasetEvent(
                      dataset_id=ds3_id,
                      timestamp=pendulum.DateTime(
                          2022, 8, 1, i, tzinfo=UTC
                      ),
                  )
                  for i in range(3)
              ]
          )
          # dataset 4 events
          session.add_all(
              [
                  DatasetEvent(
                      dataset_id=ds4_id,
                      timestamp=pendulum.DateTime(
                          2022, 8, 1, i, tzinfo=UTC
                      ),
                  )
                  for i in range(4)
              ]
          )
          # dataset 5 events
          session.add_all(
              [
                  DatasetEvent(
                      dataset_id=ds5_id,
                      timestamp=pendulum.DateTime(
                          2022, 8, 1, i, tzinfo=UTC
                      ),
                  )
                  for i in range(5)
              ]
          )
          session.commit()
  
          response = admin_client.get("/object/datasets_summary")
  
      assert response.status_code == 200
      response_data = response.json
      assert response_data == {
          "datasets": [
              {
                  "id": ds1_id,
                  "uri": "s3://bucket/key/1",
                  "last_dataset_update": "2022-08-01T02:00:00+00:00",
                  "total_updates": 3,
              },
              {
                  "id": ds2_id,
                  "uri": "s3://bucket/key/2",
                  "last_dataset_update": None,
                  "total_updates": 0,
              },
              {
                  "id": ds3_id,
                  "uri": "s3://bucket/key/3",
                  "last_dataset_update": "2022-08-01T02:00:00+00:00",
                  "total_updates": 3,
              },
              {
                  "id": ds4_id,
                  "uri": "s3://bucket/key/4",
                  "last_dataset_update": "2022-08-01T03:00:00+00:00",
                  "total_updates": 4,
              },
              {
                  "id": ds5_id,
                  "uri": "s3://bucket/key/5",
                  "last_dataset_update": "2022-08-01T04:00:00+00:00",
                  "total_updates": 5,
              },
          ],
          "total_entries": 5,
      }
  
imports: from __future__ import annotations
  import pendulum
  import pytest
  from dateutil.tz import UTC
  from airflow.datasets import Dataset
  from airflow.models.dataset import DatasetEvent, DatasetModel
  from airflow.operators.empty import EmptyOperator
  from tests.test_utils.asserts import assert_queries_count
  from tests.test_utils.db import clear_db_datasets
label_desc: Use image node_id 7 for calling the TestGetDatasets obj's underlying member method code with example usage: obj.test_correct_counts_update(admin_client, session, dag_maker, app, monkeypatch) without return types