diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 4aa2469..beff75b 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -57,4 +57,6 @@ jobs: tests/test_model.py \ tests/test_export.py \ tests/test_inference.py \ - tests/test_model_rules.py + tests/test_model_rules.py \ + tests/test_review.py \ + tests/test_model_cli.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index b8c0418..5b9e33d 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -11,3 +11,4 @@ v0.0.0 - Add composite rule required phrase updates. - Add required phrase model training and ONNX export. - Add model prediction and rule integration. +- Add resumable review for model-predicted required phrases. diff --git a/README.rst b/README.rst index 81683ae..72ba740 100644 --- a/README.rst +++ b/README.rst @@ -70,14 +70,34 @@ return candidate required phrases without changing a ScanCode rule or file: Predictions require human review before they are added to license rules. -Prepare predicted phrases for review -==================================== - -The ``model_rules`` module loads eligible rules, validates model predictions, -and prepares complete rule updates without changing the original rules. It also -provides an atomic writer that requires the exact rule path and its current file -hash. User-facing review and application are added by the stacked review -workflow. +Review model-predicted phrases +============================== + +Install the inference dependencies and run the review command. It uses the +pinned public model by default: + +.. code-block:: console + + python -m pip install ".[inference]" + add-model-required-phrases --rule path/to/example.RULE + +The default mode shows each phrase, model score, text context, and exact rule +diff before asking for approval. Decisions are saved in a resumable session. +Use ``--rules-dir`` for top-level rule files in a directory or ``--all`` for +eligible installed ScanCode rules. + +Read-only prediction never creates a session or changes a rule: + +.. code-block:: console + + add-model-required-phrases --rule path/to/example.RULE --predict-only + +A wrong required phrase can cause a false negative. Batch processing therefore +requires explicit score thresholds and ``--yes`` before any write. Rules with +pending phrases are deferred unchanged while fully decided rules can be applied. +``--dry-run`` always writes zero rules. Run ``scancode-reindex-licenses`` after +changing installed ScanCode rules. Use ``--model`` for a custom local or remote +model. Development =========== diff --git a/azure-pipelines.yml b/azure-pipelines.yml index 3be6213..6086c06 100644 --- a/azure-pipelines.yml +++ b/azure-pipelines.yml @@ -29,7 +29,9 @@ jobs: tests/test_model.py \ tests/test_export.py \ tests/test_inference.py \ - tests/test_model_rules.py + tests/test_model_rules.py \ + tests/test_review.py \ + tests/test_model_cli.py displayName: Run training unit tests - template: etc/ci/azure-posix.yml diff --git a/docs/source/model_rules.rst b/docs/source/model_rules.rst index 904be6c..017ce2b 100644 --- a/docs/source/model_rules.rst +++ b/docs/source/model_rules.rst @@ -1,19 +1,89 @@ -Model prediction and rule preparation -===================================== +Review model-predicted required phrases +======================================= -Install the ``inference`` extra before loading a final model. A local model must -pass the final-model publication checks. A remote Hugging Face model also -requires its full commit hash. +Install the inference dependencies before using the command: -The ``scancode_required_phrases.model_rules`` module provides reusable functions -to: +.. code-block:: console -- load eligible rules from one file, a directory, or installed ScanCode data; -- return model predictions with their ScanCode validation result; -- prepare a complete rule update without mutating the original rule; -- serialize and atomically write a rule to its exact source path. + python -m pip install ".[inference]" -Phrase text found more than once is rejected because ScanCode's mutation helper -would mark every occurrence. A complete phrase set is prepared before any file -is written. The stacked review workflow provides the user-facing command and -human approval process. +The public model is available without a Hugging Face token: + +.. code-block:: text + + Kaushik-Kumar-CEG/scancode-required-phrases-deberta-bioes-crf-hardened + 11215925b0f9b64cfcfbbb5492b52d6aeb5a572b + +A wrong required phrase can prevent a true license match. Review predictions +carefully and use disposable rule copies before changing installed data. + +Interactive review +------------------ + +Review one rule: + +.. code-block:: console + + add-model-required-phrases --rule path/to/example.RULE + +Use ``--rules-dir`` to review sorted top-level ``.RULE`` files in a directory. +Use ``--all`` to review eligible rules installed with ScanCode Toolkit. The +command shows the rule, expression, predicted phrase, model score, context, and +exact diff. Approve, reject, edit, skip, or save and quit at each prompt. + +Sessions are created automatically and retained for audit. The command prints an +exact resume command when review remains unfinished: + +.. code-block:: console + + add-model-required-phrases --resume path/to/session.jsonl + +Review and application are resumable. An interrupted prediction run starts +prediction again. + +Read-only prediction +-------------------- + +Print validated and rejected model output without creating decisions or writing +rules: + +.. code-block:: console + + add-model-required-phrases --rule path/to/example.RULE --predict-only + +Use ``--json predictions.json`` for machine-readable output or ``--json -`` for +JSON on standard output. + +Batch classification +-------------------- + +Batch mode has no default thresholds. Both values must be chosen explicitly: + +.. code-block:: console + + add-model-required-phrases --all --batch \ + --auto-score 0.90 --review-score 0.70 --dry-run + +Scores at or above ``--auto-score`` are staged for automatic approval only after +ScanCode validation. Scores from ``--review-score`` up to ``--auto-score`` stay +pending. Lower scores are ignored. Truncated rules always require review. A rule +with any pending phrase is deferred unchanged. Other fully decided rules may be +applied after complete preflight. + +``--yes`` permits non-interactive writes for ready rules. ``--dry-run`` always +takes precedence and writes zero rules. Use ``--model`` and ``--model-revision`` +to override the pinned public model. + +Safe application +---------------- + +Review never mutates rule files. Before application, the command verifies every +path and file hash, prepares all approved updates, and displays a final summary. +Each changed rule is then written once with an atomic replacement at its exact +source path. Rejected and below-threshold phrases are not written. + +After changing installed ScanCode rules, run: + +.. code-block:: console + + scancode-reindex-licenses diff --git a/setup.cfg b/setup.cfg index 9d92118..f79bb4c 100644 --- a/setup.cfg +++ b/setup.cfg @@ -55,6 +55,7 @@ where = src [options.entry_points] console_scripts = add-composite-required-phrases = scancode_required_phrases.composite_rules:add_composite_required_phrases + add-model-required-phrases = scancode_required_phrases.model_cli:add_model_required_phrases build-required-phrases-dataset = scancode_required_phrases.dataset:main export-required-phrase-model = scancode_required_phrases.export:main train-required-phrase-model = scancode_required_phrases.training:main diff --git a/src/scancode_required_phrases/model_cli.py b/src/scancode_required_phrases/model_cli.py new file mode 100644 index 0000000..4ff047c --- /dev/null +++ b/src/scancode_required_phrases/model_cli.py @@ -0,0 +1,558 @@ +# Copyright (c) nexB Inc. and others. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Review and add model-predicted required phrases.""" + +import json +import os +from pathlib import Path +import sys +import warnings + +import click + +from licensedcode.models import InvalidRule +from licensedcode.models import rules_data_dir + +from scancode_required_phrases.model_rules import load_prediction_rule +from scancode_required_phrases.model_rules import load_prediction_rules +from scancode_required_phrases.model_rules import load_predictor +from scancode_required_phrases.model_rules import predict_rule_candidates +from scancode_required_phrases.model_rules import select_installed_prediction_rules +from scancode_required_phrases.review import APPROVED +from scancode_required_phrases.review import create_metadata +from scancode_required_phrases.review import create_prediction +from scancode_required_phrases.review import create_rule_record +from scancode_required_phrases.review import create_session_path +from scancode_required_phrases.review import load_session_rules +from scancode_required_phrases.review import prepare_rule_updates +from scancode_required_phrases.review import read_session +from scancode_required_phrases.review import write_rule_updates +from scancode_required_phrases.review import write_session +from scancode_required_phrases.review_ui import print_summary +from scancode_required_phrases.review_ui import resume_command +from scancode_required_phrases.review_ui import review_predictions + + +DEFAULT_MODEL = "Kaushik-Kumar-CEG/scancode-required-phrases-deberta-bioes-crf-hardened" +DEFAULT_MODEL_REVISION = "11215925b0f9b64cfcfbbb5492b52d6aeb5a572b" + + +def stdin_is_tty(): + """Return whether interactive input is available.""" + return sys.stdin.isatty() + + +def validate_options( + rule_path, + rules_directory, + all_rules, + license_expression, + limit, + predict_only, + batch, + resume, + model, + model_revision, + auto_score, + review_score, + yes, + session_path, + json_output, +): + """Validate command options before loading rules or a model.""" + if resume: + conflicts = ( + rule_path, + rules_directory, + all_rules, + license_expression, + limit, + predict_only, + batch, + model, + model_revision, + auto_score, + review_score, + yes, + session_path, + json_output, + ) + if any(value not in (None, False, 0) for value in conflicts): + raise click.UsageError("--resume cannot be used with new-run options") + if not stdin_is_tty(): + raise click.UsageError("--resume requires an interactive terminal") + return + + targets = (rule_path is not None, rules_directory is not None, all_rules) + if sum(targets) != 1: + raise click.UsageError("Use exactly one of --rule, --rules-dir, or --all") + if predict_only and batch: + raise click.UsageError("--predict-only and --batch are mutually exclusive") + if json_output and not predict_only: + raise click.UsageError("--json requires --predict-only") + if session_path and predict_only: + raise click.UsageError("--session cannot be used with --predict-only") + if session_path and (Path(session_path).exists() or Path(session_path).is_symlink()): + raise click.UsageError(f"Session file already exists: {session_path}") + + if batch: + if auto_score is None or review_score is None: + raise click.UsageError("--batch requires --auto-score and --review-score") + if review_score > auto_score: + raise click.UsageError("--review-score cannot exceed --auto-score") + elif auto_score is not None or review_score is not None or yes: + raise click.UsageError("Score thresholds and --yes require --batch") + + if not predict_only and not batch and not stdin_is_tty(): + raise click.UsageError("Interactive review requires an interactive terminal") + + +def resolve_model(model, model_revision): + """Return the requested model or the pinned project default.""" + if not model: + return DEFAULT_MODEL, model_revision or DEFAULT_MODEL_REVISION + if model == DEFAULT_MODEL and not model_revision: + return model, DEFAULT_MODEL_REVISION + return model, model_revision + + +def select_targets(rule_path, rules_directory, all_rules, license_expression, limit): + """Return target metadata and selected rules for a new run.""" + if rule_path: + selected = [load_prediction_rule(rule_path, license_expression)] + target_mode = "rule" + target = selected[0][0] + rules_scanned = 1 + elif rules_directory: + root = Path(rules_directory).resolve(strict=True) + rules_scanned = len(list(root.glob("*.RULE"))) + selected = load_prediction_rules(root, license_expression) + target_mode = "rules_dir" + target = root + else: + selected = select_installed_prediction_rules(license_expression) + target_mode = "all" + target = Path(rules_data_dir).resolve(strict=True) + rules_scanned = len(selected) + + if limit: + selected = selected[:limit] + return target_mode, target, selected, rules_scanned + + +def predict_targets( + selected, + predictor, + batch=False, + auto_score=None, + progress_file=None, +): + """Return session records, output rows, and the truncated rule count.""" + records = [] + rows = [] + truncated_rules = 0 + with click.progressbar( + selected, + label="Predicting rules", + file=progress_file, + ) as progress: + for rule_path, rule in progress: + result, candidates = predict_rule_candidates(rule, predictor) + if result.truncated: + truncated_rules += 1 + predictions = [] + for prediction, issue in candidates: + entry = create_prediction(prediction, issue) + if ( + batch + and issue is None + and not result.truncated + and prediction.score >= auto_score + ): + entry["decision"] = APPROVED + entry["decision_source"] = "score" + predictions.append(entry) + rows.append( + { + "path": str(rule_path), + "identifier": rule.identifier, + "license_expression": rule.license_expression, + "phrase": prediction.text, + "score": prediction.score, + "start_word": prediction.start_word, + "end_word": prediction.end_word, + "truncated": result.truncated, + "validation_issue": issue, + } + ) + if predictions: + records.append( + create_rule_record( + rule_path=rule_path, + rule=rule, + predictions=predictions, + truncated=result.truncated, + ) + ) + return records, rows, truncated_rules + + +def write_json_predictions(rows, json_output): + """Write prediction rows as JSON to a file or stdout.""" + content = json.dumps(rows, ensure_ascii=False, allow_nan=False, indent=2) + if json_output == "-": + click.echo(content) + return + output_path = Path(json_output) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(content + "\n", encoding="utf-8") + + +def print_prediction_rows(rows): + """Print concise read-only prediction output.""" + if not rows: + click.echo("No model predictions found") + return + for row in rows: + status = row["validation_issue"] or "valid" + truncated = " truncated" if row["truncated"] else "" + click.echo( + f"{row['identifier']} {row['license_expression']} " + f"{row['score']:.1%} {status}{truncated}" + ) + click.echo(f" {row['phrase']}") + + +def finish_session(session_path, metadata, records, dry_run, allow_write): + """Preflight a complete session and optionally write its rules.""" + loaded_rules, reconciled = load_session_rules(metadata, records) + if reconciled: + click.echo(f"Recovered interrupted writes: {', '.join(reconciled)}") + write_session(session_path, metadata, records) + + work, unchanged, deferred = prepare_rule_updates(metadata, records, loaded_rules) + write_session(session_path, metadata, records) + print_summary(metadata, records, unchanged=len(unchanged)) + if not work: + click.echo(f"Session saved: {session_path}") + if deferred: + click.echo(f"Resume with: {resume_command(session_path)}") + return + if dry_run: + click.echo("Dry run: no rules were written") + click.echo(f"Session saved: {session_path}") + if deferred: + click.echo(f"Resume with: {resume_command(session_path)}") + return + + if allow_write is None: + choice = click.prompt( + "[d] dry-run [a] apply [s] save and exit", + default="s", + show_default=False, + type=click.Choice(["d", "a", "s"], case_sensitive=False), + ).lower() + if choice != "a": + if choice == "d": + click.echo("Dry run: no rules were written") + click.echo(f"Session saved: {session_path}") + if deferred: + click.echo(f"Resume with: {resume_command(session_path)}") + return + elif not allow_write: + click.echo(f"Session saved: {session_path}") + if deferred: + click.echo(f"Resume with: {resume_command(session_path)}") + return + + write_rule_updates(session_path, metadata, records, work) + print_summary(metadata, records, unchanged=len(unchanged)) + if deferred: + click.echo(f"Resume with: {resume_command(session_path)}") + if metadata["target_mode"] == "all": + click.echo("Run scancode-reindex-licenses to use the new required phrases") + + +def run_new( + rule_path, + rules_directory, + all_rules, + license_expression, + limit, + predict_only, + batch, + model, + model_revision, + auto_score, + review_score, + yes, + session_path, + json_output, + dry_run, + verbose, + color, +): + """Run prediction and the selected new-run mode.""" + progress_file = ( + click.get_text_stream("stderr") + if json_output == "-" + else click.get_text_stream("stdout") + ) + click.echo("Selecting rules...", file=progress_file) + target_mode, target, selected, rules_scanned = select_targets( + rule_path, + rules_directory, + all_rules, + license_expression, + limit, + ) + if not selected: + if predict_only and json_output: + write_json_predictions([], json_output) + click.echo("No eligible rules found", err=json_output == "-") + return + + click.echo(f"Selected {len(selected)} eligible rules.", file=progress_file) + click.echo("Loading model...", file=progress_file) + with warnings.catch_warnings(): + warnings.filterwarnings( + "ignore", + message=r"`torch\.jit\.script` is deprecated\..*", + category=FutureWarning, + module=r"torch\.jit\._script", + ) + predictor = load_predictor( + model, + hf_token=os.environ.get("HF_TOKEN"), + revision=model_revision, + ) + click.echo("Model loaded.", file=progress_file) + records, rows, truncated_rules = predict_targets( + selected, + predictor, + batch=batch, + auto_score=auto_score, + progress_file=progress_file, + ) + + if predict_only: + if json_output: + write_json_predictions(rows, json_output) + else: + print_prediction_rows(rows) + return + if not rows: + click.echo("No model predictions found") + return + + session_path = create_session_path(session_path) + metadata = create_metadata( + model=model, + model_revision=model_revision, + target_mode=target_mode, + target=target, + run_mode="batch" if batch else "interactive", + rules_scanned=rules_scanned, + rules_eligible=len(selected), + truncated_rules=truncated_rules, + auto_score=auto_score, + review_score=review_score, + ) + write_session(session_path, metadata, records) + if verbose: + click.echo(f"Session: {session_path}") + + if batch: + finish_session( + session_path, + metadata, + records, + dry_run=dry_run, + allow_write=yes, + ) + return + + loaded_rules, reconciled = load_session_rules(metadata, records) + if reconciled: + write_session(session_path, metadata, records) + complete = review_predictions( + session_path, + metadata, + records, + loaded_rules, + color, + verbose, + ) + if not complete: + print_summary(metadata, records) + click.echo(f"Resume with: {resume_command(session_path)}") + return + finish_session( + session_path, + metadata, + records, + dry_run=dry_run, + allow_write=None, + ) + + +def run_resume(session_path, dry_run, color, verbose): + """Resume review or application without running inference.""" + metadata, records = read_session(session_path) + loaded_rules, reconciled = load_session_rules(metadata, records) + if reconciled: + click.echo(f"Recovered interrupted writes: {', '.join(reconciled)}") + write_session(session_path, metadata, records) + + complete = review_predictions( + session_path, + metadata, + records, + loaded_rules, + color, + verbose, + ) + if not complete: + print_summary(metadata, records) + click.echo(f"Resume with: {resume_command(session_path)}") + return + finish_session( + session_path, + metadata, + records, + dry_run=dry_run, + allow_write=None, + ) + + +@click.command(name="add-model-required-phrases") +@click.option( + "--rule", + "rule_path", + type=click.Path(path_type=Path), + help="Review one .RULE file.", +) +@click.option( + "--rules-dir", + "rules_directory", + type=click.Path(path_type=Path), + help="Review top-level .RULE files in a directory.", +) +@click.option("--all", "all_rules", is_flag=True, help="Review eligible installed rules.") +@click.option("-l", "--license-expression", help="Only use rules for this expression.") +@click.option( + "--limit", + default=0, + type=click.IntRange(min=0), + help="Stop after this many eligible rules; zero uses all.", +) +@click.option("--predict-only", is_flag=True, help="Print predictions without decisions or writes.") +@click.option("--batch", is_flag=True, help="Classify predictions using explicit scores.") +@click.option( + "--resume", + type=click.Path(path_type=Path), + help="Resume an existing review session.", +) +@click.option("--model", help="Local model or alternate Hugging Face repository.") +@click.option("--model-revision", help="Full commit hash for a remote model.") +@click.option( + "--auto-score", + type=click.FloatRange(min=0, max=1), + help="Batch automatic-approval score.", +) +@click.option( + "--review-score", + type=click.FloatRange(min=0, max=1), + help="Batch pending-review score.", +) +@click.option("--yes", is_flag=True, help="Permit batch writes after preflight.") +@click.option( + "--session", + "session_path", + type=click.Path(path_type=Path), + help="Use this path for a new review session.", +) +@click.option( + "--json", + "json_output", + type=click.Path(path_type=str, allow_dash=True), + help="Write predict-only JSON to a file or '-' for stdout.", +) +@click.option("--dry-run", is_flag=True, help="Validate and preview without writing rules.") +@click.option("--no-color", is_flag=True, help="Disable colored output.") +@click.option("-v", "--verbose", is_flag=True, help="Print additional processing details.") +@click.help_option("-h", "--help") +def add_model_required_phrases( + rule_path, + rules_directory, + all_rules, + license_expression, + limit, + predict_only, + batch, + resume, + model, + model_revision, + auto_score, + review_score, + yes, + session_path, + json_output, + dry_run, + no_color, + verbose, +): + """Review and add model-predicted required phrases to license rules.""" + color = ( + not no_color + and "NO_COLOR" not in os.environ + and click.get_text_stream("stdout").isatty() + ) + try: + validate_options( + rule_path, + rules_directory, + all_rules, + license_expression, + limit, + predict_only, + batch, + resume, + model, + model_revision, + auto_score, + review_score, + yes, + session_path, + json_output, + ) + if resume: + run_resume(resume, dry_run=dry_run, color=color, verbose=verbose) + else: + model, model_revision = resolve_model(model, model_revision) + run_new( + rule_path, + rules_directory, + all_rules, + license_expression, + limit, + predict_only, + batch, + model, + model_revision, + auto_score, + review_score, + yes, + session_path, + json_output, + dry_run, + verbose, + color, + ) + except (InvalidRule, OSError, ValueError) as error: + raise click.ClickException(str(error)) from error + + +if __name__ == "__main__": + add_model_required_phrases() diff --git a/src/scancode_required_phrases/review.py b/src/scancode_required_phrases/review.py new file mode 100644 index 0000000..6ca6162 --- /dev/null +++ b/src/scancode_required_phrases/review.py @@ -0,0 +1,584 @@ +# Copyright (c) nexB Inc. and others. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Store and apply reviewed model predictions.""" + +import hashlib +import json +import math +import os +from pathlib import Path +import tempfile + +import click + +from licensedcode.models import Rule +from licensedcode.models import rules_data_dir + +from scancode_required_phrases.inference import words_from_text +from scancode_required_phrases.model_rules import file_sha256 +from scancode_required_phrases.model_rules import prepare_predicted_phrases +from scancode_required_phrases.model_rules import serialize_rule +from scancode_required_phrases.model_rules import write_rule_atomically + + +FORMAT_VERSION = 1 +PENDING = "pending" +APPROVED = "approved" +REJECTED = "rejected" +DECISIONS = {PENDING, APPROVED, REJECTED} +DECISION_SOURCES = {None, "human", "score"} +VALIDATION_ISSUES = {None, "rejected", "not_found", "ambiguous"} +TARGET_MODES = {"rule", "rules_dir", "all"} +RUN_MODES = {"interactive", "batch"} + +METADATA_FIELDS = { + "record_type", + "format_version", + "model", + "model_revision", + "target_mode", + "target", + "run_mode", + "auto_score", + "review_score", + "rules_scanned", + "rules_eligible", + "truncated_rules", +} +RULE_FIELDS = { + "record_type", + "path", + "identifier", + "license_expression", + "original_hash", + "truncated", + "predictions", + "expected_hash", + "applied_hash", +} +PREDICTION_FIELDS = { + "predicted_text", + "text", + "start_word", + "end_word", + "score", + "validation_issue", + "decision", + "decision_source", +} + + +def _is_hash(value): + return ( + type(value) is str + and len(value) == 64 + and all(character in "0123456789abcdef" for character in value) + ) + + +def _is_number(value): + return not isinstance(value, bool) and isinstance(value, (int, float)) + + +def create_metadata( + model, + model_revision, + target_mode, + target, + run_mode, + rules_scanned, + rules_eligible, + truncated_rules, + auto_score=None, + review_score=None, +): + """Return metadata for a new review session.""" + return { + "record_type": "metadata", + "format_version": FORMAT_VERSION, + "model": model, + "model_revision": model_revision, + "target_mode": target_mode, + "target": str(target), + "run_mode": run_mode, + "auto_score": auto_score, + "review_score": review_score, + "rules_scanned": rules_scanned, + "rules_eligible": rules_eligible, + "truncated_rules": truncated_rules, + } + + +def create_prediction(prediction, validation_issue=None): + """Return a pending session entry for one model prediction.""" + return { + "predicted_text": prediction.text, + "text": prediction.text, + "start_word": prediction.start_word, + "end_word": prediction.end_word, + "score": prediction.score, + "validation_issue": validation_issue, + "decision": PENDING, + "decision_source": None, + } + + +def create_rule_record(rule_path, rule, predictions, truncated): + """Return a session entry for one rule with predictions.""" + return { + "record_type": "rule", + "path": str(rule_path), + "identifier": rule.identifier, + "license_expression": rule.license_expression, + "original_hash": file_sha256(rule_path), + "truncated": truncated, + "predictions": predictions, + "expected_hash": None, + "applied_hash": None, + } + + +def _validate_metadata(metadata, session_path): + location = f"{session_path} metadata" + if type(metadata) is not dict or set(metadata) != METADATA_FIELDS: + raise ValueError(f"{location}: invalid fields") + if metadata["record_type"] != "metadata": + raise ValueError(f"{location}: invalid record type") + if metadata["format_version"] != FORMAT_VERSION: + raise ValueError( + f"{location}: unsupported format version {metadata['format_version']!r}" + ) + if type(metadata["model"]) is not str or not metadata["model"]: + raise ValueError(f"{location}: model must be a non-empty string") + + revision = metadata["model_revision"] + if revision is not None and ( + type(revision) is not str + or len(revision) != 40 + or any(character not in "0123456789abcdef" for character in revision) + ): + raise ValueError(f"{location}: model revision is invalid") + if metadata["target_mode"] not in TARGET_MODES: + raise ValueError(f"{location}: target mode is invalid") + if metadata["run_mode"] not in RUN_MODES: + raise ValueError(f"{location}: run mode is invalid") + + target = metadata["target"] + if type(target) is not str or not target or not Path(target).is_absolute(): + raise ValueError(f"{location}: target must be an absolute path") + + for field in ("rules_scanned", "rules_eligible", "truncated_rules"): + value = metadata[field] + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ValueError(f"{location}: {field} must be a non-negative integer") + if metadata["rules_eligible"] > metadata["rules_scanned"]: + raise ValueError(f"{location}: eligible rule count exceeds scanned rules") + if metadata["truncated_rules"] > metadata["rules_eligible"]: + raise ValueError(f"{location}: truncated rule count exceeds eligible rules") + + auto_score = metadata["auto_score"] + review_score = metadata["review_score"] + if metadata["run_mode"] == "interactive": + if auto_score is not None or review_score is not None: + raise ValueError(f"{location}: interactive sessions cannot have thresholds") + else: + if not _is_number(auto_score) or not _is_number(review_score): + raise ValueError(f"{location}: batch thresholds must be numbers") + if not ( + math.isfinite(auto_score) + and math.isfinite(review_score) + and 0 <= review_score <= auto_score <= 1 + ): + raise ValueError(f"{location}: batch thresholds are invalid") + + +def _validate_prediction(prediction, metadata, truncated, location): + if type(prediction) is not dict or set(prediction) != PREDICTION_FIELDS: + raise ValueError(f"{location}: invalid fields") + + for field in ("predicted_text", "text"): + if type(prediction[field]) is not str or not prediction[field]: + raise ValueError(f"{location}: {field} must be a non-empty string") + for field in ("start_word", "end_word"): + value = prediction[field] + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"{location}: {field} must be an integer") + if prediction["start_word"] < 0 or prediction["end_word"] < prediction["start_word"]: + raise ValueError(f"{location}: word offsets are invalid") + + score = prediction["score"] + if not _is_number(score) or not math.isfinite(score) or not 0 <= score <= 1: + raise ValueError(f"{location}: score must be between 0 and 1") + issue = prediction["validation_issue"] + decision = prediction["decision"] + source = prediction["decision_source"] + if issue not in VALIDATION_ISSUES: + raise ValueError(f"{location}: validation issue is invalid") + if decision not in DECISIONS: + raise ValueError(f"{location}: decision is invalid") + if source not in DECISION_SOURCES: + raise ValueError(f"{location}: decision source is invalid") + + if issue and decision != PENDING: + raise ValueError(f"{location}: invalid predictions cannot have decisions") + if decision == PENDING and source is not None: + raise ValueError(f"{location}: pending prediction cannot have a source") + if decision == APPROVED and source not in {"human", "score"}: + raise ValueError(f"{location}: approved prediction has no decision source") + if decision == REJECTED and source != "human": + raise ValueError(f"{location}: rejected prediction must be a human decision") + if ( + metadata["run_mode"] == "batch" + and issue is None + and score < metadata["review_score"] + and decision != PENDING + ): + raise ValueError(f"{location}: below-threshold prediction cannot have a decision") + if source == "score": + if metadata["run_mode"] != "batch": + raise ValueError(f"{location}: score decision requires batch mode") + if score < metadata["auto_score"]: + raise ValueError(f"{location}: score decision is below the automatic threshold") + if truncated: + raise ValueError(f"{location}: truncated prediction cannot be score-approved") + if prediction["text"] != prediction["predicted_text"]: + raise ValueError(f"{location}: score-approved prediction cannot be edited") + + +def _validate_rule(record, metadata, session_path, line_number): + location = f"{session_path} line {line_number}" + if type(record) is not dict or set(record) != RULE_FIELDS: + raise ValueError(f"{location}: invalid fields") + if record["record_type"] != "rule": + raise ValueError(f"{location}: invalid record type") + + path = record["path"] + identifier = record["identifier"] + expression = record["license_expression"] + if type(path) is not str or not Path(path).is_absolute(): + raise ValueError(f"{location}: path must be absolute") + if type(identifier) is not str or Path(path).name != identifier: + raise ValueError(f"{location}: identifier does not match path") + if not identifier.endswith(".RULE"): + raise ValueError(f"{location}: identifier must end with .RULE") + if type(expression) is not str or not expression: + raise ValueError(f"{location}: license expression must be a non-empty string") + if not _is_hash(record["original_hash"]): + raise ValueError(f"{location}: original hash is invalid") + if type(record["truncated"]) is not bool: + raise ValueError(f"{location}: truncated must be a boolean") + + expected_hash = record["expected_hash"] + applied_hash = record["applied_hash"] + if expected_hash is not None and not _is_hash(expected_hash): + raise ValueError(f"{location}: expected hash is invalid") + if applied_hash is not None and not _is_hash(applied_hash): + raise ValueError(f"{location}: applied hash is invalid") + if expected_hash == record["original_hash"]: + raise ValueError(f"{location}: expected hash equals original hash") + if applied_hash is not None and applied_hash != expected_hash: + raise ValueError(f"{location}: applied hash does not match expected hash") + + predictions = record["predictions"] + if type(predictions) is not list or not predictions: + raise ValueError(f"{location}: predictions must be a non-empty list") + seen = set() + for prediction_number, prediction in enumerate(predictions, 1): + prediction_location = f"{location}, prediction {prediction_number}" + _validate_prediction( + prediction, + metadata=metadata, + truncated=record["truncated"], + location=prediction_location, + ) + identity = ( + prediction["predicted_text"], + prediction["start_word"], + prediction["end_word"], + ) + if identity in seen: + raise ValueError(f"{location}: duplicate prediction") + seen.add(identity) + + has_pending = any( + prediction_needs_review(metadata, prediction) for prediction in predictions + ) + has_approved = any( + prediction["decision"] == APPROVED + and prediction["validation_issue"] is None + and not prediction_is_below_threshold(metadata, prediction) + for prediction in predictions + ) + if expected_hash is not None and has_pending: + raise ValueError(f"{location}: prepared rule has pending predictions") + if expected_hash is not None and not has_approved: + raise ValueError(f"{location}: prepared rule has no approved predictions") + + +def validate_session(metadata, records, session_path): + """Validate complete session data without opening any rule path.""" + _validate_metadata(metadata, session_path) + if type(records) is not list: + raise ValueError(f"{session_path}: rule records must be a list") + + paths = set() + identifiers = set() + for line_number, record in enumerate(records, 2): + _validate_rule(record, metadata, session_path, line_number) + if record["path"] in paths: + raise ValueError(f"{session_path} line {line_number}: duplicate rule path") + if record["identifier"] in identifiers: + raise ValueError(f"{session_path} line {line_number}: duplicate rule identifier") + paths.add(record["path"]) + identifiers.add(record["identifier"]) + + +def create_session_path(session_path=None): + """Return a new explicit or automatically reserved session path.""" + if session_path: + session_path = Path(session_path) + if session_path.exists() or session_path.is_symlink(): + raise ValueError(f"Session file already exists: {session_path}") + session_path.parent.mkdir(parents=True, exist_ok=True) + session_path = session_path.parent.resolve() / session_path.name + descriptor = os.open( + session_path, + os.O_CREAT | os.O_EXCL | os.O_WRONLY, + 0o600, + ) + os.close(descriptor) + return session_path + + sessions_directory = ( + Path(click.get_app_dir("scancode-required-phrases", roaming=False)) / "sessions" + ) + sessions_directory.mkdir(parents=True, exist_ok=True) + descriptor, name = tempfile.mkstemp( + prefix="model-required-phrases-", + suffix=".jsonl", + dir=sessions_directory, + ) + os.close(descriptor) + return Path(name).resolve() + + +def write_session(session_path, metadata, records): + """Validate and atomically replace a JSONL session.""" + session_path = Path(session_path) + validate_session(metadata, records, session_path) + if session_path.is_symlink(): + raise ValueError(f"Session file cannot be a symbolic link: {session_path}") + if not session_path.parent.is_dir(): + raise ValueError(f"Session directory does not exist: {session_path.parent}") + + descriptor, name = tempfile.mkstemp( + prefix=f".{session_path.name}.", + suffix=".tmp", + dir=session_path.parent, + ) + temporary_path = Path(name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as output: + output.write(json.dumps(metadata, ensure_ascii=False, allow_nan=False) + "\n") + for record in records: + output.write(json.dumps(record, ensure_ascii=False, allow_nan=False) + "\n") + output.flush() + os.fsync(output.fileno()) + if session_path.is_symlink(): + raise ValueError(f"Session file cannot be a symbolic link: {session_path}") + os.replace(temporary_path, session_path) + finally: + if temporary_path.exists(): + temporary_path.unlink() + + +def read_session(session_path): + """Return validated metadata and records from a JSONL session.""" + session_path = Path(session_path) + if session_path.is_symlink(): + raise ValueError(f"Session file cannot be a symbolic link: {session_path}") + try: + lines = session_path.read_text(encoding="utf-8").splitlines() + except (OSError, UnicodeError) as error: + raise ValueError(f"Cannot read session {session_path}: {error}") from error + if not lines: + raise ValueError(f"Session is empty: {session_path}") + + entries = [] + for line_number, line in enumerate(lines, 1): + if not line: + raise ValueError(f"{session_path} line {line_number}: empty line") + try: + entries.append(json.loads(line)) + except json.JSONDecodeError as error: + raise ValueError( + f"{session_path} line {line_number}: malformed JSON: {error.msg}" + ) from error + + metadata, records = entries[0], entries[1:] + validate_session(metadata, records, session_path) + return metadata, records + + +def validate_session_paths(metadata, records): + """Validate every rule path against the recorded target.""" + target = Path(metadata["target"]) + if target.is_symlink(): + raise ValueError(f"Session target cannot be a symbolic link: {target}") + resolved_target = target.resolve(strict=False) + if resolved_target != target: + raise ValueError(f"Session target is not canonical: {target}") + + if metadata["target_mode"] == "all": + installed_rules = Path(rules_data_dir).resolve(strict=True) + if target != installed_rules: + raise ValueError("Session target is not the installed ScanCode rules directory") + + if metadata["target_mode"] == "rule" and len(records) > 1: + raise ValueError("Single-rule session contains multiple rule records") + + for record in records: + rule_path = Path(record["path"]) + if rule_path.is_symlink(): + raise ValueError(f"Rule file cannot be a symbolic link: {rule_path}") + resolved_path = rule_path.resolve(strict=False) + if resolved_path != rule_path: + raise ValueError(f"Rule path is not canonical: {rule_path}") + if metadata["target_mode"] == "rule": + if rule_path != target: + raise ValueError(f"Rule path does not match the session target: {rule_path}") + elif not rule_path.is_relative_to(target): + raise ValueError(f"Rule path is outside the session target: {rule_path}") + + +def prediction_is_below_threshold(metadata, prediction): + """Return whether a valid batch prediction is below the review threshold.""" + return ( + metadata["run_mode"] == "batch" + and prediction["validation_issue"] is None + and prediction["score"] < metadata["review_score"] + ) + + +def prediction_needs_review(metadata, prediction): + """Return whether a prediction still needs a human decision.""" + return ( + prediction["validation_issue"] is None + and not prediction_is_below_threshold(metadata, prediction) + and prediction["decision"] == PENDING + ) + + +def _load_rule(record): + """Load an unchanged rule and validate its recorded predictions.""" + rule_path = Path(record["path"]) + if not rule_path.is_file(): + raise ValueError(f"Rule file is missing: {rule_path}") + if file_sha256(rule_path) != record["original_hash"]: + raise ValueError(f"Rule file is stale: {rule_path}") + + rule = Rule.from_file(str(rule_path)) + if rule.identifier != record["identifier"]: + raise ValueError(f"Rule identifier changed: {rule_path}") + if rule.license_expression != record["license_expression"]: + raise ValueError(f"Rule license expression changed: {rule_path}") + + words = words_from_text(rule.text) + for prediction in record["predictions"]: + start = prediction["start_word"] + end = prediction["end_word"] + if end >= len(words) or " ".join(words[start : end + 1]) != prediction["predicted_text"]: + raise ValueError(f"Prediction offsets do not match rule text: {rule_path}") + return rule + + +def load_session_rules(metadata, records): + """Reconcile applied rules and load every remaining unchanged rule.""" + validate_session(metadata, records, "session") + validate_session_paths(metadata, records) + loaded_rules = {} + reconciled = [] + + for record in records: + rule_path = Path(record["path"]) + if not rule_path.is_file(): + raise ValueError(f"Rule file is missing: {rule_path}") + current_hash = file_sha256(rule_path) + original_hash = record["original_hash"] + expected_hash = record["expected_hash"] + applied_hash = record["applied_hash"] + + if applied_hash is not None: + if current_hash == applied_hash: + continue + if current_hash == original_hash: + raise ValueError(f"Applied rule was reverted: {rule_path}") + raise ValueError(f"Applied rule is stale: {rule_path}") + + if expected_hash is not None and current_hash == expected_hash: + record["applied_hash"] = expected_hash + reconciled.append(record["identifier"]) + continue + if current_hash != original_hash: + raise ValueError(f"Rule file is stale: {rule_path}") + loaded_rules[record["path"]] = _load_rule(record) + + return loaded_rules, reconciled + + +def prepare_rule_updates(metadata, records, loaded_rules): + """Prepare fully decided rule updates before any write.""" + work = [] + unchanged = [] + deferred = [] + for record in records: + if record["applied_hash"] is not None: + continue + if any( + prediction_needs_review(metadata, prediction) + for prediction in record["predictions"] + ): + record["expected_hash"] = None + deferred.append(record["identifier"]) + continue + + phrases = [ + prediction["text"] + for prediction in record["predictions"] + if prediction["decision"] == APPROVED + and prediction["validation_issue"] is None + and not prediction_is_below_threshold(metadata, prediction) + ] + if not phrases: + record["expected_hash"] = None + continue + + rule_path = Path(record["path"]) + rule = loaded_rules[record["path"]] + updated_rule = prepare_predicted_phrases(rule, phrases) + content = serialize_rule(updated_rule, rule_path) + original_content = rule_path.read_bytes() + if hashlib.sha256(original_content).hexdigest() != record["original_hash"]: + raise ValueError(f"Rule file is stale: {rule_path}") + if content == original_content: + record["expected_hash"] = None + unchanged.append(record["identifier"]) + continue + + record["expected_hash"] = hashlib.sha256(content).hexdigest() + work.append((record, rule_path, content)) + return work, unchanged, deferred + + +def write_rule_updates(session_path, metadata, records, work): + """Write prepared rules and save progress after each successful write.""" + write_session(session_path, metadata, records) + for record, rule_path, content in work: + written_hash = write_rule_atomically( + rule_path=rule_path, + content=content, + expected_sha256=record["original_hash"], + ) + record["applied_hash"] = written_hash + write_session(session_path, metadata, records) diff --git a/src/scancode_required_phrases/review_ui.py b/src/scancode_required_phrases/review_ui.py new file mode 100644 index 0000000..f878310 --- /dev/null +++ b/src/scancode_required_phrases/review_ui.py @@ -0,0 +1,306 @@ +# Copyright (c) nexB Inc. and others. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Display and collect model prediction reviews.""" + +import difflib +import os +from pathlib import Path +import shlex +import shutil +from subprocess import list2cmdline + +import click + +from scancode_required_phrases.inference import words_from_text +from scancode_required_phrases.model_rules import candidate_issue +from scancode_required_phrases.model_rules import prepare_predicted_phrases +from scancode_required_phrases.model_rules import serialize_rule +from scancode_required_phrases.review import APPROVED +from scancode_required_phrases.review import prediction_is_below_threshold +from scancode_required_phrases.review import prediction_needs_review +from scancode_required_phrases.review import REJECTED +from scancode_required_phrases.review import write_session + + +def resume_command(session_path): + """Return a command that resumes a session on the current platform.""" + command = ["add-model-required-phrases", "--resume", str(session_path)] + if os.name == "nt": + return list2cmdline(command) + return shlex.join(command) + + +def approved_phrases(record, current_prediction=None, replacement=None): + """Return approved phrase text, optionally including the current candidate.""" + phrases = [ + prediction["text"] + for prediction in record["predictions"] + if prediction["decision"] == APPROVED + ] + if current_prediction is not None: + phrases.append(replacement or current_prediction["text"]) + return phrases + + +def preview_update(rule_path, rule, record, prediction, replacement=None): + """Return exact serialized bytes for a cumulative candidate update.""" + phrases = approved_phrases(record, prediction, replacement) + updated_rule = prepare_predicted_phrases(rule, phrases) + return serialize_rule(updated_rule, rule_path) + + +def print_diff(rule_path, before, after, color): + """Print relevant unified-diff hunks from exact rule bytes.""" + lines = difflib.unified_diff( + before.decode("utf-8").splitlines(keepends=True), + after.decode("utf-8").splitlines(keepends=True), + fromfile=f"a/{rule_path.name}", + tofile=f"b/{rule_path.name}", + n=3, + ) + colors = {"+": "green", "-": "red", "@": "cyan"} + for line in lines: + line = line.rstrip("\n") + click.echo( + click.style(line, fg=colors.get(line[:1])) if color else line, + color=color, + ) + + +def phrase_context(rule, prediction, color): + """Return short ScanCode-tokenized context around a prediction.""" + words = words_from_text(rule.text) + start = prediction["start_word"] + end = prediction["end_word"] + before = " ".join(words[max(0, start - 8) : start]) + phrase = " ".join(words[start : end + 1]) + after = " ".join(words[end + 1 : end + 9]) + phrase = click.style(phrase, bold=True, fg="yellow") if color else phrase + return " ".join(part for part in (before, phrase, after) if part) + + +def show_prediction( + rule_path, + rule, + record, + prediction, + position, + total, + color, + verbose, +): + """Show one pending prediction and its cumulative diff.""" + identifier = click.style(record["identifier"], bold=True) if color else record["identifier"] + phrase = click.style(prediction["text"], fg="cyan", bold=True) if color else prediction["text"] + score = f"{prediction['score']:.1%}" + score = click.style(score, fg="yellow", bold=True) if color else score + click.echo(f"\n[{position}/{total}] {identifier}", color=color) + click.echo(f"{phrase} {score}", color=color) + + expression = record["license_expression"] + if verbose: + click.echo(f"path: {rule_path}") + click.echo(f"expression: {expression}") + else: + width = max(40, shutil.get_terminal_size((100, 20)).columns - 12) + if len(expression) > width: + expression = f"{expression[: width - 3]}..." + click.echo(f"expression: {expression}") + if record["truncated"]: + warning = click.style("warning: rule input was truncated", fg="yellow") + click.echo(warning, color=color) + click.echo(f"context: {phrase_context(rule, prediction, color)}", color=color) + + try: + content = preview_update(rule_path, rule, record, prediction) + except ValueError as error: + click.echo(f"cannot approve with current decisions: {error}") + return None + print_diff(rule_path, rule_path.read_bytes(), content, color) + return content + + +def edit_prediction(rule_path, rule, record, prediction, color): + """Prompt for and approve a valid replacement phrase.""" + click.echo("\nrule text") + click.echo(rule.text) + replacement = click.prompt( + "phrase, empty to cancel", + default="", + show_default=False, + ).strip() + if not replacement: + return False + issue = candidate_issue(rule, replacement) + if issue: + click.echo(f"The phrase is not a valid candidate: {issue}") + return False + try: + content = preview_update( + rule_path, + rule, + record, + prediction, + replacement=replacement, + ) + except ValueError as error: + click.echo(f"The phrase cannot be approved: {error}") + return False + + print_diff(rule_path, rule_path.read_bytes(), content, color) + prediction["text"] = replacement + prediction["decision"] = APPROVED + prediction["decision_source"] = "human" + return True + + +def review_predictions(session_path, metadata, records, loaded_rules, color, verbose=False): + """Review each actionable prediction once and return False when the user quits.""" + pending = [ + (record, prediction) + for record in records + for prediction in record["predictions"] + if prediction_needs_review(metadata, prediction) + ] + total = len(pending) + + for position, (record, prediction) in enumerate(pending, 1): + rule_path = Path(record["path"]) + rule = loaded_rules[record["path"]] + preview = show_prediction( + rule_path, + rule, + record, + prediction, + position, + total, + color, + verbose, + ) + while True: + answer = click.prompt( + "[y] approve [n] reject [e] edit [s] skip [q] quit [?] help", + default="", + show_default=False, + ).strip().lower() + if answer == "y": + if preview is None: + click.echo("This phrase cannot be approved with the current decisions.") + continue + prediction["decision"] = APPROVED + prediction["decision_source"] = "human" + write_session(session_path, metadata, records) + click.echo("Approved") + break + if answer == "n": + prediction["decision"] = REJECTED + prediction["decision_source"] = "human" + write_session(session_path, metadata, records) + click.echo("Rejected") + break + if answer == "e": + if edit_prediction(rule_path, rule, record, prediction, color): + write_session(session_path, metadata, records) + click.echo("Approved edited phrase") + break + continue + if answer == "s": + click.echo("Skipped for this review pass") + break + if answer == "q": + return False + if answer == "?": + click.echo("Approve, reject, edit, skip for later, or save and quit.") + continue + click.echo("Enter y, n, e, s, q, or ?.") + + return True + + +def session_summary(metadata, records): + """Return user-facing counts derived from a session.""" + predictions = [ + prediction + for record in records + for prediction in record["predictions"] + ] + deferred = sum( + record["applied_hash"] is None + and any( + prediction_needs_review(metadata, prediction) + for prediction in record["predictions"] + ) + for record in records + ) + ready = sum( + record["applied_hash"] is None + and not any( + prediction_needs_review(metadata, prediction) + for prediction in record["predictions"] + ) + and any( + prediction["decision"] == APPROVED + for prediction in record["predictions"] + ) + for record in records + ) + return { + "rules_scanned": metadata["rules_scanned"], + "rules_eligible": metadata["rules_eligible"], + "rules_with_predictions": len(records), + "rules_ready": ready, + "rules_deferred": deferred, + "approved": sum( + prediction["decision"] == APPROVED + and prediction["decision_source"] == "human" + and prediction["text"] == prediction["predicted_text"] + for prediction in predictions + ), + "edited": sum( + prediction["decision"] == APPROVED + and prediction["text"] != prediction["predicted_text"] + for prediction in predictions + ), + "auto_approved": sum( + prediction["decision"] == APPROVED + and prediction["decision_source"] == "score" + for prediction in predictions + ), + "rejected": sum(prediction["decision"] == REJECTED for prediction in predictions), + "pending": sum( + prediction_needs_review(metadata, prediction) for prediction in predictions + ), + "below_threshold": sum( + prediction_is_below_threshold(metadata, prediction) for prediction in predictions + ), + "invalid": sum(prediction["validation_issue"] is not None for prediction in predictions), + "truncated": metadata["truncated_rules"], + "rules_written": sum(record["applied_hash"] is not None for record in records), + } + + +def print_summary(metadata, records, unchanged=0): + """Print concise session counts.""" + counts = session_summary(metadata, records) + click.echo("\nSummary") + click.echo( + "Rules: " + f"scanned {counts['rules_scanned']} | " + f"ready {counts['rules_ready']} | " + f"deferred {counts['rules_deferred']} | " + f"unchanged {unchanged} | " + f"written {counts['rules_written']}" + ) + click.echo( + "Phrases: " + f"approved {counts['approved']} | " + f"edited {counts['edited']} | " + f"score-approved {counts['auto_approved']} | " + f"rejected {counts['rejected']} | " + f"pending {counts['pending']} | " + f"below {counts['below_threshold']} | " + f"invalid {counts['invalid']}" + ) + if counts["truncated"]: + click.echo(f"Truncated rules: {counts['truncated']}") diff --git a/tests/test_model_cli.py b/tests/test_model_cli.py new file mode 100644 index 0000000..ab06fcf --- /dev/null +++ b/tests/test_model_cli.py @@ -0,0 +1,871 @@ +# Copyright (c) nexB Inc. and others. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import json +from pathlib import Path + +from click.testing import CliRunner +import pytest + +from licensedcode.models import Rule + +from scancode_required_phrases import model_cli +from scancode_required_phrases import review +from scancode_required_phrases import review_ui +from scancode_required_phrases.inference import PhrasePrediction +from scancode_required_phrases.inference import PredictionResult + + +TEXT = ( + "Permission is granted under the MIT License to use copy modify merge publish " + "distribute sublicense and sell copies of this software without restriction" +) + + +def make_rule(identifier="mit_test.RULE", text=TEXT): + return Rule( + identifier=identifier, + license_expression="mit", + text=text, + is_license_notice=True, + relevance=100, + ) + + +def write_rule(tmp_path, identifier="mit_test.RULE", text=TEXT): + rule = make_rule(identifier, text) + rule.dump(str(tmp_path)) + return (tmp_path / identifier).resolve() + + +class FakePredictor: + def __init__(self, predictions=None, truncated=False): + self.predictions = predictions or [ + PhrasePrediction("MIT License", 5, 6, 0.9), + ] + self.truncated = truncated + self.calls = [] + + def predict(self, text): + self.calls.append(text) + return PredictionResult( + words=tuple(text.split()), + phrases=tuple(self.predictions), + truncated=self.truncated, + ) + + +def use_fake_model(monkeypatch, predictor=None): + predictor = predictor or FakePredictor() + loads = [] + + def load(*args, **kwargs): + loads.append((args, kwargs)) + return predictor + + monkeypatch.setattr(model_cli, "load_predictor", load) + return predictor, loads + + +def use_tty(monkeypatch): + monkeypatch.setattr(model_cli, "stdin_is_tty", lambda: True) + + +def invoke_rule(runner, rule_path, *options, input=None): + return runner.invoke( + model_cli.add_model_required_phrases, + ["--rule", str(rule_path), "--model", "unused", *options], + input=input, + ) + + +@pytest.mark.parametrize( + "options,message", + [ + ([], "exactly one"), + (["--all", "--rule", "x.RULE", "--model", "unused"], "exactly one"), + ( + ["--all", "--model", "unused", "--predict-only", "--batch"], + "mutually exclusive", + ), + (["--all", "--model", "unused", "--json", "-"], "requires --predict-only"), + ( + ["--all", "--model", "unused", "--predict-only", "--session", "x"], + "cannot be used", + ), + (["--all", "--model", "unused", "--batch"], "requires --auto-score"), + ( + [ + "--all", + "--model", + "unused", + "--batch", + "--auto-score", + "0.6", + "--review-score", + "0.8", + ], + "cannot exceed", + ), + (["--all", "--model", "unused", "--yes"], "require --batch"), + ], +) +def test_command_rejects_invalid_option_combinations(options, message): + result = CliRunner().invoke(model_cli.add_model_required_phrases, options) + + assert result.exit_code != 0 + assert message in result.output + + +def test_model_resolution_uses_only_the_pinned_default(): + assert model_cli.resolve_model(None, None) == ( + model_cli.DEFAULT_MODEL, + model_cli.DEFAULT_MODEL_REVISION, + ) + assert model_cli.resolve_model("local-model", None) == ("local-model", None) + assert model_cli.resolve_model("owner/model", "a" * 40) == ( + "owner/model", + "a" * 40, + ) + + +def test_command_rejects_resume_with_new_run_options(tmp_path, monkeypatch): + use_tty(monkeypatch) + result = CliRunner().invoke( + model_cli.add_model_required_phrases, + ["--resume", str(tmp_path / "session.jsonl"), "--all"], + ) + + assert result.exit_code != 0 + assert "cannot be used with new-run options" in result.output + + +def test_command_rejects_non_tty_before_loading_model(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + monkeypatch.setattr(model_cli, "stdin_is_tty", lambda: False) + monkeypatch.setattr( + model_cli, + "load_prediction_rule", + lambda *args, **kwargs: pytest.fail("rule selection must not run"), + ) + monkeypatch.setattr( + model_cli, + "load_predictor", + lambda *args, **kwargs: pytest.fail("model must not load"), + ) + + result = invoke_rule(CliRunner(), rule_path) + + assert result.exit_code != 0 + assert "interactive terminal" in result.output + + +def test_command_rejects_existing_session_before_loading_model(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + session_path.write_text("keep", encoding="utf-8") + use_tty(monkeypatch) + monkeypatch.setattr( + model_cli, + "load_predictor", + lambda *args, **kwargs: pytest.fail("model must not load"), + ) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + ) + + assert result.exit_code != 0 + assert "already exists" in result.output + assert session_path.read_text(encoding="utf-8") == "keep" + + +def test_predict_only_json_stdout_is_clean(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + _predictor, loads = use_fake_model(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--predict-only", + "--json", + "-", + ) + + assert result.exit_code == 0, result.output + rows = json.loads(result.stdout) + assert rows == [ + { + "path": str(rule_path), + "identifier": "mit_test.RULE", + "license_expression": "mit", + "phrase": "MIT License", + "score": 0.9, + "start_word": 5, + "end_word": 6, + "truncated": False, + "validation_issue": None, + } + ] + assert "Selecting rules..." in result.stderr + assert "Loading model..." in result.stderr + assert "Model loaded." in result.stderr + assert len(loads) == 1 + assert "{{" not in Rule.from_file(str(rule_path)).text + + +def test_predict_only_json_stdout_is_empty_array_without_eligible_rules(monkeypatch): + monkeypatch.setattr(model_cli, "select_installed_prediction_rules", lambda expression: []) + monkeypatch.setattr(model_cli, "rules_data_dir", str(Path.cwd())) + monkeypatch.setattr( + model_cli, + "load_predictor", + lambda *args, **kwargs: pytest.fail("model must not load"), + ) + + result = CliRunner().invoke( + model_cli.add_model_required_phrases, + ["--all", "--model", "unused", "--predict-only", "--json", "-"], + ) + + assert result.exit_code == 0, result.output + assert json.loads(result.stdout) == [] + assert "No eligible rules found" in result.stderr + + +def test_new_run_uses_pinned_default_model(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + _predictor, loads = use_fake_model(monkeypatch) + + result = CliRunner().invoke( + model_cli.add_model_required_phrases, + ["--rule", str(rule_path), "--predict-only"], + ) + + assert result.exit_code == 0, result.output + assert loads == [ + ( + (model_cli.DEFAULT_MODEL,), + {"hf_token": None, "revision": model_cli.DEFAULT_MODEL_REVISION}, + ) + ] + + +def test_predict_only_prints_validation_status(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + use_fake_model( + monkeypatch, + FakePredictor([PhrasePrediction("is", 1, 1, 0.99)], truncated=True), + ) + + result = invoke_rule(CliRunner(), rule_path, "--predict-only") + + assert result.exit_code == 0, result.output + assert "mit_test.RULE" in result.output + assert "99.0%" in result.output + assert "rejected" in result.output + assert "truncated" in result.output + + +def test_predict_only_writes_json_file(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + output_path = tmp_path / "output" / "predictions.json" + use_fake_model(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--predict-only", + "--json", + str(output_path), + ) + + assert result.exit_code == 0, result.output + assert json.loads(output_path.read_text(encoding="utf-8"))[0]["phrase"] == "MIT License" + + +def test_interactive_dry_run_shows_review_and_writes_nothing(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + before = rule_path.read_bytes() + use_fake_model(monkeypatch) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + "--dry-run", + input="y\n", + ) + + assert result.exit_code == 0, result.output + assert rule_path.read_bytes() == before + assert session_path.is_file() + for text in ( + "[1/1] mit_test.RULE", + "MIT License 90.0%", + "expression: mit", + "context:", + "{{MIT License}}", + "Rules: scanned 1 | ready 1 | deferred 0 | unchanged 0 | written 0", + "Dry run: no rules were written", + ): + assert text in result.output + + +def test_verbose_review_shows_full_path(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + use_fake_model(monkeypatch) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + "--dry-run", + "--verbose", + input="n\n", + ) + + assert result.exit_code == 0, result.output + assert f"path: {rule_path}" in result.output + assert "expression: mit" in result.output + assert f"Session: {session_path}" in result.output + + +def test_interactive_save_exits_without_writing(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + before = rule_path.read_bytes() + use_fake_model(monkeypatch) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + input="y\ns\n", + ) + + assert result.exit_code == 0, result.output + assert rule_path.read_bytes() == before + assert "Session saved:" in result.output + + +def test_interactive_apply_writes_the_exact_rule_once(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + use_fake_model(monkeypatch) + use_tty(monkeypatch) + writes = [] + real_write = model_cli.write_rule_updates + + def write(*args, **kwargs): + writes.append(args[0]) + return real_write(*args, **kwargs) + + monkeypatch.setattr(model_cli, "write_rule_updates", write) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + input="y\na\n", + ) + + assert result.exit_code == 0, result.output + saved = Rule.from_file(str(rule_path)) + assert "{{MIT License}}" in saved.text + assert saved.source == "ml_model" + assert writes == [session_path] + assert "written 1" in result.output + + +def test_interactive_reject_writes_nothing(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + before = rule_path.read_bytes() + use_fake_model(monkeypatch) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + input="n\n", + ) + + assert result.exit_code == 0, result.output + assert rule_path.read_bytes() == before + _metadata, records = review.read_session(session_path) + assert records[0]["predictions"][0]["decision"] == review.REJECTED + + +def test_interactive_edit_preserves_prediction(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + use_fake_model(monkeypatch) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + "--dry-run", + input="e\nMIT License to use\n", + ) + + assert result.exit_code == 0, result.output + _metadata, records = review.read_session(session_path) + prediction = records[0]["predictions"][0] + assert prediction["predicted_text"] == "MIT License" + assert prediction["text"] == "MIT License to use" + assert prediction["decision"] == review.APPROVED + assert prediction["decision_source"] == "human" + counts = review_ui.session_summary(_metadata, records) + assert counts["approved"] == 0 + assert counts["edited"] == 1 + + +def test_interactive_help_then_rejects(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + use_fake_model(monkeypatch) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + input="?\nn\n", + ) + + assert result.exit_code == 0, result.output + assert "Approve, reject, edit, skip for later" in result.output + + +def test_skip_leaves_pending_and_resume_does_not_load_model(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + use_fake_model(monkeypatch) + use_tty(monkeypatch) + runner = CliRunner() + + skipped = invoke_rule( + runner, + rule_path, + "--session", + str(session_path), + input="s\n", + ) + assert skipped.exit_code == 0, skipped.output + assert "Resume with:" in skipped.output + _metadata, records = review.read_session(session_path) + assert records[0]["predictions"][0]["decision"] == review.PENDING + + monkeypatch.setattr( + model_cli, + "load_predictor", + lambda *args, **kwargs: pytest.fail("resume must not load a model"), + ) + resumed = runner.invoke( + model_cli.add_model_required_phrases, + ["--resume", str(session_path), "--dry-run"], + input="y\n", + ) + + assert resumed.exit_code == 0, resumed.output + assert "Dry run: no rules were written" in resumed.output + + +def test_skip_moves_to_next_prediction_without_reappearing(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + predictor = FakePredictor( + [ + PhrasePrediction("MIT License", 5, 6, 0.9), + PhrasePrediction("without restriction", 21, 22, 0.7), + ] + ) + use_fake_model(monkeypatch, predictor) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + input="s\nn\n", + ) + + assert result.exit_code == 0, result.output + _metadata, records = review.read_session(session_path) + first, second = records[0]["predictions"] + assert first["decision"] == review.PENDING + assert second["decision"] == review.REJECTED + assert result.output.count("[1/2]") == 1 + assert result.output.count("[2/2]") == 1 + + +def test_interactive_refuses_overlapping_approval(tmp_path, monkeypatch): + rule_path = write_rule( + tmp_path, + text="binary Redistribution clause applies to these software copies", + ) + session_path = tmp_path / "session.jsonl" + predictor = FakePredictor( + [ + PhrasePrediction("binary Redistribution clause", 0, 2, 0.9), + PhrasePrediction("Redistribution clause", 1, 2, 0.8), + ] + ) + use_fake_model(monkeypatch, predictor) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + input="y\ny\nn\nd\n", + ) + + assert result.exit_code == 0, result.output + assert "cannot be approved with the current decisions" in result.output + _metadata, records = review.read_session(session_path) + first, second = records[0]["predictions"] + assert first["decision"] == review.APPROVED + assert second["decision"] == review.REJECTED + + +def test_quit_prints_exact_resume_command(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session file.jsonl" + use_fake_model(monkeypatch) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + input="q\n", + ) + + assert result.exit_code == 0, result.output + assert result.output.count(model_cli.resume_command(session_path)) == 1 + + +def test_batch_classifies_exact_score_boundaries_and_blocks_mixed_write( + tmp_path, + monkeypatch, +): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "batch.jsonl" + predictions = [ + PhrasePrediction("MIT License", 5, 6, 0.8), + PhrasePrediction("without restriction", 21, 22, 0.5), + PhrasePrediction("Permission is granted", 0, 2, 0.49), + ] + use_fake_model(monkeypatch, FakePredictor(predictions)) + before = rule_path.read_bytes() + + result = invoke_rule( + CliRunner(), + rule_path, + "--batch", + "--auto-score", + "0.8", + "--review-score", + "0.5", + "--yes", + "--session", + str(session_path), + ) + + assert result.exit_code == 0, result.output + assert rule_path.read_bytes() == before + metadata, records = review.read_session(session_path) + first, second, third = records[0]["predictions"] + assert (first["decision"], first["decision_source"]) == (review.APPROVED, "score") + assert second["decision"] == review.PENDING + assert review.prediction_needs_review(metadata, second) + assert review.prediction_is_below_threshold(metadata, third) + assert "Resume with:" in result.output + + +def test_batch_applies_ready_rule_and_defers_pending_rule(tmp_path, monkeypatch): + rules_directory = tmp_path / "rules" + rules_directory.mkdir() + first_path = write_rule(rules_directory, "first.RULE") + second_path = write_rule(rules_directory, "second.RULE") + session_path = tmp_path / "batch.jsonl" + + class PerRulePredictor: + def __init__(self): + self.calls = 0 + + def predict(self, text): + self.calls += 1 + predictions = [PhrasePrediction("MIT License", 5, 6, 0.9)] + if self.calls == 2: + predictions.append(PhrasePrediction("without restriction", 21, 22, 0.7)) + return PredictionResult( + words=tuple(text.split()), + phrases=tuple(predictions), + truncated=False, + ) + + use_fake_model(monkeypatch, PerRulePredictor()) + + result = CliRunner().invoke( + model_cli.add_model_required_phrases, + [ + "--rules-dir", + str(rules_directory), + "--batch", + "--auto-score", + "0.8", + "--review-score", + "0.5", + "--yes", + "--session", + str(session_path), + "--model", + "unused", + ], + ) + + assert result.exit_code == 0, result.output + assert "{{MIT License}}" in Rule.from_file(str(first_path)).text + assert "{{" not in Rule.from_file(str(second_path)).text + _metadata, records = review.read_session(session_path) + assert records[0]["applied_hash"] is not None + assert records[1]["applied_hash"] is None + assert "deferred 1" in result.output + assert "Resume with:" in result.output + + +def test_batch_without_yes_saves_without_writing(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "batch.jsonl" + before = rule_path.read_bytes() + use_fake_model(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--batch", + "--auto-score", + "0.8", + "--review-score", + "0.5", + "--session", + str(session_path), + ) + + assert result.exit_code == 0, result.output + assert rule_path.read_bytes() == before + assert "Session saved:" in result.output + + +def test_batch_dry_run_overrides_yes(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "batch.jsonl" + before = rule_path.read_bytes() + use_fake_model(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--batch", + "--auto-score", + "0.8", + "--review-score", + "0.5", + "--yes", + "--dry-run", + "--session", + str(session_path), + ) + + assert result.exit_code == 0, result.output + assert rule_path.read_bytes() == before + assert "Dry run: no rules were written" in result.output + + +def test_batch_yes_applies_when_no_review_is_pending(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "batch.jsonl" + use_fake_model(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--batch", + "--auto-score", + "0.8", + "--review-score", + "0.5", + "--yes", + "--session", + str(session_path), + ) + + assert result.exit_code == 0, result.output + assert "{{MIT License}}" in Rule.from_file(str(rule_path)).text + + +def test_batch_never_score_approves_truncated_rule(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "batch.jsonl" + before = rule_path.read_bytes() + use_fake_model(monkeypatch, FakePredictor(truncated=True)) + + result = invoke_rule( + CliRunner(), + rule_path, + "--batch", + "--auto-score", + "0.8", + "--review-score", + "0.5", + "--yes", + "--session", + str(session_path), + ) + + assert result.exit_code == 0, result.output + assert rule_path.read_bytes() == before + metadata, records = review.read_session(session_path) + assert records[0]["predictions"][0]["decision"] == review.PENDING + assert metadata["truncated_rules"] == 1 + + +def test_batch_score_does_not_override_candidate_validation(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "batch.jsonl" + before = rule_path.read_bytes() + use_fake_model(monkeypatch, FakePredictor([PhrasePrediction("is", 1, 1, 1.0)])) + + result = invoke_rule( + CliRunner(), + rule_path, + "--batch", + "--auto-score", + "0.8", + "--review-score", + "0.5", + "--yes", + "--session", + str(session_path), + ) + + assert result.exit_code == 0, result.output + assert rule_path.read_bytes() == before + _metadata, records = review.read_session(session_path) + prediction = records[0]["predictions"][0] + assert prediction["validation_issue"] == "rejected" + assert prediction["decision"] == review.PENDING + + +def test_directory_limit_predicts_only_selected_rules(tmp_path, monkeypatch): + rules_directory = tmp_path / "rules" + rules_directory.mkdir() + write_rule(rules_directory, "b.RULE") + write_rule(rules_directory, "a.RULE") + predictor, loads = use_fake_model(monkeypatch) + + result = CliRunner().invoke( + model_cli.add_model_required_phrases, + [ + "--rules-dir", + str(rules_directory), + "--limit", + "1", + "--model", + "unused", + "--predict-only", + ], + ) + + assert result.exit_code == 0, result.output + assert len(loads) == 1 + assert predictor.calls == [TEXT] + assert "a.RULE" in result.output + assert "b.RULE" not in result.output + + +def test_all_target_reuses_installed_selection(tmp_path, monkeypatch): + rules_directory = tmp_path / "rules" + rules_directory.mkdir() + rule_path = write_rule(rules_directory) + rule = Rule.from_file(str(rule_path)) + selected = [(rule_path, rule)] + calls = [] + monkeypatch.setattr( + model_cli, + "select_installed_prediction_rules", + lambda expression: calls.append(expression) or selected, + ) + monkeypatch.setattr(model_cli, "rules_data_dir", str(rules_directory)) + use_fake_model(monkeypatch) + + result = CliRunner().invoke( + model_cli.add_model_required_phrases, + [ + "--all", + "--license-expression", + "mit", + "--model", + "unused", + "--predict-only", + ], + ) + + assert result.exit_code == 0, result.output + assert calls == ["mit"] + + +def test_no_color_removes_terminal_escape_sequences(tmp_path, monkeypatch): + rule_path = write_rule(tmp_path) + session_path = tmp_path / "session.jsonl" + use_fake_model(monkeypatch) + use_tty(monkeypatch) + + result = invoke_rule( + CliRunner(), + rule_path, + "--session", + str(session_path), + "--dry-run", + "--no-color", + input="y\n", + ) + + assert result.exit_code == 0, result.output + assert "\x1b[" not in result.output + + +def test_command_help_is_single_workflow(): + result = CliRunner().invoke(model_cli.add_model_required_phrases, ["--help"]) + + assert result.exit_code == 0 + assert "--rule" in result.output + assert "--rules-dir" in result.output + assert "--all" in result.output + assert "--predict-only" in result.output + assert "--batch" in result.output + assert "--resume" in result.output diff --git a/tests/test_review.py b/tests/test_review.py new file mode 100644 index 0000000..cb5277e --- /dev/null +++ b/tests/test_review.py @@ -0,0 +1,582 @@ +# Copyright (c) nexB Inc. and others. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import hashlib +import json +from pathlib import Path + +import pytest + +from licensedcode.models import Rule + +from scancode_required_phrases import review +from scancode_required_phrases.inference import PhrasePrediction +from scancode_required_phrases.model_rules import file_sha256 + + +TEXT = ( + "Permission is granted under the MIT License to use copy modify merge publish " + "distribute sublicense and sell copies of this software without restriction" +) + + +def make_rule(identifier="mit_test.RULE", text=TEXT): + return Rule( + identifier=identifier, + license_expression="mit", + text=text, + is_license_notice=True, + relevance=100, + ) + + +def make_prediction(text="MIT License", start=5, end=6, score=0.9): + return PhrasePrediction( + text=text, + start_word=start, + end_word=end, + score=score, + ) + + +def make_session(tmp_path, run_mode="interactive", score=0.9): + rule = make_rule() + rule.dump(str(tmp_path)) + rule_path = (tmp_path / rule.identifier).resolve() + metadata = review.create_metadata( + model="model-dir", + model_revision=None, + target_mode="rule", + target=rule_path, + run_mode=run_mode, + rules_scanned=1, + rules_eligible=1, + truncated_rules=0, + auto_score=0.8 if run_mode == "batch" else None, + review_score=0.5 if run_mode == "batch" else None, + ) + prediction = review.create_prediction(make_prediction(score=score)) + record = review.create_rule_record( + rule_path=rule_path, + rule=rule, + predictions=[prediction], + truncated=False, + ) + return metadata, [record], rule_path + + +def approve(prediction, source="human"): + prediction["decision"] = review.APPROVED + prediction["decision_source"] = source + + +def test_session_round_trip(tmp_path): + metadata, records, _rule_path = make_session(tmp_path) + session_path = tmp_path / "session.jsonl" + + review.write_session(session_path, metadata, records) + + assert review.read_session(session_path) == (metadata, records) + assert not list(tmp_path.glob(".session.jsonl.*.tmp")) + + +def test_large_session_round_trip(tmp_path): + metadata, records, _rule_path = make_session(tmp_path) + metadata["target_mode"] = "rules_dir" + metadata["target"] = str(tmp_path.resolve()) + metadata["rules_scanned"] = 500 + metadata["rules_eligible"] = 500 + template = records[0] + records = [] + for index in range(500): + identifier = f"mit_{index}.RULE" + record = dict(template) + record["path"] = str((tmp_path / identifier).resolve()) + record["identifier"] = identifier + record["predictions"] = [dict(template["predictions"][0])] + records.append(record) + session_path = tmp_path / "large.jsonl" + + review.write_session(session_path, metadata, records) + loaded_metadata, loaded_records = review.read_session(session_path) + + assert loaded_metadata == metadata + assert len(loaded_records) == 500 + assert loaded_records[-1]["identifier"] == "mit_499.RULE" + + +def test_write_session_keeps_existing_file_when_replace_fails(tmp_path, monkeypatch): + metadata, records, _rule_path = make_session(tmp_path) + session_path = tmp_path / "session.jsonl" + review.write_session(session_path, metadata, records) + before = session_path.read_bytes() + monkeypatch.setattr( + review.os, + "replace", + lambda *args: (_ for _ in ()).throw(OSError("replace failed")), + ) + + with pytest.raises(OSError, match="replace failed"): + review.write_session(session_path, metadata, records) + + assert session_path.read_bytes() == before + assert not list(tmp_path.glob(".session.jsonl.*.tmp")) + + +def test_read_session_rejects_old_schema_and_malformed_input(tmp_path): + old_session = tmp_path / "old.jsonl" + old_session.write_text(json.dumps({"format_version": 0}) + "\n", encoding="utf-8") + with pytest.raises(ValueError, match="invalid fields"): + review.read_session(old_session) + + malformed = tmp_path / "malformed.jsonl" + malformed.write_text("{bad}\n", encoding="utf-8") + with pytest.raises(ValueError, match="line 1: malformed JSON"): + review.read_session(malformed) + + invalid_utf8 = tmp_path / "invalid.jsonl" + invalid_utf8.write_bytes(b"\xff") + with pytest.raises(ValueError, match="Cannot read session"): + review.read_session(invalid_utf8) + + +def test_validate_session_rejects_unknown_fields(tmp_path): + metadata, records, _rule_path = make_session(tmp_path) + records[0]["extra"] = True + + with pytest.raises(ValueError, match="invalid fields"): + review.validate_session(metadata, records, "session.jsonl") + + +@pytest.mark.parametrize( + "change,error", + [ + (lambda metadata, prediction, record: prediction.update(score=float("nan")), "score"), + ( + lambda metadata, prediction, record: prediction.update( + validation_issue="ambiguous", + decision="approved", + decision_source="human", + ), + "invalid predictions", + ), + ( + lambda metadata, prediction, record: prediction.update( + decision="pending", + decision_source="human", + ), + "pending prediction", + ), + ( + lambda metadata, prediction, record: prediction.update( + decision="approved", + decision_source=None, + ), + "no decision source", + ), + ( + lambda metadata, prediction, record: prediction.update( + decision="rejected", + decision_source="score", + ), + "human decision", + ), + ( + lambda metadata, prediction, record: record.update( + expected_hash=record["original_hash"] + ), + "equals original", + ), + ( + lambda metadata, prediction, record: record.update(applied_hash="a" * 64), + "does not match expected", + ), + ], +) +def test_validate_session_rejects_inconsistent_state(tmp_path, change, error): + metadata, records, _rule_path = make_session(tmp_path) + change(metadata, records[0]["predictions"][0], records[0]) + + with pytest.raises(ValueError, match=error): + review.validate_session(metadata, records, "session.jsonl") + + +def test_validate_session_rejects_invalid_score_approval(tmp_path): + metadata, records, _rule_path = make_session(tmp_path, run_mode="batch", score=0.7) + approve(records[0]["predictions"][0], source="score") + + with pytest.raises(ValueError, match="below the automatic threshold"): + review.validate_session(metadata, records, "session.jsonl") + + +def test_validate_session_rejects_truncated_score_approval(tmp_path): + metadata, records, _rule_path = make_session(tmp_path, run_mode="batch") + records[0]["truncated"] = True + approve(records[0]["predictions"][0], source="score") + + with pytest.raises(ValueError, match="truncated"): + review.validate_session(metadata, records, "session.jsonl") + + +def test_validate_session_rejects_edited_score_approval(tmp_path): + metadata, records, _rule_path = make_session(tmp_path, run_mode="batch") + prediction = records[0]["predictions"][0] + prediction["text"] = "MIT License to use" + approve(prediction, source="score") + + with pytest.raises(ValueError, match="cannot be edited"): + review.validate_session(metadata, records, "session.jsonl") + + +def test_validate_session_rejects_below_threshold_decision(tmp_path): + metadata, records, _rule_path = make_session(tmp_path, run_mode="batch", score=0.4) + approve(records[0]["predictions"][0]) + + with pytest.raises(ValueError, match="below-threshold"): + review.validate_session(metadata, records, "session.jsonl") + + +def test_validate_session_rejects_duplicate_paths_and_identifiers(tmp_path): + metadata, records, _rule_path = make_session(tmp_path) + duplicate = dict(records[0]) + duplicate["predictions"] = [dict(records[0]["predictions"][0])] + + with pytest.raises(ValueError, match="duplicate rule path"): + review.validate_session(metadata, [records[0], duplicate], "session.jsonl") + + metadata["target_mode"] = "rules_dir" + metadata["target"] = str(tmp_path.resolve()) + records[0]["path"] = str((tmp_path / "first" / "mit_test.RULE").resolve()) + duplicate["path"] = str((tmp_path / "second" / "mit_test.RULE").resolve()) + with pytest.raises(ValueError, match="duplicate rule identifier"): + review.validate_session(metadata, [records[0], duplicate], "session.jsonl") + + +def test_validate_session_rejects_duplicate_predictions(tmp_path): + metadata, records, _rule_path = make_session(tmp_path) + prediction = dict(records[0]["predictions"][0]) + records[0]["predictions"].append(prediction) + + with pytest.raises(ValueError, match="duplicate prediction"): + review.validate_session(metadata, records, "session.jsonl") + + +def test_create_session_path_reserves_default_path(tmp_path, monkeypatch): + monkeypatch.setattr(review.click, "get_app_dir", lambda *args, **kwargs: str(tmp_path)) + + session_path = review.create_session_path() + + assert session_path.parent == (tmp_path / "sessions").resolve() + assert session_path.name.startswith("model-required-phrases-") + assert session_path.suffix == ".jsonl" + assert session_path.is_file() + assert session_path.stat().st_size == 0 + + +def test_create_session_path_refuses_explicit_collision(tmp_path): + session_path = tmp_path / "session.jsonl" + session_path.write_text("keep", encoding="utf-8") + + with pytest.raises(ValueError, match="already exists"): + review.create_session_path(session_path) + + assert session_path.read_text(encoding="utf-8") == "keep" + + +def test_validate_session_rejects_expected_hash_without_approval(tmp_path): + metadata, records, _rule_path = make_session(tmp_path) + records[0]["predictions"][0]["decision"] = review.REJECTED + records[0]["predictions"][0]["decision_source"] = "human" + records[0]["expected_hash"] = "a" * 64 + + with pytest.raises(ValueError, match="no approved predictions"): + review.validate_session(metadata, records, "session.jsonl") + + +def test_create_session_path_creates_explicit_parent(tmp_path): + session_path = tmp_path / "new" / "session.jsonl" + + created = review.create_session_path(session_path) + + assert created == session_path.resolve() + assert created.is_file() + assert created.stat().st_size == 0 + + +def test_validate_session_paths_rejects_path_outside_directory(tmp_path): + root = (tmp_path / "rules").resolve() + root.mkdir() + metadata, records, _rule_path = make_session(tmp_path) + metadata["target_mode"] = "rules_dir" + metadata["target"] = str(root) + + with pytest.raises(ValueError, match="outside the session target"): + review.validate_session_paths(metadata, records) + + +def test_validate_session_paths_rejects_symlink_before_hashing(tmp_path, monkeypatch): + metadata, records, rule_path = make_session(tmp_path) + target = tmp_path / "target.RULE" + rule_path.rename(target) + try: + rule_path.symlink_to(target) + except OSError: + pytest.skip("symbolic links are unavailable") + monkeypatch.setattr( + review, + "file_sha256", + lambda path: pytest.fail("path must be rejected before hashing"), + ) + + with pytest.raises(ValueError, match="symbolic link"): + review.load_session_rules(metadata, records) + + +def test_load_session_rules_loads_unchanged_rule(tmp_path): + metadata, records, rule_path = make_session(tmp_path) + + loaded, reconciled = review.load_session_rules(metadata, records) + + assert list(loaded) == [str(rule_path)] + assert loaded[str(rule_path)].identifier == rule_path.name + assert reconciled == [] + + +def expected_rule_content(record, rule_path): + rule = Rule.from_file(str(rule_path)) + updated = review.prepare_predicted_phrases(rule, ["MIT License"]) + return review.serialize_rule(updated, rule_path) + + +def test_load_session_rules_reconciles_interrupted_session_update(tmp_path): + metadata, records, rule_path = make_session(tmp_path) + approve(records[0]["predictions"][0]) + content = expected_rule_content(records[0], rule_path) + expected_hash = hashlib.sha256(content).hexdigest() + records[0]["expected_hash"] = expected_hash + rule_path.write_bytes(content) + + loaded, reconciled = review.load_session_rules(metadata, records) + + assert loaded == {} + assert reconciled == ["mit_test.RULE"] + assert records[0]["applied_hash"] == expected_hash + + +def test_load_session_rules_accepts_applied_rule(tmp_path): + metadata, records, rule_path = make_session(tmp_path) + approve(records[0]["predictions"][0]) + content = expected_rule_content(records[0], rule_path) + expected_hash = hashlib.sha256(content).hexdigest() + records[0]["expected_hash"] = expected_hash + records[0]["applied_hash"] = expected_hash + rule_path.write_bytes(content) + + loaded, reconciled = review.load_session_rules(metadata, records) + + assert loaded == {} + assert reconciled == [] + + +def test_load_session_rules_rejects_reverted_applied_rule(tmp_path): + metadata, records, _rule_path = make_session(tmp_path) + approve(records[0]["predictions"][0]) + records[0]["expected_hash"] = "a" * 64 + records[0]["applied_hash"] = "a" * 64 + + with pytest.raises(ValueError, match="reverted"): + review.load_session_rules(metadata, records) + + +def test_load_session_rules_rejects_other_stale_content(tmp_path): + metadata, records, rule_path = make_session(tmp_path) + rule_path.write_bytes(b"changed") + + with pytest.raises(ValueError, match="stale"): + review.load_session_rules(metadata, records) + + +def test_load_session_rules_rejects_changed_prediction_offsets(tmp_path): + metadata, records, _rule_path = make_session(tmp_path) + records[0]["predictions"][0]["start_word"] = 0 + records[0]["predictions"][0]["end_word"] = 1 + + with pytest.raises(ValueError, match="offsets"): + review.load_session_rules(metadata, records) + + +def test_prepare_rule_updates_defers_pending_rule(tmp_path): + metadata, records, _rule_path = make_session(tmp_path) + loaded, _reconciled = review.load_session_rules(metadata, records) + + work, unchanged, deferred = review.prepare_rule_updates(metadata, records, loaded) + + assert work == [] + assert unchanged == [] + assert deferred == ["mit_test.RULE"] + + +def test_prepare_rule_updates_keeps_pending_rule_untouched(tmp_path): + first = make_rule("first.RULE") + second = make_rule("second.RULE") + first.dump(str(tmp_path)) + second.dump(str(tmp_path)) + first_path = (tmp_path / first.identifier).resolve() + second_path = (tmp_path / second.identifier).resolve() + metadata = review.create_metadata( + model="model-dir", + model_revision=None, + target_mode="rules_dir", + target=tmp_path.resolve(), + run_mode="batch", + rules_scanned=2, + rules_eligible=2, + truncated_rules=0, + auto_score=0.8, + review_score=0.5, + ) + first_prediction = review.create_prediction(make_prediction()) + approve(first_prediction, source="score") + records = [ + review.create_rule_record(first_path, first, [first_prediction], False), + review.create_rule_record( + second_path, + second, + [review.create_prediction(make_prediction(score=0.7))], + False, + ), + ] + loaded, _reconciled = review.load_session_rules(metadata, records) + + work, unchanged, deferred = review.prepare_rule_updates(metadata, records, loaded) + + assert unchanged == [] + assert deferred == ["second.RULE"] + assert [record["identifier"] for record, _path, _content in work] == ["first.RULE"] + session_path = tmp_path / "session.jsonl" + review.write_rule_updates(session_path, metadata, records, work) + assert "{{MIT License}}" in Rule.from_file(str(first_path)).text + assert "{{" not in Rule.from_file(str(second_path)).text + + +def test_prepare_rule_updates_uses_only_approved_phrases(tmp_path): + metadata, records, rule_path = make_session(tmp_path) + rejected = review.create_prediction(make_prediction("without restriction", 21, 22, 0.7)) + rejected["decision"] = review.REJECTED + rejected["decision_source"] = "human" + records[0]["predictions"].append(rejected) + approve(records[0]["predictions"][0]) + loaded, _reconciled = review.load_session_rules(metadata, records) + + work, unchanged, deferred = review.prepare_rule_updates(metadata, records, loaded) + + assert unchanged == [] + assert deferred == [] + assert len(work) == 1 + _record, prepared_path, content = work[0] + assert prepared_path == rule_path + saved = tmp_path / "saved" + saved.mkdir() + (saved / rule_path.name).write_bytes(content) + updated = Rule.from_file(str(saved / rule_path.name)) + assert "{{MIT License}}" in updated.text + assert "{{without restriction}}" not in updated.text + assert records[0]["expected_hash"] == hashlib.sha256(content).hexdigest() + + +def test_prepare_rule_updates_clears_old_expected_hash_for_noop( + tmp_path, + monkeypatch, +): + metadata, records, rule_path = make_session(tmp_path) + approve(records[0]["predictions"][0]) + records[0]["expected_hash"] = "a" * 64 + loaded, _reconciled = review.load_session_rules(metadata, records) + monkeypatch.setattr( + review, + "serialize_rule", + lambda rule, path: rule_path.read_bytes(), + ) + + work, unchanged, deferred = review.prepare_rule_updates(metadata, records, loaded) + + assert work == [] + assert deferred == [] + assert unchanged == ["mit_test.RULE"] + assert records[0]["expected_hash"] is None + assert records[0]["applied_hash"] is None + + +def test_write_rule_updates_saves_expected_hash_before_writing(tmp_path, monkeypatch): + metadata, records, _rule_path = make_session(tmp_path) + approve(records[0]["predictions"][0]) + loaded, _reconciled = review.load_session_rules(metadata, records) + work, _unchanged, _deferred = review.prepare_rule_updates(metadata, records, loaded) + session_path = tmp_path / "session.jsonl" + writes = [] + + monkeypatch.setattr( + review, + "write_rule_atomically", + lambda **kwargs: writes.append(kwargs) or records[0]["expected_hash"], + ) + + review.write_rule_updates(session_path, metadata, records, work) + + assert len(writes) == 1 + saved_metadata, saved_records = review.read_session(session_path) + assert saved_metadata == metadata + assert saved_records[0]["applied_hash"] == saved_records[0]["expected_hash"] + + +def test_write_rule_updates_does_not_write_if_initial_session_save_fails( + tmp_path, + monkeypatch, +): + metadata, records, _rule_path = make_session(tmp_path) + approve(records[0]["predictions"][0]) + loaded, _reconciled = review.load_session_rules(metadata, records) + work, _unchanged, _deferred = review.prepare_rule_updates(metadata, records, loaded) + monkeypatch.setattr( + review, + "write_session", + lambda *args: (_ for _ in ()).throw(OSError("session save failed")), + ) + monkeypatch.setattr( + review, + "write_rule_atomically", + lambda **kwargs: pytest.fail("rule must not be written"), + ) + + with pytest.raises(OSError, match="session save failed"): + review.write_rule_updates(tmp_path / "session.jsonl", metadata, records, work) + + +def test_interrupted_session_save_after_rule_write_is_reconciled(tmp_path, monkeypatch): + metadata, records, rule_path = make_session(tmp_path) + approve(records[0]["predictions"][0]) + loaded, _reconciled = review.load_session_rules(metadata, records) + work, _unchanged, _deferred = review.prepare_rule_updates(metadata, records, loaded) + session_path = tmp_path / "session.jsonl" + real_write_session = review.write_session + calls = 0 + + def fail_second_save(*args): + nonlocal calls + calls += 1 + if calls == 2: + raise OSError("session save failed") + real_write_session(*args) + + monkeypatch.setattr(review, "write_session", fail_second_save) + + with pytest.raises(OSError, match="session save failed"): + review.write_rule_updates(session_path, metadata, records, work) + + saved_metadata, saved_records = review.read_session(session_path) + assert saved_records[0]["applied_hash"] is None + assert file_sha256(rule_path) == saved_records[0]["expected_hash"] + + loaded, reconciled = review.load_session_rules(saved_metadata, saved_records) + assert loaded == {} + assert reconciled == ["mit_test.RULE"]