|
14 | 14 | ExplicitContractOutputRuleIntegrationTestCase, |
15 | 15 | RulePassIntegrationTestCase, |
16 | 16 | RulesIntegrationTestCase, |
| 17 | + TypedContractRuleIntegrationTestCase, |
17 | 18 | ) |
18 | 19 |
|
19 | 20 |
|
@@ -240,6 +241,60 @@ def test_given_enforced_contract_when_compiling_explicit_output_rule_then_enforc |
240 | 241 | assert diagnostic_codes.count("SQBRCONTRACT105") == test_case.expected_rule_findings |
241 | 242 |
|
242 | 243 |
|
| 244 | +@pytest.mark.parametrize( |
| 245 | + "test_case", |
| 246 | + ( |
| 247 | + TypedContractRuleIntegrationTestCase( |
| 248 | + description="enforced contract with an untyped column fails only the typed-column rule", |
| 249 | + columns_sql='order_id (type INTEGER), status (description "Current status")', |
| 250 | + expected_exit_code=1, |
| 251 | + expected_contract_101_findings=0, |
| 252 | + expected_contract_105_findings=0, |
| 253 | + expected_contract_106_findings=1, |
| 254 | + ), |
| 255 | + TypedContractRuleIntegrationTestCase( |
| 256 | + description="enforced contract with fully typed explicit outputs passes", |
| 257 | + columns_sql="order_id (type INTEGER), status (type VARCHAR)", |
| 258 | + expected_exit_code=0, |
| 259 | + expected_contract_101_findings=0, |
| 260 | + expected_contract_105_findings=0, |
| 261 | + expected_contract_106_findings=0, |
| 262 | + ), |
| 263 | + ), |
| 264 | + ids=lambda case: case.description, |
| 265 | +) |
| 266 | +def test_given_contract_rules_when_compiling_then_contract_is_enforced_typed_and_explicit( |
| 267 | + test_case: TypedContractRuleIntegrationTestCase, |
| 268 | + tmp_path: Path, |
| 269 | + capsys: pytest.CaptureFixture[str], |
| 270 | +) -> None: |
| 271 | + (tmp_path / "sqlbuild_project.toml").write_text( |
| 272 | + 'name = "orders"\nadapter = "duckdb"\n\n[rules]\n' |
| 273 | + 'select = ["SQBRCONTRACT101", "SQBRCONTRACT105", "SQBRCONTRACT106"]\n', |
| 274 | + encoding="utf-8", |
| 275 | + ) |
| 276 | + model: Path = tmp_path / "models" / "orders.sql" |
| 277 | + model.parent.mkdir() |
| 278 | + model.write_text( |
| 279 | + "MODEL (\n" |
| 280 | + " contract enforced,\n" |
| 281 | + f" columns ({test_case.columns_sql}),\n" |
| 282 | + ");\n\n" |
| 283 | + "SELECT CAST(1 AS INTEGER) AS order_id, CAST('ready' AS VARCHAR) AS status\n", |
| 284 | + encoding="utf-8", |
| 285 | + ) |
| 286 | + |
| 287 | + exit_code: int = main(["--project-dir", str(tmp_path), "compile", "--json", "--no-cache"]) |
| 288 | + result: dict[str, object] = json.loads(capsys.readouterr().out) |
| 289 | + diagnostics: list[dict[str, object]] = cast(list[dict[str, object]], result["diagnostics"]) |
| 290 | + diagnostic_codes: tuple[object, ...] = tuple(item["code"] for item in diagnostics) |
| 291 | + |
| 292 | + assert exit_code == test_case.expected_exit_code |
| 293 | + assert diagnostic_codes.count("SQBRCONTRACT101") == (test_case.expected_contract_101_findings) |
| 294 | + assert diagnostic_codes.count("SQBRCONTRACT105") == (test_case.expected_contract_105_findings) |
| 295 | + assert diagnostic_codes.count("SQBRCONTRACT106") == (test_case.expected_contract_106_findings) |
| 296 | + |
| 297 | + |
243 | 298 | @pytest.mark.parametrize( |
244 | 299 | "test_case", |
245 | 300 | [RulesIntegrationTestCase("selected SQL rule blocks artifacts", 1, "SQBRSQL004")], |
|
0 commit comments