clean up and improve code in test"

Author: aMAAmina

tests/test_data_normalisation.py

@@ -1,12 +1,7 @@
 import csv
-import os
-import tempfile
-import builtins
-import pytest
-
 import nifipulse.config as config
 from nifipulse.data_normalisation import process_data   
-
+from types import SimpleNamespace
 
 
 def create_csv(path, rows):
@@ -52,11 +47,8 @@ def test_process_data(tmp_path, monkeypatch):
 
     create_csv(input_csv, rows)
 
-    
-    monkeypatch.setattr(config, "env", type("E", (), {
-        "CSV_SINK": str(input_csv),
-        "CLEAN_DATA": str(output_csv)
-    }))
+    cfg = SimpleNamespace(CSV_SINK=str(input_csv), CLEAN_DATA=str(output_csv))
+    monkeypatch.setattr(config, "env", cfg, raising=False)
 
     # Run the function
     process_data()

tests/test_extract_metrics.py

@@ -2,8 +2,6 @@
 from unittest.mock import patch, MagicMock
 import nifipulse.extract_metrics as em
 import nifipulse.config as config
-import os
-
 
 # FIXTURE POUR MOCKER config.env + CRÉER results/
 @pytest.fixture
@@ -25,65 +23,63 @@ class Env:
 # TESTS poll_metrics()
 
 def test_poll_metrics_no_metrics(mock_env):
-
     with patch("builtins.print") as mock_print:
         em.poll_metrics(metrics=[])
         mock_print.assert_called_with("No metrics provided to poll.")
 
 
 
 def test_poll_metrics_calls_prometheus_and_writes_csv(mock_env):
-
     metrics = ["nifi_metric"]
     # Mock réponse Prometheus
     fake_resp = MagicMock()
+    fake_resp.raise_for_status.return_value = None
     fake_resp.json.return_value = {
         "data": {
             "result": [{
-                "metric": {"instance": "nifi:9092", "component_name": "UpdateAttribute"},
-                "value": ["0", "5"]
+                "metric": {"instance": "nifi:9092", "component_name": "UpdateAttribute", "component_id": "cid1", "component_type": "PROCESSOR"},
+                "value": [1700000000, "5"]
             }]
         }
     }
 
-    with patch("requests.get", return_value=fake_resp):
+    # Patch where it's used
+    with patch("nifipulse.extract_metrics.requests.get", return_value=fake_resp):
         em.poll_metrics(metrics=metrics, interval=0, count=1)
 
     # Vérifier écriture du CSV
     with open(config.env.CSV_SINK) as f:
         lines = f.readlines()
 
     assert len(lines) == 2  # header + data line
-
+    assert "UpdateAttribute" in lines[1]
+    assert ",5" in lines[1]
 
 
 # TESTS nifipulse()
 
 def test_nifipulse_no_rows(mock_env):
-
     with patch("nifipulse.extract_metrics.path_tofolder", return_value=True), \
          patch("nifipulse.extract_metrics.files") as mock_files, \
+         patch("nifipulse.extract_metrics.poll_metrics") as mock_poll, \
          patch("nifipulse.extract_metrics._csv_has_rows", return_value=False), \
          patch("builtins.print") as mock_print:
         mock_files.return_value.joinpath.return_value.read_text.return_value = "metric1"
         em.nifipulse(poll_count=1, interval=0)
+        mock_poll.assert_called_once()
         mock_print.assert_any_call("No polled rows written; skipping normalization.")
 
 
 
 def test_nifipulse_full_flow(mock_env):
-
     with patch("nifipulse.extract_metrics.path_tofolder", return_value=True), \
          patch("nifipulse.extract_metrics.files") as mock_files, \
          patch("nifipulse.extract_metrics._csv_has_rows", side_effect=[True, True]), \
          patch("nifipulse.extract_metrics.process_data") as mock_process, \
          patch("nifipulse.extract_metrics.load_postgres") as mock_pg, \
          patch("nifipulse.extract_metrics.poll_metrics"):
-
         mock_files.return_value.joinpath.return_value.read_text.return_value = "metricX"
-
         em.nifipulse(poll_count=1, interval=0)
-
         mock_process.assert_called_once()
         mock_pg.assert_called_once()
 

tests/test_load_postgre.py

@@ -1,11 +1,10 @@
-import pytest
 import pandas as pd
 from unittest.mock import patch, MagicMock
+from sqlalchemy.sql.elements import TextClause
 from nifipulse.load_postgres import load_postgres
 
 
 def test_load_postgres_success(tmp_path):
-
     # Create fake CSV
     csv_file = tmp_path / "sample.csv"
     csv_file.write_text(
@@ -62,14 +61,18 @@ def test_load_postgres_success(tmp_path):
 
     with patch("nifipulse.load_postgres.create_engine", return_value=mock_engine), \
          patch("nifipulse.load_postgres.pd.read_sql", side_effect=read_sql_side_effect):
-
         load_postgres(str(csv_file))
 
-    # Assertions
-    calls = [c.args[0].text for c in mock_conn.execute.call_args_list]
+    # rom sqlalchemy.sql.elements import TextClause
+    texts = []
+    for c in mock_conn.execute.call_args_list:
+        arg0 = c.args[0]
+        if isinstance(arg0, TextClause):
+            texts.append(arg0.text)
 
-    assert any("INSERT INTO dim_instance" in sql for sql in calls)
-    assert any("INSERT INTO dim_metric" in sql for sql in calls)
-    assert any("INSERT INTO dim_component" in sql for sql in calls)
-    assert any("INSERT INTO dim_date" in sql for sql in calls)
-    assert any("INSERT INTO fact_metrics" in sql for sql in calls)
+    assert any("SELECT 1 FROM dim_instance" in sql for sql in texts), "Schema check missing"
+    assert any("INSERT INTO dim_instance" in sql for sql in texts)
+    assert any("INSERT INTO dim_metric" in sql for sql in texts)
+    assert any("INSERT INTO dim_component" in sql for sql in texts)
+    assert any("INSERT INTO dim_date" in sql for sql in texts)
+    assert any("INSERT INTO fact_metrics" in sql for sql in texts), "Fact insert missing"