""" Tests for verify_wbs_task_v1.py and validate_quant_engine_wbs_v1.py """ from __future__ import annotations import json import subprocess import sys from pathlib import Path import pytest import yaml # Import the modules to test sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from tools.verify_wbs_task_v1 import parse_dotnet_connection_string class TestParseDotnetConnectionString: """Test .NET connection string parsing.""" def test_parse_basic_connection_string(self): """Test parsing a basic connection string.""" s = "Host=localhost;Port=5432;Database=testdb;Username=user;Password=pass;" result = parse_dotnet_connection_string(s) assert result["host"] == "localhost" assert result["port"] == "5432" assert result["dbname"] == "testdb" assert result["user"] == "user" assert result["password"] == "pass" assert "options" not in result def test_parse_with_search_path(self): """Test parsing with Search Path.""" s = "Host=localhost;Database=testdb;Username=user;Password=pass;Search Path=myschema;" result = parse_dotnet_connection_string(s) assert result["host"] == "localhost" assert result["dbname"] == "testdb" assert result["options"] == "-c search_path=myschema" def test_parse_case_insensitive(self): """Test case-insensitive key handling.""" s = "HOST=localhost;DATABASE=testdb;USERNAME=user;PASSWORD=pass;" result = parse_dotnet_connection_string(s) assert result["host"] == "localhost" assert result["dbname"] == "testdb" assert result["user"] == "user" assert result["password"] == "pass" def test_parse_unknown_keys_ignored(self): """Test that unknown keys are ignored.""" s = "Host=localhost;UnknownKey=value;Database=testdb;" result = parse_dotnet_connection_string(s) assert result["host"] == "localhost" assert result["dbname"] == "testdb" assert "UnknownKey" not in result and "unknownkey" not in result def test_parse_empty_string(self): """Test parsing empty string.""" result = parse_dotnet_connection_string("") assert result == {} def test_parse_with_spaces(self): """Test parsing with extra spaces.""" s = " Host = localhost ; Database = testdb ; " result = parse_dotnet_connection_string(s) assert result["host"] == "localhost" assert result["dbname"] == "testdb" class TestJsonGateComparison: """Test json_gate comparison operator parsing.""" def test_json_gate_exact_equality(self, tmp_path): """Test exact equality in json_gate.""" repo_root = tmp_path / "repo" repo_root.mkdir() spec_path = repo_root / "spec" / "60_quant_engine_wbs.yaml" spec_path.parent.mkdir() spec = { "meta": {"formula_id": "QUANT_ENGINE_WBS_V1"}, "tasks": { "TEST-01": { "title": "Test task", "status": "PENDING", "depends_on": [], "success_criteria": { "expected_success_value": {}, "evidence_artifacts": [], "verification_commands": [] }, "evidence_checks": [ { "type": "json_gate", "path": "Temp/test_output.json", "expect": {"gate": "PASS", "count": 5} } ] } } } spec_path.write_text(yaml.dump(spec), encoding="utf-8") # Create test artifact artifact_path = repo_root / "Temp" / "test_output.json" artifact_path.parent.mkdir(parents=True) artifact_path.write_text(json.dumps({"gate": "PASS", "count": 5}), encoding="utf-8") # Run verify_wbs_task_v1 tools_dir = Path(__file__).resolve().parents[2] / "tools" result = subprocess.run( [sys.executable, str(tools_dir / "verify_wbs_task_v1.py"), "--task", "TEST-01", "--repo-root", str(repo_root), "--spec", str(spec_path)], cwd=repo_root, capture_output=True, text=True ) assert result.returncode == 0, result.stderr def test_json_gate_greater_equal_comparison(self, tmp_path): """Test >= comparison in json_gate.""" repo_root = tmp_path / "repo" repo_root.mkdir() spec_path = repo_root / "spec" / "60_quant_engine_wbs.yaml" spec_path.parent.mkdir() spec = { "meta": {"formula_id": "QUANT_ENGINE_WBS_V1"}, "tasks": { "TEST-02": { "title": "Test task", "status": "PENDING", "depends_on": [], "success_criteria": { "expected_success_value": {}, "evidence_artifacts": [], "verification_commands": [] }, "evidence_checks": [ { "type": "json_gate", "path": "Temp/test_output.json", "expect": {"compared_count": ">=20"} } ] } } } spec_path.write_text(yaml.dump(spec), encoding="utf-8") # Create test artifact - PASS case artifact_path = repo_root / "Temp" / "test_output.json" artifact_path.parent.mkdir(parents=True) artifact_path.write_text(json.dumps({"compared_count": 25}), encoding="utf-8") tools_dir = Path(__file__).resolve().parents[2] / "tools" result = subprocess.run( [sys.executable, str(tools_dir / "verify_wbs_task_v1.py"), "--task", "TEST-02", "--repo-root", str(repo_root), "--spec", str(spec_path)], cwd=repo_root, capture_output=True, text=True ) assert result.returncode == 0, result.stderr def test_json_gate_greater_equal_fail(self, tmp_path): """Test >= comparison fails when below threshold.""" repo_root = tmp_path / "repo" repo_root.mkdir() spec_path = repo_root / "spec" / "60_quant_engine_wbs.yaml" spec_path.parent.mkdir() spec = { "meta": {"formula_id": "QUANT_ENGINE_WBS_V1"}, "tasks": { "TEST-03": { "title": "Test task", "status": "PENDING", "depends_on": [], "success_criteria": { "expected_success_value": {}, "evidence_artifacts": [], "verification_commands": [] }, "evidence_checks": [ { "type": "json_gate", "path": "Temp/test_output.json", "expect": {"compared_count": ">=20"} } ] } } } spec_path.write_text(yaml.dump(spec), encoding="utf-8") # Create test artifact - FAIL case (value too low) artifact_path = repo_root / "Temp" / "test_output.json" artifact_path.parent.mkdir(parents=True) artifact_path.write_text(json.dumps({"compared_count": 15}), encoding="utf-8") tools_dir = Path(__file__).resolve().parents[2] / "tools" result = subprocess.run( [sys.executable, str(tools_dir / "verify_wbs_task_v1.py"), "--task", "TEST-03", "--repo-root", str(repo_root), "--spec", str(spec_path)], cwd=repo_root, capture_output=True, text=True ) assert result.returncode == 1, f"Expected failure, got: {result.stdout}" class TestLogPatternCheck: """Test log_pattern check type.""" def test_log_pattern_min_matches_pass(self, tmp_path): """Test log_pattern with min_matches that passes.""" repo_root = tmp_path / "repo" repo_root.mkdir() spec_path = repo_root / "spec" / "60_quant_engine_wbs.yaml" spec_path.parent.mkdir() spec = { "meta": {"formula_id": "QUANT_ENGINE_WBS_V1"}, "tasks": { "TEST-04": { "title": "Test task", "status": "PENDING", "depends_on": [], "success_criteria": { "expected_success_value": {}, "evidence_artifacts": [], "verification_commands": [] }, "evidence_checks": [ { "type": "log_pattern", "file_glob": "Temp/test.log", "pattern": "SUCCESS", "expect": {"min_matches": 1} } ] } } } spec_path.write_text(yaml.dump(spec), encoding="utf-8") # Create log file with matches log_path = repo_root / "Temp" / "test.log" log_path.parent.mkdir(parents=True) log_path.write_text("Operation SUCCESS\nAnother line\nOperation SUCCESS\n", encoding="utf-8") tools_dir = Path(__file__).resolve().parents[2] / "tools" result = subprocess.run( [sys.executable, str(tools_dir / "verify_wbs_task_v1.py"), "--task", "TEST-04", "--repo-root", str(repo_root), "--spec", str(spec_path)], cwd=repo_root, capture_output=True, text=True ) assert result.returncode == 0, result.stderr def test_log_pattern_max_matches_zero(self, tmp_path): """Test log_pattern with max_matches=0 (no file should match).""" repo_root = tmp_path / "repo" repo_root.mkdir() spec_path = repo_root / "spec" / "60_quant_engine_wbs.yaml" spec_path.parent.mkdir() spec = { "meta": {"formula_id": "QUANT_ENGINE_WBS_V1"}, "tasks": { "TEST-05": { "title": "Test task", "status": "PENDING", "depends_on": [], "success_criteria": { "expected_success_value": {}, "evidence_artifacts": [], "verification_commands": [] }, "evidence_checks": [ { "type": "log_pattern", "file_glob": "Temp/nonexistent.log", "pattern": "ERROR", "expect": {"max_matches": 0} } ] } } } spec_path.write_text(yaml.dump(spec), encoding="utf-8") tools_dir = Path(__file__).resolve().parents[2] / "tools" result = subprocess.run( [sys.executable, str(tools_dir / "verify_wbs_task_v1.py"), "--task", "TEST-05", "--repo-root", str(repo_root), "--spec", str(spec_path)], cwd=repo_root, capture_output=True, text=True ) # Should pass because file doesn't exist = 0 matches assert result.returncode == 0, result.stderr class TestFileExistsCheck: """Test file_exists check type.""" def test_file_exists_with_min_bytes_fail(self, tmp_path): """Test file_exists fails when file is too small.""" repo_root = tmp_path / "repo" repo_root.mkdir() spec_path = repo_root / "spec" / "60_quant_engine_wbs.yaml" spec_path.parent.mkdir() spec = { "meta": {"formula_id": "QUANT_ENGINE_WBS_V1"}, "tasks": { "TEST-06": { "title": "Test task", "status": "PENDING", "depends_on": [], "success_criteria": { "expected_success_value": {}, "evidence_artifacts": [], "verification_commands": [] }, "evidence_checks": [ { "type": "file_exists", "paths": ["Temp/small_file.txt"], "expect": {"min_bytes": 100} } ] } } } spec_path.write_text(yaml.dump(spec), encoding="utf-8") # Create a file smaller than min_bytes file_path = repo_root / "Temp" / "small_file.txt" file_path.parent.mkdir(parents=True) file_path.write_text("small", encoding="utf-8") tools_dir = Path(__file__).resolve().parents[2] / "tools" result = subprocess.run( [sys.executable, str(tools_dir / "verify_wbs_task_v1.py"), "--task", "TEST-06", "--repo-root", str(repo_root), "--spec", str(spec_path)], cwd=repo_root, capture_output=True, text=True ) assert result.returncode == 1, f"Expected failure, got: {result.stdout}" class TestValidatorLogic: """Test validate_quant_engine_wbs_v1.py logic.""" def test_validator_done_task_with_passing_verdict(self, tmp_path): """Test validator passes when DONE task has PASS verdict.""" repo_root = tmp_path / "repo" repo_root.mkdir() spec_path = repo_root / "spec" / "60_quant_engine_wbs.yaml" spec_path.parent.mkdir() spec = { "meta": {"formula_id": "QUANT_ENGINE_WBS_V1"}, "tasks": { "TEST-07": { "title": "Test task", "status": "DONE", "depends_on": [], "success_criteria": { "expected_success_value": {}, "evidence_artifacts": ["Temp/evidence/TEST-07/verdict.json"], "verification_commands": [] }, "evidence_checks": [{"type": "file_exists", "paths": ["spec/60_quant_engine_wbs.yaml"]}] } } } spec_path.write_text(yaml.dump(spec), encoding="utf-8") # Create a passing verdict verdict_path = repo_root / "Temp" / "evidence" / "TEST-07" / "verdict.json" verdict_path.parent.mkdir(parents=True) verdict_path.write_text( json.dumps({ "task_id": "TEST-07", "formula_id": "QUANT_ENGINE_WBS_TASK_V1", "gate": "PASS", "checks": [] }), encoding="utf-8" ) # Create roadmap pointer roadmap_path = repo_root / "docs" / "ROADMAP_WBS.md" roadmap_path.parent.mkdir(parents=True) roadmap_path.write_text("# Roadmap\n\nQUANT_ENGINE_WBS_V1 is the standard.\n", encoding="utf-8") tools_dir = Path(__file__).resolve().parents[2] / "tools" result = subprocess.run( [sys.executable, str(tools_dir / "validate_quant_engine_wbs_v1.py"), "--repo-root", str(repo_root), "--spec", str(spec_path)], cwd=repo_root, capture_output=True, text=True ) assert result.returncode == 0, result.stderr def test_validator_done_task_without_verdict(self, tmp_path): """Test validator fails when DONE task is missing verdict.""" repo_root = tmp_path / "repo" repo_root.mkdir() spec_path = repo_root / "spec" / "60_quant_engine_wbs.yaml" spec_path.parent.mkdir() spec = { "meta": {"formula_id": "QUANT_ENGINE_WBS_V1"}, "tasks": { "TEST-08": { "title": "Test task", "status": "DONE", "depends_on": [], "success_criteria": { "expected_success_value": {}, "evidence_artifacts": [], "verification_commands": [] }, "evidence_checks": [] } } } spec_path.write_text(yaml.dump(spec), encoding="utf-8") # Create roadmap pointer roadmap_path = repo_root / "docs" / "ROADMAP_WBS.md" roadmap_path.parent.mkdir(parents=True) roadmap_path.write_text("# Roadmap\n\nQUANT_ENGINE_WBS_V1 is the standard.\n", encoding="utf-8") tools_dir = Path(__file__).resolve().parents[2] / "tools" result = subprocess.run( [sys.executable, str(tools_dir / "validate_quant_engine_wbs_v1.py"), "--repo-root", str(repo_root), "--spec", str(spec_path)], cwd=repo_root, capture_output=True, text=True ) assert result.returncode == 1, f"Expected failure, got: {result.stdout}" def test_validator_pending_tasks_no_verdict_required(self, tmp_path): """Test validator passes for PENDING tasks without verdicts.""" repo_root = tmp_path / "repo" repo_root.mkdir() spec_path = repo_root / "spec" / "60_quant_engine_wbs.yaml" spec_path.parent.mkdir() spec = { "meta": {"formula_id": "QUANT_ENGINE_WBS_V1"}, "tasks": { "TEST-09": { "title": "Test task", "status": "PENDING", "depends_on": [], "success_criteria": { "expected_success_value": {}, "evidence_artifacts": [], "verification_commands": [] }, "evidence_checks": [ {"type": "file_exists", "paths": ["spec/60_quant_engine_wbs.yaml"]} ] } } } spec_path.write_text(yaml.dump(spec), encoding="utf-8") # Create roadmap pointer roadmap_path = repo_root / "docs" / "ROADMAP_WBS.md" roadmap_path.parent.mkdir(parents=True) roadmap_path.write_text("# Roadmap\n\nQUANT_ENGINE_WBS_V1 is the standard.\n", encoding="utf-8") tools_dir = Path(__file__).resolve().parents[2] / "tools" result = subprocess.run( [sys.executable, str(tools_dir / "validate_quant_engine_wbs_v1.py"), "--repo-root", str(repo_root), "--spec", str(spec_path)], cwd=repo_root, capture_output=True, text=True ) assert result.returncode == 0, result.stderr if __name__ == "__main__": pytest.main([__file__, "-v"])