Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(lea): add dashboard competence promotion dry run

Browse Source
This commit is contained in:
Dom
2026-05-29 21:48:00 +02:00
parent bd3aaf7d64
commit 34527b5cc5
8 changed files with 1341 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
tests/unit/test_competence_promotions.py Normal file
View File

@@ -0,0 +1,213 @@
from datetime import datetime, timezone
from uuid import uuid4
import pytest
from core.competences.promotions import (
CompetencePromotionError,
iter_competence_promotions,
promote_competence_from_verdicts,
summarize_competence_promotions,
)
from core.competences.verdicts import store_competence_verdict
def _write_competence(root, state="observed", competence_id="demo_competence"):
state_dir = root / state
state_dir.mkdir(parents=True, exist_ok=True)
path = state_dir / f"{competence_id}.yaml"
path.write_text(
"\n".join([
"schema_version: 1",
f"id: {competence_id}",
"name: Demo competence",
f"learning_state: {state}",
"intent:",
" fr: demo competence",
"methods: []",
"promotion:",
" candidate_requires:",
" - cleaned_segment_validated",
" stable_requires:",
" min_successes: 3",
" distinct_contexts: 3",
" max_unexplained_failures: 0",
"generalisation:",
" seen_contexts: []",
"failure_log: []",
"created_at: '2026-05-29T00:00:00+00:00'",
"last_updated_at: '2026-05-29T00:00:00+00:00'",
]),
encoding="utf-8",
)
return path
def _valid_verdict(root, log_path, competence_id="demo_competence", machine="machine_a"):
verdict_id = str(uuid4())
return store_competence_verdict(
competence_id,
{
"verdict_id": verdict_id,
"verdict_kind": "valid",
"verdict_by": "human:dom",
"workflow_id": f"workflow_{machine}",
"step_results": [
{"step_id": "step_1", "action_type": "keyboard_shortcut", "status": "success"}
],
"context_signature": {
"machine_id": machine,
"screen_state_initial": f"before_{machine}",
"screen_state_after_action": f"after_{machine}",
},
},
competence_root=root,
log_path=log_path,
now=datetime(2026, 5, 29, 17, 0, tzinfo=timezone.utc),
)
def test_promote_competence_dry_run_does_not_write_yaml(tmp_path):
root = tmp_path / "competences"
source = _write_competence(root, state="observed")
verdict_log = tmp_path / "verdicts.jsonl"
verdict = _valid_verdict(root, verdict_log)
before = source.read_text(encoding="utf-8")
result = promote_competence_from_verdicts(
"demo_competence",
{
"promotion_id": str(uuid4()),
"target_state": "candidate",
"verdict_ids": [verdict["verdict_id"]],
"confirmed_by": "human:dom",
"dry_run": True,
},
competence_root=root,
verdict_log_path=verdict_log,
promotion_log_path=tmp_path / "promotions.jsonl",
now=datetime(2026, 5, 29, 18, 0, tzinfo=timezone.utc),
)
assert result["dry_run"] is True
assert result["eligible"] is True
assert result["write_applied"] is False
assert "learning_state: candidate" in result["yaml_diff"]
assert source.read_text(encoding="utf-8") == before
def test_promote_competence_confirm_writes_backup_and_audit(tmp_path):
root = tmp_path / "competences"
source = _write_competence(root, state="observed")
verdict_log = tmp_path / "verdicts.jsonl"
promotion_log = tmp_path / "promotions.jsonl"
verdict = _valid_verdict(root, verdict_log)
promotion_id = str(uuid4())
dry_run = promote_competence_from_verdicts(
"demo_competence",
{
"promotion_id": promotion_id,
"target_state": "candidate",
"verdict_ids": [verdict["verdict_id"]],
"confirmed_by": "human:dom",
"dry_run": True,
},
competence_root=root,
verdict_log_path=verdict_log,
promotion_log_path=promotion_log,
now=datetime(2026, 5, 29, 18, 0, tzinfo=timezone.utc),
)
result = promote_competence_from_verdicts(
"demo_competence",
{
"promotion_id": promotion_id,
"target_state": "candidate",
"verdict_ids": [verdict["verdict_id"]],
"confirmed_by": "human:dom",
"dry_run": False,
"dry_run_token": dry_run["dry_run_token"],
},
competence_root=root,
verdict_log_path=verdict_log,
promotion_log_path=promotion_log,
now=datetime(2026, 5, 29, 18, 0, tzinfo=timezone.utc),
)
target = root / "candidate" / "demo_competence.yaml"
assert result["write_applied"] is True
assert target.exists()
assert not source.exists()
written = target.read_text(encoding="utf-8")
assert "learning_state: candidate" in written
assert verdict["verdict_id"] in written
assert list(source.parent.glob("demo_competence.yaml.*.bak"))
promotions = iter_competence_promotions(log_path=promotion_log)
assert promotions[0]["promotion_id"] == promotion_id
assert promotions[0]["yaml_write"] is True
def test_promote_competence_requires_prior_dry_run_token(tmp_path):
root = tmp_path / "competences"
_write_competence(root, state="observed")
verdict_log = tmp_path / "verdicts.jsonl"
verdict = _valid_verdict(root, verdict_log)
with pytest.raises(CompetencePromotionError, match="dry_run_token"):
promote_competence_from_verdicts(
"demo_competence",
{
"promotion_id": str(uuid4()),
"target_state": "candidate",
"verdict_ids": [verdict["verdict_id"]],
"confirmed_by": "human:dom",
"dry_run": False,
},
competence_root=root,
verdict_log_path=verdict_log,
promotion_log_path=tmp_path / "promotions.jsonl",
)
def test_stable_promotion_requires_three_distinct_contexts(tmp_path):
root = tmp_path / "competences"
_write_competence(root, state="candidate")
verdict_log = tmp_path / "verdicts.jsonl"
verdict = _valid_verdict(root, verdict_log, machine="machine_a")
result = promote_competence_from_verdicts(
"demo_competence",
{
"promotion_id": str(uuid4()),
"target_state": "stable",
"verdict_ids": [verdict["verdict_id"]],
"confirmed_by": "human:dom",
"dry_run": True,
},
competence_root=root,
verdict_log_path=verdict_log,
promotion_log_path=tmp_path / "promotions.jsonl",
)
assert result["eligible"] is False
assert any("3 verdicts valid requis" in reason for reason in result["blocking_reasons"])
assert any("3 contextes distincts requis" in reason for reason in result["blocking_reasons"])
def test_summarize_competence_promotions_reports_eligible_candidate(tmp_path):
root = tmp_path / "competences"
_write_competence(root, state="observed")
verdict_log = tmp_path / "verdicts.jsonl"
verdict = _valid_verdict(root, verdict_log)
summary = summarize_competence_promotions(
competence_root=root,
verdict_log_path=verdict_log,
)
assert summary[0]["id"] == "demo_competence"
assert summary[0]["verdict_counts"]["valid"] == 1
candidate = summary[0]["eligible_targets"]["candidate"]
assert candidate["eligible"] is True
assert candidate["recommended_verdict_ids"] == [verdict["verdict_id"]]

8
tests/unit/test_dashboard_routes.py
View File

@@ -54,6 +54,14 @@ class TestDashboardRoutes:
data = resp.get_json()
assert 'faiss' in data
def test_knowledge_base_stats_include_competences(self, client):
"""La base de connaissances expose les competences supervisees."""
resp = client.get('/api/knowledge-base/stats')
assert resp.status_code == 200
data = resp.get_json()
assert 'competences' in data
assert 'items' in data['competences']
def test_version(self, client):
"""L'API version retourne la version actuelle."""
resp = client.get('/api/version')

39
tests/unit/test_lea_competence_verdict_api.py
View File

@@ -107,3 +107,42 @@ def test_list_competence_verdicts_endpoint(monkeypatch):
assert data["verdicts"] == [
{"competence_id": "key_win_r_wait_explorer_exe", "verdict_kind": "valid"}
]
def test_promote_competence_endpoint_dry_run(monkeypatch):
def fake_promote(competence_id, payload):
assert competence_id == "key_win_r_wait_explorer_exe"
assert payload["dry_run"] is True
return {
"promotion_id": payload["promotion_id"],
"competence_id": competence_id,
"target_state": "candidate",
"dry_run": True,
"eligible": True,
"dry_run_token": "token",
"write_applied": False,
}
monkeypatch.setattr(
lea_competences_module,
"promote_competence_from_verdicts",
fake_promote,
)
with _app().test_client() as client:
response = client.post(
"/api/v1/lea/competences/key_win_r_wait_explorer_exe/promote",
json={
"promotion_id": "123e4567-e89b-42d3-a456-426614174000",
"target_state": "candidate",
"verdict_ids": ["123e4567-e89b-42d3-a456-426614174001"],
"confirmed_by": "human:dom",
"dry_run": True,
},
)
data = response.get_json()
assert response.status_code == 200
assert data["success"] is True
assert data["dry_run"] is True
assert data["yaml_write"] is False