499 lines
19 KiB
Python
499 lines
19 KiB
Python
"""
|
|
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"])
|