diff --git a/gate_v1_function/tests/test_core.py b/gate_v1_function/tests/test_core.py new file mode 100644 index 0000000..5b0e9ce --- /dev/null +++ b/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)