From 9ebeaa442747e5d511ed65a3f0436eec2dd6bbbd Mon Sep 17 00:00:00 2001 From: Kaushik Kumar Date: Sat, 15 Aug 2026 20:42:01 +0530 Subject: [PATCH 1/7] add review_ml_phrases with the review file io --- .../dataset_pipeline/review_ml_phrases.py | 116 ++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 etc/scripts/dataset_pipeline/review_ml_phrases.py diff --git a/etc/scripts/dataset_pipeline/review_ml_phrases.py b/etc/scripts/dataset_pipeline/review_ml_phrases.py new file mode 100644 index 0000000000..46fa13e89f --- /dev/null +++ b/etc/scripts/dataset_pipeline/review_ml_phrases.py @@ -0,0 +1,116 @@ +# review the required phrases the tagger predicts before they touch a rule +# predict scores every phrase and files it, review walks the uncertain ones past +# a maintainer, apply injects what came back accepted +# a wrong required phrase silently drops a real license detection, so nothing is +# written on the model's word alone +import json +import os +import tempfile + +import click + +# tiers a phrase can land in, and the decision each one starts life with +AUTO = 'auto' +REVIEW = 'review' +LOW = 'low' + +PENDING = 'pending' +APPROVED = 'approved' +REJECTED = 'rejected' +DROPPED = 'dropped' + +# what a record and a phrase entry must carry, checked when reading a file that +# may have been hand edited +RECORD_KEYS = ('identifier', 'license_expression', 'truncated', 'phrases') +PHRASE_KEYS = ('text', 'predicted_text', 'confidence', 'tier', 'decision') + + +def phrase_sort_key(phrase): + """Longest first, the order required phrases are applied in elsewhere""" + return -len(phrase), phrase + + +def new_phrase(text, confidence, tier, decision): + """A phrase entry, built here so every record shares one key order + + predict appends records and review rewrites them, and json keeps the order + it read, so setting it once here is enough to keep the two in step + """ + return { + 'text': text, + 'predicted_text': text, + 'confidence': confidence, + 'tier': tier, + 'decision': decision, + } + + +def new_record(rule, phrases, truncated): + """One rule's worth of predictions""" + return { + 'identifier': rule.identifier, + 'license_expression': rule.license_expression, + 'truncated': truncated, + 'phrases': sorted(phrases, key=lambda phrase: phrase_sort_key(phrase['text'])), + } + + +def check_keys(record, number, path): + """Complain about the line at fault rather than dying somewhere later""" + for key in RECORD_KEYS: + if key not in record: + raise click.ClickException(f'{path} line {number}: no {key!r}') + + for phrase in record['phrases']: + for key in PHRASE_KEYS: + if key not in phrase: + raise click.ClickException(f'{path} line {number}: phrase has no {key!r}') + + +def read_review_file(path): + """Records from a review file""" + records = [] + + with open(path, encoding='utf-8') as lines: + for number, line in enumerate(lines, 1): + line = line.strip() + if not line: + continue + + try: + record = json.loads(line) + except ValueError as e: + raise click.ClickException(f'{path} line {number}: {e}') + + check_keys(record, number, path) + records.append(record) + + return records + + +def write_review_record(handle, record): + """One record, one line""" + handle.write(json.dumps(record) + '\n') + + +def write_review_file(path, records): + """Replace the review file with these records + + Written to a temporary file next to it and moved into place, so a decision + already recorded survives a crash mid rewrite + """ + handle = tempfile.NamedTemporaryFile( + mode='w', + encoding='utf-8', + dir=os.path.dirname(os.path.abspath(path)), + suffix='.tmp', + delete=False, + ) + try: + with handle: + for record in records: + write_review_record(handle, record) + os.replace(handle.name, path) + except Exception: + os.unlink(handle.name) + raise From f330b7328f7a9fbf31c421cc2afa94ea9d2fa7f8 Mon Sep 17 00:00:00 2001 From: Kaushik Kumar Date: Sat, 15 Aug 2026 21:00:13 +0530 Subject: [PATCH 2/7] score a predicted span with the crf marginal --- .../dataset_pipeline/review_ml_phrases.py | 96 +++++++++++++++++++ 1 file changed, 96 insertions(+) diff --git a/etc/scripts/dataset_pipeline/review_ml_phrases.py b/etc/scripts/dataset_pipeline/review_ml_phrases.py index 46fa13e89f..a3c61fa935 100644 --- a/etc/scripts/dataset_pipeline/review_ml_phrases.py +++ b/etc/scripts/dataset_pipeline/review_ml_phrases.py @@ -5,10 +5,25 @@ # written on the model's word alone import json import os +import sys import tempfile +from pathlib import Path import click +# transformers pulls in keras otherwise and blows up on keras 3 +os.environ.setdefault('USE_TF', '0') + +sys.path.insert(0, str(Path(__file__).parent)) + +from train_model import extract_spans +from train_model import first_subword_positions +from train_model import ID2LABEL + +# far enough below the real emission scores to take a label out of the running, +# finite so the forward algorithm never ends up with inf minus inf +PIN_PENALTY = 10000.0 + # tiers a phrase can land in, and the decision each one starts life with AUTO = 'auto' REVIEW = 'review' @@ -114,3 +129,84 @@ def write_review_file(path, records): except Exception: os.unlink(handle.name) raise + + +def span_confidence(crf, word_emissions, tags, mask, free, span): + """How sure the CRF is that this span is tagged the way it decoded it + + Viterbi hands back one best path and no scores, so score that path twice: + once as it stands, once with every label but the decoded one pinned out of + reach at the span's words. The path score is the same in both and cancels, + which leaves log Z(pinned) - log Z(free), the share of the probability mass + held by every path that tags this span this way + """ + start, end = span + pinned = word_emissions.clone() + floor = float(word_emissions.min()) - PIN_PENALTY + + for position in range(start, end + 1): + label = int(tags[0, position]) + keep = float(pinned[0, position, label]) + pinned[0, position] = floor + pinned[0, position, label] = keep + + constrained = crf(pinned, tags, mask=mask, reduction='none') + # detached because the crf parameters carry grad and we only want the number + confidence = float((free - constrained).detach().exp()) + return min(max(confidence, 0.0), 1.0) + + +def tag_and_score(tagger, tokenizer, max_length, words): + """The phrases the tagger predicts for one rule, each with a confidence + + This repeats what add_ml_phrases.predict_phrases does, because that returns + the phrase texts only and PhraseTagger.predict_words offers no way to get + the emissions back out. Calling either as well would mean running the + backbone a second time for every rule + """ + import torch + + encoding = tokenizer( + words, + is_split_into_words=True, + truncation=True, + max_length=max_length, + return_tensors='pt', + ) + positions = first_subword_positions(encoding.word_ids()) + # nothing to tag, so leave the backbone alone + if not positions: + return [], False + + with torch.no_grad(): + emissions = tagger.emissions(encoding['input_ids'], encoding['attention_mask']) + word_emissions = emissions[:, positions] + mask = torch.ones( + word_emissions.shape[:2], + dtype=torch.bool, + device=emissions.device, + ) + decoded = tagger.crf.decode(word_emissions, mask=mask)[0] + tags = torch.tensor([decoded], device=emissions.device) + # the same for every span of this rule, so pay for it once + free = tagger.crf(word_emissions, tags, mask=mask, reduction='none') + + labels = [ID2LABEL.get(int(label), 'O') for label in decoded] + truncated = len(labels) < len(words) + + # two spans can give the same text, keep the one we are surest of + confidences = {} + for start, end in extract_spans(labels): + # extract_spans closes whatever is still open at the end, so on a + # truncated rule that last span is a phrase cut in half + if truncated and end == len(labels) - 1: + continue + + text = ' '.join(words[start:end + 1]) + confidence = span_confidence( + tagger.crf, word_emissions, tags, mask, free, (start, end), + ) + if confidence > confidences.get(text, 0.0): + confidences[text] = confidence + + return list(confidences.items()), truncated From 39c4b76de3ed9baac573e1f2626c4ce5844f747e Mon Sep 17 00:00:00 2001 From: Kaushik Kumar Date: Sat, 15 Aug 2026 21:03:43 +0530 Subject: [PATCH 3/7] add the predict command --- .../dataset_pipeline/review_ml_phrases.py | 177 ++++++++++++++++++ 1 file changed, 177 insertions(+) diff --git a/etc/scripts/dataset_pipeline/review_ml_phrases.py b/etc/scripts/dataset_pipeline/review_ml_phrases.py index a3c61fa935..bae19f4227 100644 --- a/etc/scripts/dataset_pipeline/review_ml_phrases.py +++ b/etc/scripts/dataset_pipeline/review_ml_phrases.py @@ -16,6 +16,14 @@ sys.path.insert(0, str(Path(__file__).parent)) +from licensedcode.required_phrases import find_phrase_spans_in_text +from licensedcode.required_phrases import RequiredPhraseRuleCandidate + +from add_ml_phrases import load_model +from add_ml_phrases import MIN_SINGLE_TOKEN_LEN +from add_ml_phrases import MIN_TOKENS +from add_ml_phrases import select_rules +from add_ml_phrases import words_from_text from train_model import extract_spans from train_model import first_subword_positions from train_model import ID2LABEL @@ -24,6 +32,8 @@ # finite so the forward algorithm never ends up with inf minus inf PIN_PENALTY = 10000.0 +HISTOGRAM_BINS = 20 + # tiers a phrase can land in, and the decision each one starts life with AUTO = 'auto' REVIEW = 'review' @@ -210,3 +220,170 @@ def tag_and_score(tagger, tokenizer, max_length, words): confidences[text] = confidence return list(confidences.items()), truncated + + +def new_predict_counts(): + """What a predict run tallies + + add_ml_phrases.new_counts is the injection tally and half of it means + nothing here, apply reuses that one instead + """ + return dict(rules=0, truncated=0, rejected=0, not_found=0, auto=0, review=0, low=0) + + +def check_thresholds(auto_threshold, review_threshold): + """Both in range and the right way round, before anything expensive starts""" + for name, value in ( + ('--auto-threshold', auto_threshold), + ('--review-threshold', review_threshold), + ): + if not 0.0 <= value <= 1.0: + raise click.ClickException(f'{name} is {value}, it must be between 0 and 1') + + if review_threshold > auto_threshold: + raise click.ClickException( + f'--review-threshold {review_threshold} is above ' + f'--auto-threshold {auto_threshold}, nothing would ever be reviewed' + ) + + +def tier_for(confidence, auto_threshold, review_threshold): + """The tier a confidence lands in and the decision it starts with""" + if confidence >= auto_threshold: + return AUTO, AUTO + if confidence >= review_threshold: + return REVIEW, PENDING + return LOW, DROPPED + + +def is_injectable(rule, phrase, counts): + """True if scancode would take this phrase, counting the ones it would not + + is_good goes first because find_phrase_spans_in_text reads the first token + of the normalized phrase and raises when there is none + """ + candidate = RequiredPhraseRuleCandidate.create(rule.license_expression, phrase) + if not candidate.is_good(rule, MIN_TOKENS, MIN_SINGLE_TOKEN_LEN): + counts['rejected'] += 1 + return False + + # the words came from NFKC normalized text but the markers go into the raw + # text, so check the phrase can still be found there + if not find_phrase_spans_in_text(rule.text, phrase): + counts['not_found'] += 1 + return False + + return True + + +def print_histogram(confidences): + """Where the confidences actually fell + + The thresholds are guesses until someone looks at this + """ + counted = [0] * HISTOGRAM_BINS + for confidence in confidences: + counted[min(int(confidence * HISTOGRAM_BINS), HISTOGRAM_BINS - 1)] += 1 + + width = 1.0 / HISTOGRAM_BINS + click.echo('\nconfidence spread') + for index, count in enumerate(counted): + low = index * width + click.echo(f' {low:.2f} - {low + width:.2f} : {count}') + + +@click.group() +@click.help_option('-h', '--help') +def cli(): + """Review the required phrases the phrase tagger predicts, then inject them""" + + +@cli.command() +@click.option('--model', required=True, + help='Trained model directory, or a huggingface repo id to download') +@click.option('--review-file', required=True, type=click.Path(dir_okay=False), + help='Where to write the predictions, must not exist yet') +@click.option('--license-expression', default=None, + help='Only tag rules for this license expression, example: apache-2.0') +@click.option('--auto-threshold', default=0.95, show_default=True, type=float, + help='Confidence at or above which a phrase skips review, provisional') +@click.option('--review-threshold', default=0.60, show_default=True, type=float, + help='Confidence below which a phrase is dropped, provisional') +@click.option('--limit', default=0, type=int, + help='Stop after this many rules, 0 does all of them') +@click.option('-v', '--verbose', is_flag=True, default=False, + help='Print the phrases predicted for each rule') +@click.help_option('-h', '--help') +def predict(model, review_file, license_expression, auto_threshold, review_threshold, + limit, verbose): + """Predict required phrases and file them for review + + The confidence is how much of its own probability mass the model puts on a + span, not proof the phrase is right. is_good and the review pass are what + keep a bad phrase out of a rule. Both thresholds are starting points until + the confidences of the final checkpoint have been looked at + """ + check_thresholds(auto_threshold, review_threshold) + if os.path.exists(review_file): + raise click.ClickException( + f'{review_file} exists already, move or delete it: predict appends a ' + f'record per rule and a second run would double them up' + ) + + try: + tagger, tokenizer, max_length = load_model(model, hf_token=os.environ.get('HF_TOKEN')) + except ImportError as e: + raise click.ClickException(f'{e}, install etc/requirements-ml.txt') + + selected = select_rules(license_expression=license_expression) + rules = [rule for expression in selected.values() for rule in expression] + click.echo(f'tagging {len(rules)} rules in {len(selected)} license expressions') + if limit: + rules = rules[:limit] + + counts = new_predict_counts() + confidences = [] + + # written as we go, a full pass takes hours and a crash should not cost all of it + with open(review_file, 'w', encoding='utf-8') as out: + for rule in rules: + counts['rules'] += 1 + words = words_from_text(rule.text) + scored, truncated = tag_and_score(tagger, tokenizer, max_length, words) + if truncated: + counts['truncated'] += 1 + + phrases = [] + for text, confidence in scored: + if not is_injectable(rule, text, counts): + continue + + tier, decision = tier_for(confidence, auto_threshold, review_threshold) + counts[tier] += 1 + confidences.append(confidence) + phrases.append(new_phrase(text, round(confidence, 4), tier, decision)) + + if not phrases: + continue + + if verbose: + click.echo(f' {rule.identifier}: {[phrase["text"] for phrase in phrases]}') + + write_review_record(out, new_record(rule, phrases, truncated)) + + click.echo(f"\nrules processed : {counts['rules']}") + click.echo(f" truncated : {counts['truncated']}") + click.echo(f"phrases filed : {counts[AUTO] + counts[REVIEW] + counts[LOW]}") + click.echo(f" auto : {counts[AUTO]}") + click.echo(f" review : {counts[REVIEW]}") + click.echo(f" low : {counts[LOW]}") + click.echo(f" rejected : {counts['rejected']}") + click.echo(f" not found : {counts['not_found']}") + click.echo(f'\nreview file : {review_file}') + + if confidences: + print_histogram(confidences) + + +if __name__ == '__main__': + cli() From 02b499d4a21567f6d7bc0e361a803f024705be0f Mon Sep 17 00:00:00 2001 From: Kaushik Kumar Date: Sat, 15 Aug 2026 21:05:21 +0530 Subject: [PATCH 4/7] preview an injection by marking the rule under dry run --- .../dataset_pipeline/review_ml_phrases.py | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/etc/scripts/dataset_pipeline/review_ml_phrases.py b/etc/scripts/dataset_pipeline/review_ml_phrases.py index bae19f4227..d2bbedd825 100644 --- a/etc/scripts/dataset_pipeline/review_ml_phrases.py +++ b/etc/scripts/dataset_pipeline/review_ml_phrases.py @@ -3,6 +3,7 @@ # a maintainer, apply injects what came back accepted # a wrong required phrase silently drops a real license detection, so nothing is # written on the model's word alone +import difflib import json import os import sys @@ -16,6 +17,9 @@ sys.path.insert(0, str(Path(__file__).parent)) +from licensedcode.models import Rule +from licensedcode.models import rules_data_dir +from licensedcode.required_phrases import add_required_phrase_to_rule from licensedcode.required_phrases import find_phrase_spans_in_text from licensedcode.required_phrases import RequiredPhraseRuleCandidate @@ -292,6 +296,54 @@ def print_histogram(confidences): click.echo(f' {low:.2f} - {low + width:.2f} : {count}') +def load_rule(identifier): + """The rule a record names, or None if its file has gone + + A record names one rule, so read that one file. select_rules would reload + all 36k of them and drop the ones that already have markers, which is + exactly the ones an earlier apply run just marked + """ + rule_file = os.path.join(rules_data_dir, identifier) + if not os.path.exists(rule_file): + return None + return Rule.from_file(rule_file, is_builtin=True) + + +def preview_injection(rule, phrase): + """What apply would write for this phrase, and whether it would write at all + + Done by letting add_required_phrase_to_rule mark the rule under dry_run and + then putting the rule back as it was. Working the spans out here instead + would miss that it marks every non overlapping one, and that it refuses when + a span sits on an existing marker or an ignorable. source is restored with + the text so whatever we pass for it never lands anywhere + """ + text, source = rule.text, rule.source + changed = add_required_phrase_to_rule( + rule=rule, + required_phrase=phrase, + source='ml_model', + dry_run=True, + ) + preview = rule.text + rule.text, rule.source = text, source + return changed, preview + + +def render_diff(identifier, before, after): + """The injection as a unified diff""" + lines = difflib.unified_diff( + before.splitlines(keepends=True), + after.splitlines(keepends=True), + fromfile=f'a/{identifier}', + tofile=f'b/{identifier}', + ) + colours = {'+': 'green', '-': 'red', '@': 'cyan'} + for line in lines: + line = line.rstrip('\n') + click.echo(click.style(line, fg=colours.get(line[:1]))) + + @click.group() @click.help_option('-h', '--help') def cli(): From 17456095f219411b9bcd76e82ff7c9a3a7c2cfa2 Mon Sep 17 00:00:00 2001 From: Kaushik Kumar Date: Sat, 15 Aug 2026 21:08:36 +0530 Subject: [PATCH 5/7] add the interactive review command --- .../dataset_pipeline/review_ml_phrases.py | 150 ++++++++++++++++++ 1 file changed, 150 insertions(+) diff --git a/etc/scripts/dataset_pipeline/review_ml_phrases.py b/etc/scripts/dataset_pipeline/review_ml_phrases.py index d2bbedd825..b47b3489da 100644 --- a/etc/scripts/dataset_pipeline/review_ml_phrases.py +++ b/etc/scripts/dataset_pipeline/review_ml_phrases.py @@ -437,5 +437,155 @@ def predict(model, review_file, license_expression, auto_threshold, review_thres print_histogram(confidences) +def new_review_counts(): + """What a review session tallies""" + return dict(approved=0, edited=0, rejected=0, nothing=0, stale=0) + + +def show_phrase(record, phrase, rule, preview): + """The rule, the phrase and what marking it would do to the text""" + click.echo(f"\n{record['identifier']} {record['license_expression']}") + click.echo( + f"phrase: {phrase['text']}" + f" {phrase['confidence']:.0%} {phrase['tier']}" + ) + render_diff(record['identifier'], rule.text, preview) + + +def edit_phrase(rule, phrase): + """Retype a phrase until one passes the gates, False to go back + + Both gates run here so a phrase that scancode would refuse cannot be + approved, and is_good runs first for the same reason it does in predict + """ + click.echo('\nrule text') + click.echo(rule.text) + + while True: + click.echo(f"\ncurrently: {phrase['text']}") + text = click.prompt( + 'phrase, empty to go back', + default='', + show_default=False, + ).strip() + if not text: + return False + + candidate = RequiredPhraseRuleCandidate.create(rule.license_expression, text) + if not candidate.is_good(rule, MIN_TOKENS, MIN_SINGLE_TOKEN_LEN): + click.echo('is_good refused it: too short, or a stopword in a short phrase') + continue + + if not find_phrase_spans_in_text(rule.text, text): + click.echo('not found in the rule text, it has to be word for word') + continue + + changed, preview = preview_injection(rule, text) + if not changed: + click.echo('nothing to mark, it sits on an existing marker or an ignorable') + continue + + render_diff(rule.identifier, rule.text, preview) + phrase['text'] = text + phrase['decision'] = APPROVED + return True + + +def ask_phrase(rule, phrase, counts): + """Take a decision on one phrase, False if the maintainer wants to stop""" + while True: + answer = click.prompt( + '[y] approve [n] reject [e] edit [q] quit', + default='', + show_default=False, + ).strip().lower() + + if answer == 'y': + phrase['decision'] = APPROVED + counts['approved'] += 1 + return True + + if answer == 'n': + phrase['decision'] = REJECTED + counts['rejected'] += 1 + return True + + if answer == 'q': + return False + + if answer == 'e': + if edit_phrase(rule, phrase): + counts['approved'] += 1 + counts['edited'] += 1 + return True + continue + + click.echo('answer y, n, e or q') + + +def walk_records(records, review_file, counts): + """Ask about every pending phrase, saving after each decision""" + for record in records: + pending = [ + phrase for phrase in record['phrases'] + if phrase['decision'] == PENDING + ] + if not pending: + continue + + rule = load_rule(record['identifier']) + if rule is None: + counts['stale'] += 1 + continue + + for phrase in pending: + changed, preview = preview_injection(rule, phrase['text']) + # covers a phrase that is no longer there, one that overlaps a + # marker or an ignorable, and one that would rewrite the ignorables + if not changed: + counts['nothing'] += 1 + continue + + show_phrase(record, phrase, rule, preview) + if not ask_phrase(rule, phrase, counts): + return + + # rewritten now rather than at the end, so a stop here costs nothing + write_review_file(review_file, records) + + +@cli.command() +@click.option('--review-file', required=True, + type=click.Path(exists=True, dir_okay=False), + help='Review file written by predict') +@click.help_option('-h', '--help') +def review(review_file): + """Approve, reject or edit the phrases that need a decision + + Only the review tier is asked about. Stopping is safe, every decision is + written before the next phrase comes up, so a rerun carries on where this + one left off + """ + records = read_review_file(review_file) + counts = new_review_counts() + + waiting = sum( + 1 for record in records for phrase in record['phrases'] + if phrase['decision'] == PENDING + ) + if not waiting: + click.echo('nothing left to review') + return + + click.echo(f'{waiting} phrases waiting') + walk_records(records, review_file, counts) + + click.echo(f"\napproved : {counts['approved']}") + click.echo(f" edited : {counts['edited']}") + click.echo(f"rejected : {counts['rejected']}") + click.echo(f"nothing to add : {counts['nothing']}") + click.echo(f"stale rules : {counts['stale']}") + + if __name__ == '__main__': cli() From 94106d9dd8a3647fa70960f25a082b9d809c6191 Mon Sep 17 00:00:00 2001 From: Kaushik Kumar Date: Sat, 15 Aug 2026 21:10:07 +0530 Subject: [PATCH 6/7] add the apply command --- .../dataset_pipeline/review_ml_phrases.py | 66 +++++++++++++++++++ 1 file changed, 66 insertions(+) diff --git a/etc/scripts/dataset_pipeline/review_ml_phrases.py b/etc/scripts/dataset_pipeline/review_ml_phrases.py index b47b3489da..0dff0929bd 100644 --- a/etc/scripts/dataset_pipeline/review_ml_phrases.py +++ b/etc/scripts/dataset_pipeline/review_ml_phrases.py @@ -23,9 +23,11 @@ from licensedcode.required_phrases import find_phrase_spans_in_text from licensedcode.required_phrases import RequiredPhraseRuleCandidate +from add_ml_phrases import inject from add_ml_phrases import load_model from add_ml_phrases import MIN_SINGLE_TOKEN_LEN from add_ml_phrases import MIN_TOKENS +from add_ml_phrases import new_counts from add_ml_phrases import select_rules from add_ml_phrases import words_from_text from train_model import extract_spans @@ -587,5 +589,69 @@ def review(review_file): click.echo(f"stale rules : {counts['stale']}") +def accepted_phrases(record): + """The phrases of one record to inject, longest first""" + texts = [ + phrase['text'] for phrase in record['phrases'] + if phrase['decision'] in (APPROVED, AUTO) + ] + return sorted(texts, key=phrase_sort_key) + + +@cli.command() +@click.option('--review-file', required=True, + type=click.Path(exists=True, dir_okay=False), + help='Review file to inject from') +@click.option('--dry-run', is_flag=True, default=False, + help='Check the phrases but do not save any rule') +@click.option('-v', '--verbose', is_flag=True, default=False, + help='Print the phrases injected into each rule') +@click.help_option('-h', '--help') +def apply(review_file, dry_run, verbose): + """Inject the approved and auto approved phrases + + Every phrase goes through is_good and find_phrase_spans_in_text again on the + way in, against the rule as it is on disk, so an old review file can only + end up doing less than it says, never something wrong + """ + records = read_review_file(review_file) + work = [(record, accepted_phrases(record)) for record in records] + work = [(record, phrases) for record, phrases in work if phrases] + + if not work: + click.echo('nothing accepted to apply') + return + + counts = new_counts() + counts['stale'] = 0 + + for record, phrases in work: + counts['rules'] += 1 + # read fresh, the file may have moved on since predict ran + rule = load_rule(record['identifier']) + if rule is None: + counts['stale'] += 1 + continue + + if verbose: + click.echo(f"{record['identifier']}: {phrases}") + + if inject(rule, phrases, counts, dry_run=dry_run, verbose=verbose): + counts['written'] += 1 + + click.echo(f"\nrules : {counts['rules']}") + click.echo(f"phrases injected : {counts['injected']}") + click.echo(f" rejected : {counts['rejected']}") + click.echo(f" not found : {counts['not_found']}") + click.echo(f" nothing to add : {counts['skipped']}") + click.echo(f"stale rules : {counts['stale']}") + click.echo(f"rules written : {counts['written']}") + + if dry_run: + click.echo('dry run, no rules were saved') + elif counts['written']: + click.echo('run scancode-reindex-licenses to pick up the new required phrases') + + if __name__ == '__main__': cli() From c721c08d8edf0cb03e31f8fa50249d35d41f485d Mon Sep 17 00:00:00 2001 From: Kaushik Kumar Date: Sat, 15 Aug 2026 21:46:39 +0530 Subject: [PATCH 7/7] add tests for review_ml_phrases --- .../test_review_ml_phrases.py | 820 ++++++++++++++++++ 1 file changed, 820 insertions(+) create mode 100644 etc/scripts/dataset_pipeline/test_review_ml_phrases.py diff --git a/etc/scripts/dataset_pipeline/test_review_ml_phrases.py b/etc/scripts/dataset_pipeline/test_review_ml_phrases.py new file mode 100644 index 0000000000..783eb1f19d --- /dev/null +++ b/etc/scripts/dataset_pipeline/test_review_ml_phrases.py @@ -0,0 +1,820 @@ +# tests for review_ml_phrases.py +# no model and no network, the tagger is stubbed and the crf is built here +# etc is in pytest's norecursedirs so these do not run in CI, run them by path: +# pytest etc/scripts/dataset_pipeline/test_review_ml_phrases.py +import itertools +import json +import os +import sys +from pathlib import Path + +import click +import pytest +from click.testing import CliRunner + +sys.path.insert(0, str(Path(__file__).parent)) +import review_ml_phrases +from review_ml_phrases import accepted_phrases +from review_ml_phrases import APPROVED +from review_ml_phrases import AUTO +from review_ml_phrases import check_thresholds +from review_ml_phrases import cli +from review_ml_phrases import DROPPED +from review_ml_phrases import is_injectable +from review_ml_phrases import LOW +from review_ml_phrases import new_phrase +from review_ml_phrases import new_predict_counts +from review_ml_phrases import new_record +from review_ml_phrases import PENDING +from review_ml_phrases import phrase_sort_key +from review_ml_phrases import preview_injection +from review_ml_phrases import read_review_file +from review_ml_phrases import REJECTED +from review_ml_phrases import REVIEW +from review_ml_phrases import span_confidence +from review_ml_phrases import tag_and_score +from review_ml_phrases import tier_for +from review_ml_phrases import write_review_file +from train_model import LABEL2ID +from train_model import LABELS + +import licensedcode.required_phrases as required_phrases +from licensedcode.models import Rule + +TEXT = 'Granted under the MIT License to do things with this software' + + +class FakeEncoding(dict): + + def __init__(self, word_ids): + super().__init__() + self._word_ids = word_ids + self['input_ids'] = [[0] * len(word_ids)] + self['attention_mask'] = [[1] * len(word_ids)] + + def word_ids(self): + return self._word_ids + + +class FakeTokenizer: + """CLS, one subword per word, SEP, cut off at max_length""" + + def __call__(self, words, max_length=512, **kwargs): + word_ids = [None] + list(range(len(words))) + [None] + return FakeEncoding(word_ids[:max_length]) + + +class StubTagger: + """Puts the given labels on the given word offsets + + The crf is real but zeroed, so decoding is a plain argmax over the emissions + and the tags a test asks for are the tags it gets + """ + + def __init__(self, tagged): + import torch + from torchcrf import CRF + + self.tagged = tagged + self.crf = CRF(len(LABELS), batch_first=True) + with torch.no_grad(): + for param in self.crf.parameters(): + param.zero_() + + def emissions(self, input_ids, attention_mask): + import torch + + width = len(input_ids[0]) + scores = torch.zeros((1, width, len(LABELS))) + for offset, label in self.tagged.items(): + # subword offset + 1 because of the leading CLS + scores[0, offset + 1, LABEL2ID[label]] = 9.0 + return scores + + +def make_rule(text=TEXT, identifier='mit_1.RULE'): + rule = Rule( + license_expression='mit', + identifier=identifier, + text=text, + is_license_reference=True, + relevance=100, + ) + rule.source = None + return rule + + +def use_tmp_rules(monkeypatch, tmp_path): + """Point both rules_data_dir bindings at tmp_path + + review_ml_phrases reads its own in load_rule, and required_phrases keeps a + separate one that add_required_phrase_to_rule writes through, so patching + only the first would send real writes into the repo + """ + monkeypatch.setattr(review_ml_phrases, 'rules_data_dir', str(tmp_path)) + monkeypatch.setattr(required_phrases, 'rules_data_dir', str(tmp_path)) + + +def write_rule(tmp_path, text=TEXT, identifier='mit_1.RULE'): + rule = make_rule(text=text, identifier=identifier) + rule.dump(str(tmp_path)) + return rule + + +def rule_digests(tmp_path): + return { + path.name: path.read_bytes() + for path in sorted(Path(tmp_path).glob('*.RULE')) + } + + +def make_review_file(tmp_path, phrases, identifier='mit_1.RULE', text=TEXT): + """A rule on disk and a review file naming it""" + rule = write_rule(tmp_path, text=text, identifier=identifier) + path = tmp_path / 'review.jsonl' + write_review_file(str(path), [new_record(rule, phrases, False)]) + return str(path) + + +def decisions(path): + return [ + (phrase['text'], phrase['decision']) + for record in read_review_file(path) + for phrase in record['phrases'] + ] + + +def build_crf(num_tags, seed=7): + """A small crf with real transitions, in float64 so the maths is exact""" + import torch + from torchcrf import CRF + + torch.manual_seed(seed) + crf = CRF(num_tags, batch_first=True).to(torch.float64) + with torch.no_grad(): + crf.transitions.copy_(torch.randn(num_tags, num_tags).to(torch.float64)) + crf.start_transitions.copy_(torch.randn(num_tags).to(torch.float64)) + crf.end_transitions.copy_(torch.randn(num_tags).to(torch.float64)) + return crf + + +def decode_once(crf, emissions): + """The best path, its mask and its log likelihood""" + import torch + + mask = torch.ones(emissions.shape[:2], dtype=torch.bool) + tags = torch.tensor([crf.decode(emissions, mask=mask)[0]]) + free = crf(emissions, tags, mask=mask, reduction='none') + return tags, mask, free + + +class TestPhraseSortKey: + + def test_longest_first(self): + phrases = ['mit', 'mit license', 'a'] + assert sorted(phrases, key=phrase_sort_key) == ['mit license', 'mit', 'a'] + + def test_equal_length_goes_alphabetical(self): + assert sorted(['bbb', 'aaa'], key=phrase_sort_key) == ['aaa', 'bbb'] + + +class TestReviewFile: + + def test_round_trip_is_the_identity(self, tmp_path): + # Feature: ml-phrase-review-cli, Property 3: review file round trip is the identity + path = str(tmp_path / 'review.jsonl') + # two phrases of equal length, the case an unstable sort would break + records = [new_record(make_rule(), [ + new_phrase('bbb', 0.5, LOW, DROPPED), + new_phrase('aaa', 0.4, LOW, DROPPED), + new_phrase('MIT License', 0.98, AUTO, AUTO), + new_phrase('do things', 0.8, REVIEW, PENDING), + ], False)] + + write_review_file(path, records) + first = Path(path).read_bytes() + write_review_file(path, read_review_file(path)) + + assert Path(path).read_bytes() == first + + def test_phrases_are_stored_longest_first(self, tmp_path): + path = str(tmp_path / 'review.jsonl') + write_review_file(path, [new_record(make_rule(), [ + new_phrase('mit', 0.7, REVIEW, PENDING), + new_phrase('mit license', 0.9, REVIEW, PENDING), + ], False)]) + + texts = [phrase['text'] for phrase in read_review_file(path)[0]['phrases']] + assert texts == ['mit license', 'mit'] + + def test_a_record_carries_the_rule_and_the_truncated_flag(self): + record = new_record(make_rule(), [new_phrase('mit license', 0.9, AUTO, AUTO)], True) + assert record['identifier'] == 'mit_1.RULE' + assert record['license_expression'] == 'mit' + assert record['truncated'] is True + + def test_an_edit_keeps_the_predicted_text(self): + phrase = new_phrase('MIT License', 0.9, REVIEW, PENDING) + phrase['text'] = 'MIT License to' + assert phrase['predicted_text'] == 'MIT License' + + def test_blank_lines_are_skipped(self, tmp_path): + path = tmp_path / 'review.jsonl' + record = new_record(make_rule(), [new_phrase('mit license', 0.9, AUTO, AUTO)], False) + path.write_text(json.dumps(record) + '\n\n', encoding='utf-8') + assert len(read_review_file(str(path))) == 1 + + def test_a_bad_line_names_its_number(self, tmp_path): + path = tmp_path / 'review.jsonl' + path.write_text('{}\nnot json\n', encoding='utf-8') + with pytest.raises(click.ClickException) as caught: + read_review_file(str(path)) + assert 'line 1' in str(caught.value) + + def test_a_missing_phrase_field_names_its_number(self, tmp_path): + path = tmp_path / 'review.jsonl' + record = new_record(make_rule(), [new_phrase('mit license', 0.9, AUTO, AUTO)], False) + del record['phrases'][0]['confidence'] + path.write_text(json.dumps(record) + '\n', encoding='utf-8') + with pytest.raises(click.ClickException) as caught: + read_review_file(str(path)) + assert "line 1" in str(caught.value) and 'confidence' in str(caught.value) + + def test_a_failed_write_keeps_the_previous_file(self, tmp_path, monkeypatch): + path = str(tmp_path / 'review.jsonl') + records = [new_record(make_rule(), [new_phrase('mit license', 0.9, AUTO, AUTO)], False)] + write_review_file(path, records) + before = Path(path).read_bytes() + + def failing_replace(src, dst): + raise OSError('no') + + monkeypatch.setattr(os, 'replace', failing_replace) + with pytest.raises(OSError): + write_review_file(path, records) + + assert Path(path).read_bytes() == before + assert list(tmp_path.glob('*.tmp')) == [] + + +class TestSpanConfidence: + + def test_matches_the_brute_force_marginal(self): + # Feature: ml-phrase-review-cli, Property 1: the span marginal is the true marginal + import torch + + num_tags = len(LABELS) + length = 5 + crf = build_crf(num_tags) + torch.manual_seed(3) + emissions = torch.randn(1, length, num_tags).to(torch.float64) + tags, mask, free = decode_once(crf, emissions) + + # every possible path, scored in one call + paths = torch.tensor(list(itertools.product(range(num_tags), repeat=length))) + probabilities = crf( + emissions.expand(len(paths), -1, -1), + paths, + mask=mask.expand(len(paths), -1), + reduction='none', + ).detach().exp() + + for span in [(0, 0), (2, 2), (1, 3), (3, 4), (0, length - 1)]: + start, end = span + matching = (paths[:, start:end + 1] == tags[0, start:end + 1]).all(dim=1) + expected = float(probabilities[matching].sum() / probabilities.sum()) + got = span_confidence(crf, emissions, tags, mask, free, span) + assert abs(got - expected) < 1e-12 + assert 0.0 <= got <= 1.0 + + def test_scoring_one_span_does_not_disturb_another(self): + # Feature: ml-phrase-review-cli, Property 2: scoring one span does not disturb another + import torch + + num_tags = len(LABELS) + crf = build_crf(num_tags) + torch.manual_seed(11) + emissions = torch.randn(1, 6, num_tags).to(torch.float64) + tags, mask, free = decode_once(crf, emissions) + spans = [(0, 1), (2, 2), (4, 5)] + + forwards = [span_confidence(crf, emissions, tags, mask, free, s) for s in spans] + backwards = [ + span_confidence(crf, emissions, tags, mask, free, s) + for s in reversed(spans) + ] + + assert forwards == list(reversed(backwards)) + + def test_the_emissions_are_left_alone(self): + import torch + + crf = build_crf(len(LABELS)) + emissions = torch.randn(1, 4, len(LABELS)).to(torch.float64) + original = emissions.clone() + tags, mask, free = decode_once(crf, emissions) + + span_confidence(crf, emissions, tags, mask, free, (1, 2)) + + assert torch.equal(emissions, original) + + def test_large_negative_emissions_do_not_give_nan(self): + import torch + + crf = build_crf(len(LABELS)) + emissions = torch.full((1, 4, len(LABELS)), -1e4, dtype=torch.float64) + tags, mask, free = decode_once(crf, emissions) + + confidence = span_confidence(crf, emissions, tags, mask, free, (0, 3)) + + assert confidence == confidence + assert 0.0 <= confidence <= 1.0 + + +class TestCheckThresholds: + + def test_the_defaults_are_fine(self): + check_thresholds(0.95, 0.60) + + def test_review_above_auto_is_refused(self): + with pytest.raises(click.ClickException): + check_thresholds(0.5, 0.9) + + @pytest.mark.parametrize('auto, review', [(1.5, 0.6), (0.95, -0.1)]) + def test_out_of_range_is_refused(self, auto, review): + with pytest.raises(click.ClickException): + check_thresholds(auto, review) + + +class TestTierFor: + + @pytest.mark.parametrize('confidence, expected', [ + (1.0, (AUTO, AUTO)), + (0.95, (AUTO, AUTO)), + (0.9499, (REVIEW, PENDING)), + (0.60, (REVIEW, PENDING)), + (0.5999, (LOW, DROPPED)), + (0.0, (LOW, DROPPED)), + ]) + def test_the_boundaries(self, confidence, expected): + assert tier_for(confidence, 0.95, 0.60) == expected + + +class TestTagAndScore: + + def test_one_phrase(self): + tagger = StubTagger({3: 'B-REQ', 4: 'E-REQ'}) + scored, truncated = tag_and_score(tagger, FakeTokenizer(), 512, TEXT.split()) + assert [text for text, _ in scored] == ['MIT License'] + assert not truncated + + def test_two_phrases(self): + tagger = StubTagger({0: 'S-REQ', 3: 'B-REQ', 4: 'E-REQ'}) + scored, _ = tag_and_score(tagger, FakeTokenizer(), 512, TEXT.split()) + assert sorted(text for text, _ in scored) == ['Granted', 'MIT License'] + + def test_the_same_text_twice_is_deduped(self): + tagger = StubTagger({0: 'S-REQ', 2: 'S-REQ'}) + scored, _ = tag_and_score(tagger, FakeTokenizer(), 512, ['MIT', 'x', 'MIT']) + assert [text for text, _ in scored] == ['MIT'] + + def test_a_span_cut_by_truncation_is_dropped(self): + tagger = StubTagger({2: 'B-REQ', 3: 'I-REQ'}) + scored, truncated = tag_and_score(tagger, FakeTokenizer(), 5, TEXT.split()) + assert scored == [] + assert truncated + + def test_nothing_tagged(self): + tagger = StubTagger({}) + assert tag_and_score(tagger, FakeTokenizer(), 512, TEXT.split()) == ([], False) + + def test_no_words_never_reaches_the_backbone(self): + class Exploding(StubTagger): + def emissions(self, input_ids, attention_mask): + raise AssertionError('there is nothing to tag') + + assert tag_and_score(Exploding({}), FakeTokenizer(), 512, []) == ([], False) + + +class TestIsInjectable: + + def test_a_good_phrase(self): + counts = new_predict_counts() + assert is_injectable(make_rule(), 'MIT License', counts) + assert counts['rejected'] == 0 and counts['not_found'] == 0 + + def test_is_good_refuses_a_short_phrase(self): + counts = new_predict_counts() + assert not is_injectable(make_rule(), 'is', counts) + assert counts['rejected'] == 1 + + def test_is_good_refuses_text_with_no_tokens(self): + # find_phrase_spans_in_text would raise on this, is_good has to run first + counts = new_predict_counts() + assert not is_injectable(make_rule(), '///', counts) + assert counts['rejected'] == 1 + assert counts['not_found'] == 0 + + def test_a_phrase_that_is_not_in_the_rule(self): + counts = new_predict_counts() + assert not is_injectable(make_rule(), 'Apache License', counts) + assert counts['not_found'] == 1 + + +class TestPreviewInjection: + + @pytest.mark.parametrize('text, phrase, expected', [ + ( + 'Granted under the MIT License to do things', + 'MIT License', + 'Granted under the {{MIT License}} to do things', + ), + ( + # marked at both places, which a one span preview would miss + 'The MIT License applies. See the MIT License text.', + 'MIT License', + 'The {{MIT License}} applies. See the {{MIT License}} text.', + ), + ]) + def test_the_preview_is_what_apply_writes(self, tmp_path, monkeypatch, text, phrase, expected): + # Feature: ml-phrase-review-cli, Property 4: the preview is what apply writes + use_tmp_rules(monkeypatch, tmp_path) + rule = write_rule(tmp_path, text=text) + + changed, preview = preview_injection(rule, phrase) + assert changed + assert preview == expected + # and the rule object is as it was + assert rule.text == text + assert rule.source is None + + # what really gets written matches + from licensedcode.required_phrases import add_required_phrase_to_rule + fresh = review_ml_phrases.load_rule('mit_1.RULE') + add_required_phrase_to_rule(fresh, phrase, source='ml_model', dry_run=False) + assert review_ml_phrases.load_rule('mit_1.RULE').text == preview + + def test_a_phrase_that_cannot_be_marked(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + rule = write_rule(tmp_path) + changed, preview = preview_injection(rule, 'Apache License') + assert not changed + assert preview == rule.text + + def test_a_missing_rule_file(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + assert review_ml_phrases.load_rule('gone_1.RULE') is None + + +def run_predict(monkeypatch, tmp_path, tagger, rules, *extra): + """Drive predict with a stubbed model and a fixed set of rules""" + monkeypatch.setattr( + review_ml_phrases, 'load_model', + lambda model, hf_token=None: (tagger, FakeTokenizer(), 512), + ) + monkeypatch.setattr( + review_ml_phrases, 'select_rules', + lambda license_expression=None: {'mit': rules}, + ) + path = str(tmp_path / 'review.jsonl') + result = CliRunner().invoke(cli, [ + 'predict', '--model', 'stub', '--review-file', path, + ] + list(extra)) + return result, path + + +class TestPredict: + + def test_writes_a_record_per_rule_with_phrases(self, tmp_path, monkeypatch): + tagger = StubTagger({3: 'B-REQ', 4: 'E-REQ'}) + rules = [make_rule(identifier='mit_1.RULE'), make_rule(identifier='mit_2.RULE')] + result, path = run_predict(monkeypatch, tmp_path, tagger, rules) + + assert result.exit_code == 0 + records = read_review_file(path) + assert [record['identifier'] for record in records] == ['mit_1.RULE', 'mit_2.RULE'] + assert records[0]['phrases'][0]['text'] == 'MIT License' + assert records[0]['phrases'][0]['decision'] == AUTO + + def test_a_rule_with_nothing_to_file_gets_no_record(self, tmp_path, monkeypatch): + result, path = run_predict( + monkeypatch, tmp_path, StubTagger({}), [make_rule()], + ) + assert result.exit_code == 0 + assert read_review_file(path) == [] + + def test_the_tiers_follow_the_thresholds(self, tmp_path, monkeypatch): + tagger = StubTagger({3: 'B-REQ', 4: 'E-REQ'}) + # every confidence lands under an auto threshold of 1.0 + result, path = run_predict( + monkeypatch, tmp_path, tagger, [make_rule()], '--auto-threshold', '1.0', + ) + assert result.exit_code == 0 + phrase = read_review_file(path)[0]['phrases'][0] + assert phrase['tier'] == REVIEW + assert phrase['decision'] == PENDING + + def test_limit_stops_early(self, tmp_path, monkeypatch): + tagger = StubTagger({3: 'B-REQ', 4: 'E-REQ'}) + rules = [make_rule(identifier=f'mit_{n}.RULE') for n in range(5)] + result, path = run_predict(monkeypatch, tmp_path, tagger, rules, '--limit', '2') + + assert result.exit_code == 0 + assert len(read_review_file(path)) == 2 + + def test_the_counts_add_up(self, tmp_path, monkeypatch): + # Feature: ml-phrase-review-cli, Property 6: predict counts conserve + # one phrase is good, one is a single short token is_good will refuse + tagger = StubTagger({3: 'B-REQ', 4: 'E-REQ', 0: 'S-REQ'}) + result, path = run_predict(monkeypatch, tmp_path, tagger, [make_rule()]) + + filed = sum(len(record['phrases']) for record in read_review_file(path)) + rejected = int(result.output.split('rejected : ')[1].split('\n')[0]) + not_found = int(result.output.split('not found : ')[1].split('\n')[0]) + assert filed + rejected + not_found == 2 + + def test_the_histogram_is_left_out_when_nothing_was_filed(self, tmp_path, monkeypatch): + result, _ = run_predict(monkeypatch, tmp_path, StubTagger({}), [make_rule()]) + assert 'confidence spread' not in result.output + + def test_a_review_file_that_exists_is_refused(self, tmp_path, monkeypatch): + def exploding_load(model, hf_token=None): + raise AssertionError('the model must not be loaded') + + monkeypatch.setattr(review_ml_phrases, 'load_model', exploding_load) + path = tmp_path / 'review.jsonl' + path.write_text('keep me\n', encoding='utf-8') + + result = CliRunner().invoke(cli, [ + 'predict', '--model', 'stub', '--review-file', str(path), + ]) + + assert result.exit_code != 0 + assert path.read_text(encoding='utf-8') == 'keep me\n' + + def test_bad_thresholds_never_load_the_model(self, tmp_path, monkeypatch): + def exploding_load(model, hf_token=None): + raise AssertionError('the model must not be loaded') + + monkeypatch.setattr(review_ml_phrases, 'load_model', exploding_load) + path = tmp_path / 'review.jsonl' + + result = CliRunner().invoke(cli, [ + 'predict', '--model', 'stub', '--review-file', str(path), + '--review-threshold', '0.99', + ]) + + assert result.exit_code != 0 + assert not path.exists() + + def test_a_missing_ml_dependency_is_reported(self, tmp_path, monkeypatch): + def no_safetensors(model, hf_token=None): + raise ImportError('No module named safetensors') + + monkeypatch.setattr(review_ml_phrases, 'load_model', no_safetensors) + result = CliRunner().invoke(cli, [ + 'predict', '--model', 'stub', + '--review-file', str(tmp_path / 'review.jsonl'), + ]) + + assert result.exit_code != 0 + assert 'safetensors' in result.output + assert 'requirements-ml.txt' in result.output + + +def run_review(path, keys): + return CliRunner().invoke(cli, ['review', '--review-file', path], input=keys) + + +class TestReview: + + def two_pending(self, tmp_path): + return make_review_file(tmp_path, [ + new_phrase('MIT License', 0.82, REVIEW, PENDING), + new_phrase('do things', 0.71, REVIEW, PENDING), + ]) + + def test_approving_and_rejecting(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = self.two_pending(tmp_path) + + result = run_review(path, 'y\nn\n') + + assert result.exit_code == 0 + assert decisions(path) == [('MIT License', APPROVED), ('do things', REJECTED)] + + def test_the_diff_and_the_rule_are_shown(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = self.two_pending(tmp_path) + + result = run_review(path, 'y\ny\n') + + assert 'mit_1.RULE mit' in result.output + assert '+Granted under the {{MIT License}} to do things' in result.output + assert '82%' in result.output + + def test_an_unknown_key_asks_again(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = self.two_pending(tmp_path) + + result = run_review(path, 'x\ny\ny\n') + + assert 'answer y, n, e or q' in result.output + assert decisions(path) == [('MIT License', APPROVED), ('do things', APPROVED)] + + def test_quitting_leaves_the_rest_pending(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = self.two_pending(tmp_path) + + result = run_review(path, 'y\nq\n') + + assert result.exit_code == 0 + assert decisions(path) == [('MIT License', APPROVED), ('do things', PENDING)] + + def test_a_rerun_carries_on_where_it_stopped(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = self.two_pending(tmp_path) + run_review(path, 'y\nq\n') + + result = run_review(path, 'n\n') + + assert '1 phrases waiting' in result.output + assert decisions(path) == [('MIT License', APPROVED), ('do things', REJECTED)] + + def test_nothing_pending(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.98, AUTO, AUTO)]) + + result = run_review(path, '') + + assert result.exit_code == 0 + assert 'nothing left to review' in result.output + + def test_a_stale_record_is_skipped(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = self.two_pending(tmp_path) + os.unlink(str(tmp_path / 'mit_1.RULE')) + + result = run_review(path, '') + + assert result.exit_code == 0 + assert 'stale rules : 1' in result.output + assert decisions(path) == [('MIT License', PENDING), ('do things', PENDING)] + + def test_a_phrase_that_cannot_be_marked_is_not_asked_about(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('Apache License', 0.8, REVIEW, PENDING)]) + + result = run_review(path, '') + + assert 'nothing to add : 1' in result.output + assert decisions(path) == [('Apache License', PENDING)] + + def test_an_accepted_edit(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.82, REVIEW, PENDING)]) + + result = run_review(path, 'e\nMIT License to\n') + + assert result.exit_code == 0 + phrase = read_review_file(path)[0]['phrases'][0] + assert phrase['text'] == 'MIT License to' + assert phrase['predicted_text'] == 'MIT License' + assert phrase['confidence'] == 0.82 + assert phrase['decision'] == APPROVED + assert 'edited : 1' in result.output + + def test_an_edit_refused_by_is_good(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.82, REVIEW, PENDING)]) + + result = run_review(path, 'e\nis\n\ny\n') + + assert 'is_good refused it' in result.output + assert decisions(path) == [('MIT License', APPROVED)] + + def test_an_edit_that_is_not_in_the_rule(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.82, REVIEW, PENDING)]) + + result = run_review(path, 'e\nApache License\n\nn\n') + + assert 'not found in the rule text' in result.output + assert decisions(path) == [('MIT License', REJECTED)] + + def test_an_empty_edit_goes_back_to_the_keys(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.82, REVIEW, PENDING)]) + + result = run_review(path, 'e\n\ny\n') + + phrase = read_review_file(path)[0]['phrases'][0] + assert phrase['text'] == 'MIT License' + assert phrase['decision'] == APPROVED + assert 'edited : 0' in result.output + + +def run_apply(path, *extra): + return CliRunner().invoke(cli, ['apply', '--review-file', path] + list(extra)) + + +class TestApply: + + def test_dry_run_writes_nothing(self, tmp_path, monkeypatch): + # Feature: ml-phrase-review-cli, Property 7: nothing is written under dry-run + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.98, AUTO, AUTO)]) + before = rule_digests(tmp_path) + + result = run_apply(path, '--dry-run') + + assert result.exit_code == 0 + assert rule_digests(tmp_path) == before + assert 'dry run, no rules were saved' in result.output + + def test_a_real_write_marks_the_rule(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.98, AUTO, AUTO)]) + + result = run_apply(path) + + assert result.exit_code == 0 + marked = review_ml_phrases.load_rule('mit_1.RULE') + assert '{{MIT License}}' in marked.text + assert marked.source == 'ml_model' + assert 'scancode-reindex-licenses' in result.output + + def test_running_it_twice_changes_nothing(self, tmp_path, monkeypatch): + # Feature: ml-phrase-review-cli, Property 5: apply is idempotent + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.98, AUTO, AUTO)]) + run_apply(path) + after_first = rule_digests(tmp_path) + + result = run_apply(path) + + assert rule_digests(tmp_path) == after_first + assert 'phrases injected : 0' in result.output + + def test_only_approved_and_auto_go_in(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [ + new_phrase('MIT License', 0.98, AUTO, AUTO), + new_phrase('do things', 0.80, REVIEW, APPROVED), + new_phrase('with this software', 0.75, REVIEW, REJECTED), + new_phrase('under the', 0.70, REVIEW, PENDING), + ]) + + run_apply(path) + + text = review_ml_phrases.load_rule('mit_1.RULE').text + assert '{{MIT License}}' in text + assert '{{do things}}' in text + assert '{{with this software}}' not in text + + def test_nothing_accepted(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.4, LOW, DROPPED)]) + before = rule_digests(tmp_path) + + result = run_apply(path) + + assert result.exit_code == 0 + assert 'nothing accepted to apply' in result.output + assert rule_digests(tmp_path) == before + + def test_a_stale_record_is_counted(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.98, AUTO, AUTO)]) + os.unlink(str(tmp_path / 'mit_1.RULE')) + + result = run_apply(path) + + assert result.exit_code == 0 + assert 'stale rules : 1' in result.output + + def test_a_rule_that_already_has_a_marker_takes_the_next_phrase(self, tmp_path, monkeypatch): + use_tmp_rules(monkeypatch, tmp_path) + path = make_review_file(tmp_path, [new_phrase('MIT License', 0.98, AUTO, AUTO)]) + run_apply(path) + + # a second review file for the same rule, which now carries a marker + second = str(tmp_path / 'second.jsonl') + write_review_file(second, [new_record( + review_ml_phrases.load_rule('mit_1.RULE'), + [new_phrase('do things', 0.80, REVIEW, APPROVED)], + False, + )]) + run_apply(second) + + text = review_ml_phrases.load_rule('mit_1.RULE').text + assert '{{MIT License}}' in text + assert '{{do things}}' in text + + def test_the_phrases_go_in_longest_first(self): + record = new_record(make_rule(), [ + new_phrase('MIT', 0.9, AUTO, AUTO), + new_phrase('MIT License', 0.9, REVIEW, APPROVED), + new_phrase('skipped', 0.9, REVIEW, REJECTED), + ], False) + assert accepted_phrases(record) == ['MIT License', 'MIT']