Add gate_v1_function/tests/test_core.py

gate_v1_function/tests/test_core.py

@@ -0,0 +1,77 @@
+import json
+import pandas as pd
+import pytest
+from pathlib import Path
+from gate_v1_function import core
+
+@pytest.fixture
+def tmp_json(tmp_path):
+    def _create_json(data, filename):
+        fpath = tmp_path / filename
+        with open(fpath, 'w', encoding='utf-8') as f:
+            json.dump(data, f)
+        return fpath
+    return _create_json
+
+@pytest.fixture
+def tmp_csv(tmp_path):
+    def _create_csv(rows, filename):
+        df = pd.DataFrame(rows)
+        fpath = tmp_path / filename
+        df.to_csv(fpath, index=False)
+        return fpath
+    return _create_csv
+
+def test_calculate_gate_decision_pass(tmp_json, tmp_csv):
+    delta_summary = {"summary": {"PASS": 10, "FAIL": 0}}
+    delta_cases = [
+        {"stratum": "s1", "from_status": "PASS", "to_status": "PASS", "unknown_reason": ""}
+    ]
+    summary_path = tmp_json(delta_summary, 'delta_summary.json')
+    cases_path = tmp_csv(delta_cases, 'delta_cases.csv')
+    
+    result = core.calculate_gate_decision(str(summary_path), str(cases_path), None)
+    assert isinstance(result, dict)
+    assert result['decision'] == 'PASS'
+    assert isinstance(result['reasons'], list)
+
+
+def test_calculate_gate_decision_review_with_unknown(tmp_json, tmp_csv):
+    delta_summary = {"summary": {"PASS": 8, "UNKNOWN": 2}}
+    delta_cases = [
+        {"stratum": "s1", "from_status": "PASS", "to_status": "UNKNOWN", "unknown_reason": "flaky network"},
+        {"stratum": "s2", "from_status": "PASS", "to_status": "UNKNOWN", "unknown_reason": "timeout"}
+    ]
+    whitelist = ["timeout"]
+    summary_path = tmp_json(delta_summary, 'delta_summary.json')
+    cases_path = tmp_csv(delta_cases, 'delta_cases.csv')
+    whitelist_path = tmp_json(whitelist, 'unknown_whitelist.json')
+
+    result = core.calculate_gate_decision(str(summary_path), str(cases_path), str(whitelist_path))
+    assert result['decision'] in ('REVIEW', 'PASS')
+    assert isinstance(result['reasons'], list)
+    if result['decision'] == 'REVIEW':
+        assert any('flaky' in r or 'unknown' in r.lower() for r in result['reasons'])
+
+
+def test_calculate_gate_decision_block_on_fail(tmp_json, tmp_csv):
+    delta_summary = {"summary": {"FAIL": 2}}
+    delta_cases = [
+        {"stratum": "s1", "from_status": "PASS", "to_status": "FAIL", "unknown_reason": "crash detected"},
+        {"stratum": "s2", "from_status": "PASS", "to_status": "FAIL", "unknown_reason": "test regression"}
+    ]
+
+    summary_path = tmp_json(delta_summary, 'delta_summary.json')
+    cases_path = tmp_csv(delta_cases, 'delta_cases.csv')
+
+    result = core.calculate_gate_decision(str(summary_path), str(cases_path), None)
+    assert result['decision'] == 'BLOCK'
+    assert any('FAIL' in r or 'block' in r.lower() for r in result['reasons'])
+
+
+def test_invalid_inputs(tmp_path):
+    fake_path = tmp_path / 'nonexistent.json'
+    csv_path = tmp_path / 'empty.csv'
+    csv_path.write_text('stratum,from_status,to_status,unknown_reason\n')
+    with pytest.raises((FileNotFoundError, ValueError, json.JSONDecodeError)):
+        core.calculate_gate_decision(str(fake_path), str(csv_path), None)