Spaces:
Sleeping
Sleeping
File size: 2,421 Bytes
9b10f02 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
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
|