64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
import json
|
|
|
|
from tht.decisions import append_decision
|
|
from tht.phase import advance_problems
|
|
|
|
|
|
def _phase_four_session(tmp_path):
|
|
session = tmp_path / "session"
|
|
session.mkdir()
|
|
for phase in range(1, 4):
|
|
append_decision(session, type="phase_approved", subject=f"phase:{phase}")
|
|
return session
|
|
|
|
|
|
def test_phase_four_cannot_close_multiple_tables_without_reviewed_structured_joins(tmp_path):
|
|
session = _phase_four_session(tmp_path)
|
|
(session / "schema_linking.json").write_text(json.dumps({
|
|
"question": "q",
|
|
"candidates": [
|
|
{"kind": "table", "name": "fact_event", "decision": "promoted"},
|
|
{"kind": "table", "name": "dim_patient", "decision": "promoted"},
|
|
],
|
|
"joins": [],
|
|
}))
|
|
|
|
problems = advance_problems(session, 4)
|
|
|
|
assert any("join" in problem.lower() for problem in problems)
|
|
|
|
|
|
def test_phase_four_allows_a_single_promoted_table_without_joins(tmp_path):
|
|
session = _phase_four_session(tmp_path)
|
|
(session / "schema_linking.json").write_text(json.dumps({
|
|
"question": "q",
|
|
"candidates": [
|
|
{"kind": "table", "name": "dim_patient", "decision": "promoted"},
|
|
],
|
|
"joins": [],
|
|
}))
|
|
|
|
assert advance_problems(session, 4) == []
|
|
|
|
|
|
def test_phase_four_allows_reviewed_structured_joins_for_multiple_tables(tmp_path):
|
|
session = _phase_four_session(tmp_path)
|
|
append_decision(
|
|
session,
|
|
type="join_modified",
|
|
subject="fact_event_to_patient",
|
|
detail="fact_event.cod_paz = dim_patient.cod_paz",
|
|
)
|
|
(session / "schema_linking.json").write_text(json.dumps({
|
|
"question": "q",
|
|
"candidates": [
|
|
{"kind": "table", "name": "fact_event", "decision": "promoted"},
|
|
{"kind": "table", "name": "dim_patient", "decision": "promoted"},
|
|
],
|
|
"joins": [
|
|
{"from": "fact_event.cod_paz", "to": "dim_patient.cod_paz"},
|
|
],
|
|
}))
|
|
|
|
assert advance_problems(session, 4) == []
|