Files
rpa_vision_v3/tests/unit/test_competence_validator.py

997 lines
42 KiB
Python

from __future__ import annotations
import json
import sys
from pathlib import Path
import yaml
ROOT = Path(__file__).parent.parent.parent
sys.path.insert(0, str(ROOT))
from tools.competence_validator import validate_competence_file, validate_primitive_file
from tools import competence_validator
P0_COMPETENCE = ROOT / "data/competences/candidate/open_windows_search.yaml"
P1_SEARCH_COMPETENCE = ROOT / "data/competences/observed/saisir_requete_recherche.yaml"
P2_WORD_COMPETENCE = ROOT / "data/competences/candidate/saisir_texte_word.yaml"
P3_RUN_COMPETENCE = ROOT / "data/competences/observed/open_application_via_run.yaml"
P3_SCROLL_COMPETENCE = ROOT / "data/competences/observed/scroll_down_pdf_edge.yaml"
P4_CLICK_SEARCH_COMPETENCE = ROOT / "data/competences/candidate/open_windows_search_taskbar_click.yaml"
KEY_COMBO_PRIMITIVE = ROOT / "data/primitives/key_combo.yaml"
TEXT_INPUT_FOCUSED_PRIMITIVE = ROOT / "data/primitives/text_input_focused.yaml"
SCROLL_VIEW_PRIMITIVE = ROOT / "data/primitives/scroll_view.yaml"
CLICK_ANCHOR_PRIMITIVE = ROOT / "data/primitives/click_anchor.yaml"
WAIT_FOR_STATE_PRIMITIVE = ROOT / "data/primitives/wait_for_state.yaml"
def _issue_codes(path: Path) -> set[str]:
return {issue.code for issue in validate_competence_file(path, repo_root=ROOT).issues}
def _sequence_competence_data() -> dict:
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["methods_execution"] = "sequence"
data["chain_refs"]["cleaned_segment"]["keep_event_indices"] = [3, 5, 6, 7, 8, 9, 10, 11, 12, 13]
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [3, 5, 6, 8, 9, 10, 12]
data["chain_refs"]["cleaned_segment"]["success_event_indices"] = [13]
data["methods"] = [
{
"id": "step_1_open_search",
"kind": "key_combo",
"primitive_ref": "key_combo",
"parameters": {"keys": ["win", "s"]},
"keys": ["win", "s"],
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [3],
},
{
"id": "step_2_type_query",
"kind": "text_input",
"primitive_ref": "text_input_focused",
"parameters": {
"text": "test lea apprentissage",
"concat_rule": "concat_in_order",
},
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [5, 6, 8, 9, 10, 12],
"reconstructed_text": "test lea apprentissage",
},
]
return data
def _write_nested_session(path: Path, events: list[dict]) -> None:
path.write_text(
json.dumps(
{
"session_id": "sess_nested",
"events": [
{
"session_id": "sess_nested",
"timestamp": float(index),
"event": event,
}
for index, event in enumerate(events)
],
}
),
encoding="utf-8",
)
def _scroll_competence_data(tmp_path: Path, events: list[dict]) -> dict:
session_path = tmp_path / "nested_scroll_session.json"
live_events_path = tmp_path / "live_events.jsonl"
_write_nested_session(session_path, events)
live_events_path.write_text("", encoding="utf-8")
return {
"schema_version": 1,
"id": "scroll_test",
"name": "Scroll test",
"version": 1,
"learning_state": "observed",
"intent": {"fr": "tester un scroll"},
"parameters": {},
"preconditions": [{"id": "app_active", "kind": "active_window", "any_of": [{"process_active": "msedge.exe"}]}],
"methods": [
{
"id": "scroll_down",
"kind": "scroll",
"primitive_ref": "scroll_view",
"parameters": {"direction": "down", "amount": 3, "unit": "lines"},
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [1],
}
],
"success_marker": {
"mode": "all_of",
"timeout_ms": 5000,
"markers": [{"kind": "active_process_name_is", "value": "msedge.exe"}],
},
"failure_message_template": {
"intention": "tester un scroll",
"attendu": "la fenetre doit rester active apres le scroll",
"vu": "{observed_human_state}",
"demande": "indiquer si la fenetre active peut defiler vers le bas",
},
"chain_refs": {
"source_session": "sess_nested",
"machine_id": "DESKTOP-58D5CAC_windows",
"streaming_session_path": str(session_path),
"live_events_path": str(live_events_path),
"cleaned_segment": {
"status": "documented_offline",
"keep_event_indices": [0, 1, 2],
"method_event_indices": [1],
"success_event_indices": [2],
"excluded_event_indices": [],
"stop_before_event_index": 3,
"stop_before": ["end_of_synthetic_scroll_trace"],
},
},
"promotion": {
"candidate_requires": ["cleaned_segment_validated"],
"supervised_requires": ["replay_verified_once"],
"stable_requires": {"min_successes": 3, "distinct_contexts": 3, "max_unexplained_failures": 0},
},
"generalisation": {"seen_contexts": [], "method_success_rate": {}, "variance_log": []},
"failure_log": [],
"created_at": "2026-05-28T13:45:00+02:00",
"last_updated_at": "2026-05-28T13:45:00+02:00",
}
def _click_competence_data(tmp_path: Path, events: list[dict]) -> dict:
session_path = tmp_path / "nested_click_session.json"
live_events_path = tmp_path / "live_events.jsonl"
_write_nested_session(session_path, events)
live_events_path.write_text("", encoding="utf-8")
return {
"schema_version": 1,
"id": "click_test",
"name": "Click test",
"version": 1,
"learning_state": "observed",
"intent": {"fr": "tester un clic sur ancre"},
"parameters": {},
"preconditions": [{"id": "desktop_active", "kind": "active_window", "any_of": [{"process_active": "explorer.exe"}]}],
"methods": [
{
"id": "click_search",
"kind": "click",
"primitive_ref": "click_anchor",
"parameters": {
"anchor_ref": "windows_search_button",
"button": "left",
"click_count": 1,
"relative_offset": {"x_pct": 0.5, "y_pct": 0.5},
},
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [1],
}
],
"success_marker": {
"mode": "all_of",
"timeout_ms": 5000,
"markers": [{"kind": "active_process_name_is", "value": "SearchHost.exe"}],
},
"failure_message_template": {
"intention": "cliquer sur le bouton de recherche",
"attendu": "la fenetre rechercher doit s'ouvrir",
"vu": "{observed_human_state}",
"demande": "me montrer le bouton rechercher dans la barre des taches",
},
"chain_refs": {
"source_session": "sess_nested",
"machine_id": "windows_vm",
"streaming_session_path": str(session_path),
"live_events_path": str(live_events_path),
"cleaned_segment": {
"status": "documented_offline",
"keep_event_indices": [0, 1, 2],
"method_event_indices": [1],
"success_event_indices": [2],
"excluded_event_indices": [],
"stop_before_event_index": 3,
"stop_before": ["end_of_synthetic_click_trace"],
},
},
"promotion": {
"candidate_requires": ["cleaned_segment_validated"],
"supervised_requires": ["replay_verified_once"],
"stable_requires": {"min_successes": 3, "distinct_contexts": 3, "max_unexplained_failures": 0},
"t2_known_gaps": [
{
"id": "click_target_semantics_not_observed_offline",
"description": "la trace prouve le clic mais pas l'ancre semantique sans OCR offline",
"impact": "candidate requiert replay ou validation humaine de l'ancre",
"proposed_resolution": "ajouter preuve OCR ou screenshot diff au replay supervise",
}
],
},
"generalisation": {"seen_contexts": [], "method_success_rate": {}, "variance_log": []},
"failure_log": [],
"created_at": "2026-05-28T15:35:00+02:00",
"last_updated_at": "2026-05-28T15:35:00+02:00",
}
def _wait_state_competence_data(tmp_path: Path, events: list[dict]) -> dict:
session_path = tmp_path / "nested_wait_state_session.json"
live_events_path = tmp_path / "live_events.jsonl"
_write_nested_session(session_path, events)
live_events_path.write_text("", encoding="utf-8")
return {
"schema_version": 1,
"id": "wait_state_test",
"name": "Wait state test",
"version": 1,
"learning_state": "observed",
"intent": {"fr": "tester une attente d'etat"},
"parameters": {},
"preconditions": [{"id": "desktop_active", "kind": "active_window", "any_of": [{"process_active": "explorer.exe"}]}],
"methods": [
{
"id": "wait_search_visible",
"kind": "wait_state",
"primitive_ref": "wait_for_state",
"parameters": {
"expected_state": {
"window_title_in": ["Rechercher"],
"process_active": "SearchHost.exe",
},
"timeout_ms": 3000,
"poll_interval_ms": 250,
"evidence_required": "window_or_process",
},
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [1],
}
],
"success_marker": {
"mode": "all_of",
"timeout_ms": 5000,
"markers": [
{"kind": "active_window_title_in", "values": ["Rechercher"]},
{"kind": "active_process_name_is", "value": "SearchHost.exe"},
],
},
"failure_message_template": {
"intention": "attendre l'apparition de la recherche Windows",
"attendu": "la fenetre rechercher doit etre visible",
"vu": "{observed_human_state}",
"demande": "me montrer la fenetre rechercher ou son libelle visible",
},
"chain_refs": {
"source_session": "sess_nested",
"machine_id": "windows_vm",
"streaming_session_path": str(session_path),
"live_events_path": str(live_events_path),
"cleaned_segment": {
"status": "documented_offline",
"keep_event_indices": [0, 1, 2],
"method_event_indices": [1],
"success_event_indices": [2],
"excluded_event_indices": [],
"stop_before_event_index": 3,
"stop_before": ["end_of_synthetic_wait_state_trace"],
},
},
"promotion": {
"candidate_requires": ["cleaned_segment_validated"],
"supervised_requires": ["replay_verified_once"],
"stable_requires": {"min_successes": 3, "distinct_contexts": 3, "max_unexplained_failures": 0},
},
"generalisation": {"seen_contexts": [], "method_success_rate": {}, "variance_log": []},
"failure_log": [],
"created_at": "2026-05-28T16:35:00+02:00",
"last_updated_at": "2026-05-28T16:35:00+02:00",
}
def test_validator_imports_message_contract():
assert competence_validator.format_supervised_pause_message is not None, (
"message_contract introuvable: le validateur ignorerait silencieusement "
"failure_message_template"
)
def test_open_windows_search_candidate_validates_against_source_trace():
report = validate_competence_file(P0_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_saisir_requete_recherche_competence_validates_against_source_trace():
report = validate_competence_file(P1_SEARCH_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_saisir_texte_word_competence_validates_against_source_trace():
report = validate_competence_file(P2_WORD_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_open_application_via_run_competence_validates_against_source_trace():
report = validate_competence_file(P3_RUN_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_scroll_down_pdf_edge_competence_validates_against_source_trace():
report = validate_competence_file(P3_SCROLL_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_open_windows_search_taskbar_click_validates_against_source_trace():
report = validate_competence_file(P4_CLICK_SEARCH_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_handles_nested_event_format(tmp_path):
session_path = tmp_path / "nested_session.json"
live_events_path = tmp_path / "live_events.jsonl"
session_path.write_text(
json.dumps(
{
"session_id": "sess_nested",
"events": [
{
"session_id": "sess_nested",
"timestamp": 1.0,
"event": {
"type": "key_combo",
"keys": ["win", "s"],
"window": {"title": "Desktop", "app_name": "explorer.exe"},
},
},
{
"session_id": "sess_nested",
"timestamp": 2.0,
"event": {
"type": "window_focus_change",
"to": {"title": "Rechercher", "app_name": "SearchHost.exe"},
},
},
],
}
),
encoding="utf-8",
)
live_events_path.write_text("", encoding="utf-8")
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["source_session"] = "sess_nested"
data["chain_refs"]["streaming_session_path"] = str(session_path)
data["chain_refs"]["live_events_path"] = str(live_events_path)
data["chain_refs"]["cleaned_segment"]["keep_event_indices"] = [0, 1]
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [0]
data["chain_refs"]["cleaned_segment"]["success_event_indices"] = [1]
data["chain_refs"]["cleaned_segment"]["excluded_event_indices"] = []
data["chain_refs"]["cleaned_segment"]["stop_before_event_index"] = 2
data["chain_refs"]["cleaned_segment"]["stop_before"] = ["end_of_synthetic_nested_trace"]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_saisir_texte_word_documents_t2_known_gap():
data = yaml.safe_load(P2_WORD_COMPETENCE.read_text(encoding="utf-8"))
gaps = data["promotion"]["t2_known_gaps"]
assert gaps[0]["id"] == "marker_continuation_human"
assert "success_event #40" in gaps[0]["description"]
assert gaps[0]["proposed_resolution"]
def test_bootstrap_primitives_validate():
for path in (
KEY_COMBO_PRIMITIVE,
TEXT_INPUT_FOCUSED_PRIMITIVE,
SCROLL_VIEW_PRIMITIVE,
CLICK_ANCHOR_PRIMITIVE,
WAIT_FOR_STATE_PRIMITIVE,
):
report = validate_primitive_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_primitive_click_anchor_validates():
report = validate_primitive_file(CLICK_ANCHOR_PRIMITIVE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_primitive_wait_for_state_validates():
report = validate_primitive_file(WAIT_FOR_STATE_PRIMITIVE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_existing_competences_reference_bootstrap_primitives():
p0 = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
p1 = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
assert p0["methods"][0]["primitive_ref"] == "key_combo"
assert p0["methods"][0]["parameters"]["keys"] == ["win", "s"]
assert p1["methods"][0]["primitive_ref"] == "text_input_focused"
assert p1["methods"][0]["parameters"]["text"] == "test lea apprentissage"
def test_observed_dependency_accepts_promoted_candidate():
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
assert data["preconditions"][0]["state"] == "observed"
assert validate_competence_file(P1_SEARCH_COMPETENCE, repo_root=ROOT).valid
def test_validator_rejects_missing_observed_key_combo_in_cleaned_segment(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["keys"] = ["ctrl", "k"]
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_id_filename_mismatch(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
path = tmp_path / "wrong_filename.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "id_filename_mismatch" in _issue_codes(path)
def test_validator_full_competence_corpus():
competence_paths = sorted((ROOT / "data/competences").glob("*/*.yaml"))
primitive_paths = sorted((ROOT / "data/primitives").glob("*.yaml"))
assert competence_paths, "no competence YAML found"
assert primitive_paths, "no primitive YAML found"
failures = {
str(path.relative_to(ROOT)): [
f"{issue.code}: {issue.detail}"
for issue in validate_competence_file(path, repo_root=ROOT).issues
]
for path in competence_paths
}
failures.update(
{
str(path.relative_to(ROOT)): [
f"{issue.code}: {issue.detail}"
for issue in validate_primitive_file(path, repo_root=ROOT).issues
]
for path in primitive_paths
}
)
failures = {path: issues for path, issues in failures.items() if issues}
assert failures == {}
def test_validator_rejects_primitive_forbidden_field(tmp_path):
data = yaml.safe_load(KEY_COMBO_PRIMITIVE.read_text(encoding="utf-8"))
data["learning_state"] = "observed"
path = tmp_path / "key_combo.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
issue_codes = {issue.code for issue in validate_primitive_file(path, repo_root=ROOT).issues}
assert "primitive_forbidden_field" in issue_codes
def test_validator_rejects_primitive_empty_enum(tmp_path):
data = yaml.safe_load(SCROLL_VIEW_PRIMITIVE.read_text(encoding="utf-8"))
data["parameters_schema"]["direction"]["constraints"]["enum"] = []
path = tmp_path / "scroll_view.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
issue_codes = {issue.code for issue in validate_primitive_file(path, repo_root=ROOT).issues}
assert "primitive_schema_invalid" in issue_codes
def test_primitive_click_anchor_rejects_pos_in_parameters(tmp_path):
data = yaml.safe_load(CLICK_ANCHOR_PRIMITIVE.read_text(encoding="utf-8"))
data["parameters_schema"]["pos"] = {
"type": "list[str]",
"required": False,
"description": "coordonnees a refuser",
}
path = tmp_path / "click_anchor.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
issue_codes = {issue.code for issue in validate_primitive_file(path, repo_root=ROOT).issues}
assert "durable_coordinate_key" in issue_codes
def test_primitive_click_count_out_of_range_rejected(tmp_path):
data = _click_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["click_count"] = 3
path = tmp_path / "click_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_click_count_out_of_range" in _issue_codes(path)
def test_primitive_relative_offset_pct_out_of_range_rejected(tmp_path):
data = _click_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["relative_offset"] = {"x_pct": 1.5, "y_pct": 0.5}
path = tmp_path / "click_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
issue_codes = _issue_codes(path)
assert "primitive_relative_offset_invalid" in issue_codes
assert "durable_coordinate_key" not in issue_codes
def test_validator_click_method_requires_mouse_click_events(tmp_path):
data = _click_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "key_combo", "keys": ["win", "s"], "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
path = tmp_path / "click_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_click_method_with_valid_mouse_click_passes(tmp_path):
data = _click_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
path = tmp_path / "click_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_wait_for_state_method_with_window_focus_change_passes(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{
"type": "window_focus_change",
"to": {"title": "Rechercher", "app_name": "SearchHost.exe"},
},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_wait_for_state_expected_state_required(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"].pop("expected_state")
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_expected_state_invalid" in _issue_codes(path)
def test_wait_for_state_expected_state_must_be_non_empty_dict(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["expected_state"] = {}
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_expected_state_invalid" in _issue_codes(path)
def test_wait_for_state_timeout_out_of_range_rejected(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["timeout_ms"] = 50
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_wait_timeout_invalid" in _issue_codes(path)
def test_wait_for_state_poll_interval_out_of_range_rejected(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["poll_interval_ms"] = 10000
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_poll_interval_invalid" in _issue_codes(path)
def test_wait_for_state_evidence_required_enum_validated(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["evidence_required"] = "foo"
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_schema_invalid" in _issue_codes(path)
def test_wait_for_state_method_rejects_human_continuation_event(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "text_input", "text": "test", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_bad_t2_known_gaps_type(tmp_path):
data = yaml.safe_load(P2_WORD_COMPETENCE.read_text(encoding="utf-8"))
data["promotion"]["t2_known_gaps"] = "marker_continuation_human"
path = tmp_path / "saisir_texte_word.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "t2_known_gap_invalid" in _issue_codes(path)
def test_validator_rejects_t2_known_gap_missing_required_field(tmp_path):
data = yaml.safe_load(P2_WORD_COMPETENCE.read_text(encoding="utf-8"))
data["promotion"]["t2_known_gaps"] = [
{
"id": "marker_continuation_human",
"description": "success_event #40 est un text_input humain post-methode.",
"proposed_resolution": "Ajouter wait_state ou OCR runtime.",
}
]
path = tmp_path / "saisir_texte_word.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "t2_known_gap_invalid" in _issue_codes(path)
def test_validator_accepts_methods_execution_sequence_with_step_trace_indices(tmp_path):
data = _sequence_competence_data()
path = tmp_path / "saisir_requete_recherche.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_rejects_invalid_methods_execution_mode(tmp_path):
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["methods_execution"] = "serial"
path = tmp_path / "saisir_requete_recherche.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "methods_sequence_invalid" in _issue_codes(path)
def test_validator_rejects_sequence_without_two_methods(tmp_path):
data = _sequence_competence_data()
data["methods"] = data["methods"][:1]
path = tmp_path / "saisir_requete_recherche.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "methods_sequence_invalid" in _issue_codes(path)
def test_validator_rejects_sequence_observed_step_without_trace_indices(tmp_path):
data = _sequence_competence_data()
data["methods"][1].pop("trace_event_indices")
path = tmp_path / "saisir_requete_recherche.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_accepts_trace_event_indices_in_alternatives_mode(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["trace_event_indices"] = [3]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_rejects_trace_event_indices_outside_keep_indices_in_alternatives_mode(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["trace_event_indices"] = [5]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_trace_event_indices_outside_method_indices_in_alternatives_mode(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["cleaned_segment"]["keep_event_indices"] = [0, 1, 2, 3, 4, 7]
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [3]
data["methods"][0]["trace_event_indices"] = [4]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_alternatives_trace_event_indices_have_no_order_constraint(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [3, 7]
data["methods"][0]["trace_event_indices"] = [7]
data["methods"][1]["observed"] = True
data["methods"][1]["trace_source"] = "live_events.jsonl"
data["methods"][1]["trace_event_indices"] = [3]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "methods_sequence_invalid" not in _issue_codes(path)
def test_validator_accepts_scroll_method_with_trace_event_indices(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_scroll", "delta": [0, -1], "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_accepts_scroll_method_with_method_indices_fallback(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_scroll", "delta": [0, -1], "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
data["methods"][0].pop("trace_event_indices")
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_rejects_scroll_method_with_non_scroll_events(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_click", "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_scroll_method_without_delta(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_scroll", "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_scroll_delta_missing" in _issue_codes(path)
def test_validator_rejects_scroll_method_direction_mismatch(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_scroll", "delta": [0, 1], "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_scroll_direction_mismatch" in _issue_codes(path)
def test_validator_rejects_unknown_primitive_ref(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["primitive_ref"] = "missing_primitive"
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_ref_unknown" in _issue_codes(path)
def test_validator_rejects_primitive_kind_mismatch(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["primitive_ref"] = "text_input_focused"
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_kind_mismatch" in _issue_codes(path)
def test_validator_rejects_missing_primitive_parameter(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["parameters"] = {}
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_schema_invalid" in _issue_codes(path)
def test_validator_rejects_missing_scroll_direction_parameter(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["kind"] = "scroll"
data["methods"][0]["primitive_ref"] = "scroll_view"
data["methods"][0]["parameters"] = {"amount": 3, "unit": "lines"}
data["methods"][0].pop("keys", None)
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_schema_invalid" in _issue_codes(path)
def test_validator_rejects_durable_coordinates(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["success_marker"]["coordinates"] = {"x": 120, "y": 340}
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "durable_coordinate_key" in _issue_codes(path)
def test_validator_rejects_bad_failure_message_contract(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["failure_message_template"]["vu"] = "target_not_found score=0.87"
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "failure_message_contract" in _issue_codes(path)
def test_validator_rejects_success_marker_before_method(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["cleaned_segment"]["keep_event_indices"] = [0, 1, 2, 3, 4]
data["chain_refs"]["cleaned_segment"]["success_event_indices"] = [2]
data["chain_refs"]["cleaned_segment"]["stop_before_event_index"] = 5
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "success_marker_pre_method" in _issue_codes(path)
def test_validator_rejects_stable_state_without_3_contexts(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["learning_state"] = "stable"
data["generalisation"]["seen_contexts"] = [
{"dpi": 150, "screen": "2560x1600", "method_used": "keyboard_win_s"},
{"dpi": 150, "screen": "2560x1600", "method_used": "keyboard_win_s"},
]
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "learning_state_premature" in _issue_codes(path)
def test_validator_rejects_text_input_reconstruction_mismatch(tmp_path):
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["reconstructed_text"] = "test lea"
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_reconstructed_text_mismatch" in _issue_codes(path)
def test_validator_rejects_text_input_method_indices_with_heartbeat(tmp_path):
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [5, 6, 7]
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_missing_competence_dependency(tmp_path):
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["preconditions"][0]["competence"] = "missing_competence"
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "competence_dependency_missing" in _issue_codes(path)