tests/test_fetch_traffic_data.py

import pytest
import sys
import os
import requests
import pandas as pd
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from airflow.dags.tasks.fetch_traffic_data import _fetch_trafic_data, _process_traffic_data, fetch_traffic_data


@pytest.fixture
def fake_api_response():
    return {
        "results": [
            {
                "datetime": "2025-07-11T08:00:00+02:00",
                "denomination": "Axe 1",
                "averagevehiclespeed": 50,
                "traveltime": 5,
                "trafficstatus": "freeFlow"
            },
            {
                "datetime": "2025-07-11T08:00:00+02:00",
                "denomination": "Axe 2",
                "averagevehiclespeed": 20,
                "traveltime": 10,
                "trafficstatus": "heavy"
            },
            {
                "datetime": "2025-07-10T08:00:00+02:00",
                "denomination": "Axe 3",
                "averagevehiclespeed": 10,
                "traveltime": 15,
                "trafficstatus": "congested"
            },
        ]
    }


def test__process_traffic_data(fake_api_response):
    mean_status = _process_traffic_data(fake_api_response["results"])
    assert isinstance(mean_status, float)
    assert 0 <= mean_status <= 2


def test__fetch_trafic_data(requests_mock, fake_api_response):
    mock_url = "https://data.rennesmetropole.fr/api/explore/v2.1/catalog/datasets/etat-du-trafic-en-temps-reel/records"
    requests_mock.get(mock_url, json=fake_api_response)

    results = _fetch_trafic_data()
    assert isinstance(results, list)
    assert len(results) == 3
    assert all("denomination" in r for r in results)


def test_fetch_traffic_data_push_xcom(fake_api_response, monkeypatch):
    # Monkeypatch les fonctions internes pour mocker les appels API
    monkeypatch.setattr("airflow.dags.tasks.fetch_traffic_data._fetch_trafic_data", lambda: fake_api_response["results"])
    monkeypatch.setattr("airflow.dags.tasks.fetch_traffic_data._process_traffic_data", lambda data: 1.0)

    class DummyTaskInstance:
        def xcom_push(self, key, value):
            self.key = key
            self.value = value

    context = {"task_instance": DummyTaskInstance()}
    result = fetch_traffic_data(**context)
    assert result == 1.0
    assert context["task_instance"].key == "traffic_value"
    assert context["task_instance"].value == 1.0