mirrors/scst
1
0
Fork 0
mirror of https://github.com/SCST-project/scst.git synced 2026-10-22 06:02:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

scripts: Add qla2xxx upstream maintenance tooling

Browse Source 
Track upstream imports, compare driver snapshots, audit remaining
source differences and export patches mapped to the SCST tree. Keep
review artifacts outside Git and expose unresolved changes for review.

Move the newer QLogic kernel patch helpers into scripts/kernel with
their callers and tests. Add the update skill and guide, preserving
separate commits for upstream imports and SCST adaptations.
This commit is contained in:
Gleb Chesnokov
2026-10-19 17:36:43 +03:00
parent 29aeef4aca
commit cb1bcedc2a
12 changed files with 1860 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/kernel/extract-qla2xxx-orig
Executable
+34
View File
@@ -0,0 +1,34 @@
#!/bin/bash
# Run from the newer QLogic source directory; caches stay in the caller's cwd.
# shellcheck source=./kernel-functions
source "$(dirname "$0")/kernel-functions" || exit $?
#########################
# Argument verification #
#########################
full_kver="$1"
if [ "${full_kver}" = "" ]; then
echo "Error: missing kernel version argument."
exit 1
fi
krel=${full_kver/^*}
mkdir -p qla2xxx-orig
cd qla2xxx-orig || exit $?
download_and_extract_kernel_tree "${full_kver}" || exit $?
touch "linux-${krel}"/drivers/scsi/qla2xxx/*
rm -rf "${krel}"
mkdir -p "${krel}/include/trace/events"
mv "linux-${krel}"/drivers/scsi/qla2xxx/* "${krel}"
if [ -e "linux-${krel}"/include/trace/events/qla.h ]; then
touch "linux-${krel}"/include/trace/events/qla.h
mv "linux-${krel}"/include/trace/events/qla.h "${krel}/include/trace/events"
fi
rm -rf "linux-${krel}"
cd ..
scripts/kernel/generate-kernel-patch
+3 -1
View File
@@ -28,6 +28,8 @@ set -o pipefail
# shellcheck source=./kernel-functions
source "$(dirname "$0")/kernel-functions"
kernel_scripts=$(cd -- "$(dirname -- "$0")" && pwd) || exit $?
function usage {
echo "Usage: $0 [-d] [-h] [-n] [-p <dir>] [-s] [-u] <kernel version>"
echo "where: "
@@ -539,7 +541,7 @@ if [ "${qla2x00t}" = "true" ]; then
elif [ "${qla2x00t_32gbit}" = "true" ]; then
( cd qla2x00t-32gbit && "$PWD/generate-in-tree-patches" "${full_kver}" ) || exit $?
( cd qla2x00t-32gbit && "$kernel_scripts/generate-qla2xxx-in-tree-patches" "${full_kver}" ) || exit $?
for f in qla2x00t-32gbit/in-tree-patches/"${krel}"/*.patch; do
[ -e "$f" ] || continue
scripts/kernel/generate-qla2xxx-in-tree-patches
Executable
+62
View File
@@ -0,0 +1,62 @@
#!/bin/bash
# Run from qla2x00t-32gbit; sources and generated files stay in the caller's cwd.
full_kver="$1"
function generate_patch {
local status
{
diff -up "$1" "$2"
status=$?
if [ "$status" -gt 1 ]; then
return "$status"
fi
} > "$3"
}
if [ "$full_kver" = "" ]; then
echo "Error: missing kernel version argument."
exit 1
fi
krel=${full_kver/^*}
if [ ! -e qla2xxx-orig/"${krel}" ]; then
"$(dirname "$0")/extract-qla2xxx-orig" "${full_kver}" || exit $?
fi
mkdir -p in-tree-patches/"${krel}"
for g in Kconfig *.[ch]; do
f1="qla2xxx-orig/${krel}/$g"
f2="$g"
f3="in-tree-patches/${krel}/$g.patch"
if [ "$f1" -nt "$f3" ] || [ "$f2" -nt "$f3" ]; then
if [ -e "$f1" ]; then
generate_patch "$f1" "$f2" "$f3" || exit $?
else
generate_patch /dev/null "$f2" "$f3" || exit $?
fi
fi
done
g=include/trace/events/qla.h
f1="qla2xxx-orig/${krel}/$g"
f2="$g"
f3="in-tree-patches/${krel}/${g#include/trace/events/}.patch"
if [ "$f1" -nt "$f3" ] || [ "$f2" -nt "$f3" ]; then
if [ -e "$f1" ]; then
generate_patch "$f1" "$f2" "$f3" || exit $?
else
generate_patch /dev/null "$f2" "$f3" || exit $?
fi
fi
g=Makefile
f1="qla2xxx-orig/${krel}/$g"
f2="${g}_in-tree"
f3="in-tree-patches/${krel}/$g.patch"
if [ "$f1" -nt "$f3" ] || [ "$f2" -nt "$f3" ]; then
generate_patch "$f1" "$f2" "$f3" || exit $?
fi
scripts/kernel/qla2xxx.py
+476
View File
@@ -0,0 +1,476 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-only
"""Inspect Linux qla2xxx history, compare SCST, and export import candidates."""
import argparse
from collections import Counter
import difflib
from email.header import Header
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
import qla2xxx_audit
ROOT = Path(__file__).resolve().parents[2]
UPSTREAM = 'drivers/scsi/qla2xxx/'
TRACE = 'include/trace/events/qla.h'
DEST = 'qla2x00t-32gbit/'
PATHS = (UPSTREAM, TRACE)
MARKER = re.compile(r'^\[\s*commit ([0-9a-f]{7,40}) upstream\s*\]$', re.M)
REVERT = re.compile(r'^This reverts commit ([0-9a-f]{40})\.', re.M)
class ToolError(Exception):
"""An incomplete or invalid operation, never an empty successful result."""
class Repo:
def __init__(self, path):
self.path = Path(path).resolve()
self.git('rev-parse', '--git-dir')
if self.text('rev-parse', '--is-shallow-repository') == 'true':
raise ToolError(f'{self.path}: complete history is required (shallow repository)')
partial = self.run('config', '--get-regexp', r'^(extensions\.partialclone|remote\..*\.promisor)$')
if partial.returncode not in (0, 1):
raise ToolError(f'{self.path}: cannot inspect partial-clone configuration')
for line in partial.stdout.decode('utf-8').splitlines():
key, _, value = line.partition(' ')
if key == 'extensions.partialclone' or value.lower() not in ('false', 'no', 'off', '0'):
raise ToolError(f'{self.path}: complete objects are required (partial clone)')
def run(self, *args, data=None):
env = dict(os.environ, GIT_OPTIONAL_LOCKS='0', GIT_NO_REPLACE_OBJECTS='1',
GIT_NO_LAZY_FETCH='1', LC_ALL='C')
return subprocess.run(
['git', '--no-pager', '-C', str(self.path), *args], input=data,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, check=False)
def git(self, *args, data=None):
result = self.run(*args, data=data)
if result.returncode:
error = result.stderr.decode('utf-8', errors='replace').strip()
raise ToolError(f'{self.path}: git {args[0]} failed: {error}')
return result.stdout
def text(self, *args):
return self.git(*args).decode('utf-8').strip()
def resolve(self, ref):
return self.text('rev-parse', '--verify', '--end-of-options', f'{ref}^{{commit}}')
def ancestor(self, base, head):
result = self.run('merge-base', '--is-ancestor', base, head)
if result.returncode not in (0, 1):
raise ToolError(f'{self.path}: cannot check ancestry of {base} and {head}')
return result.returncode == 0
def files(self, ref, *paths):
result = {}
for record in self.git('ls-tree', '-rz', ref, '--', *paths).split(b'\0'):
if not record:
continue
metadata, name = record.split(b'\t', 1)
mode, kind, oid = metadata.decode('ascii').split()
if kind != 'blob':
raise ToolError(f'Unsupported tree entry: {os.fsdecode(name)} ({kind})')
result[os.fsdecode(name)] = {'mode': mode, 'oid': oid}
return result
def destination(path):
if path == TRACE:
return DEST + TRACE
if path.startswith(UPSTREAM):
relative = path[len(UPSTREAM):]
return DEST + ('Makefile_in-tree' if relative == 'Makefile' else relative)
return None
def read_imports(scst, head, linux):
"""Markers are provenance evidence, not proof that their effects remain."""
records = scst.git('log', '--full-history', '--format=%H%x00%B%x00',
head, '--', DEST).decode('utf-8').split('\0')
imports = []
reverted = set()
for offset in range(0, len(records) - 2, 2):
commit, message = records[offset].strip(), records[offset + 1]
reverted.update(REVERT.findall(message))
for marker in MARKER.findall(message):
imports.append((marker, commit))
markers = list(dict.fromkeys(marker for marker, _ in imports))
resolved = {}
if markers:
output = linux.git('cat-file', '--batch-check=%(objectname) %(objecttype)',
data=('\n'.join(markers) + '\n').encode('ascii'))
for marker, line in zip(markers, output.decode('ascii').splitlines()):
oid, kind = line.split()
resolved[marker] = oid if kind == 'commit' else None
mapping = {}
unresolved = []
for marker, commit in imports:
oid = resolved[marker]
if oid:
mapping.setdefault(oid, []).append(commit)
else:
unresolved.append({'marker': marker, 'scst_commit': commit})
latest_markers = {resolved[marker] for marker, commit in imports
if commit == imports[0][1]} if imports else set()
latest = latest_markers.pop() if len(latest_markers) == 1 else None
return mapping, reverted, unresolved, latest
def load_decisions(path, linux, scst, target, head):
"""Explicit exceptions are portable; applied commits come from Git history."""
if path is None:
return {}
document = json.loads(path.read_text(encoding='utf-8'))
if (not isinstance(document, dict) or set(document) != {'version', 'decisions'}
or document['version'] != 1 or not isinstance(document['decisions'], dict)):
raise ToolError('Decisions must contain version: 1 and a decisions object')
result = {}
for oid, decision in document['decisions'].items():
if not re.fullmatch(r'[0-9a-f]{40}', oid) or linux.resolve(oid) != oid:
raise ToolError(f'Decision requires a full upstream commit SHA: {oid!r}')
if not isinstance(decision, dict):
raise ToolError(f'{oid}: decision must be an object')
status = decision.get('status')
expected = {'status', 'reason'}
if status == 'resolved':
expected.add('scst_commit')
if (status not in ('skip', 'resolved') or set(decision) != expected
or not isinstance(decision.get('reason'), str)
or not decision['reason'].strip()):
raise ToolError(f'{oid}: use skip/resolved with a nonempty reason; '
'resolved also requires scst_commit')
if status == 'resolved':
commit = scst.resolve(decision['scst_commit'])
if commit != decision['scst_commit'] or not scst.ancestor(commit, head):
raise ToolError(f'{oid}: scst_commit must be a full SHA reachable from SCST ref')
# A shared file may contain decisions for later releases. Validate them,
# but only report decisions reachable from this target.
if linux.ancestor(oid, target):
result[oid] = decision
return result
def merge_diff(linux, oid):
return linux.git('show', '--format=', '--remerge-diff', '--no-color',
'--no-ext-diff', '--no-textconv', oid, '--', *PATHS)
def make_history(args, linux, scst, target, head):
imports, reverted, unresolved, latest = read_imports(scst, head, linux)
if args.since:
base = linux.resolve(args.since)
source = 'explicit'
elif latest:
if reverted.intersection(imports[latest]):
raise ToolError('The latest recorded import has been reverted; '
'specify --since before that import to review it')
base, source = latest, 'latest-marker'
else:
raise ToolError('Cannot infer a valid upstream base; specify --since')
if not linux.ancestor(base, target):
raise ToolError('--since (or the latest marker) is not an ancestor of --to')
decisions = load_decisions(args.decisions, linux, scst, target, head)
warnings = ['Markers record provenance, not code equivalence or complete earlier coverage.',
'Only committed trees are inspected; working-tree changes are excluded.']
if source == 'latest-marker':
warnings.append('The inferred base is a starting point, not a verified import boundary; '
'use --since for an audited release boundary.')
if unresolved:
warnings.append(f'{len(unresolved)} historical upstream markers could not be resolved.')
commits = linux.text('rev-list', '--full-history', '--simplify-merges',
'--reverse', '--topo-order',
f'{base}..{target}', '--', *PATHS).splitlines()
entries = []
for oid in commits:
metadata = linux.git('show', '-s', '--format=%P%x00%s%x00%an <%ae>%x00%aI',
oid).decode('utf-8').rstrip('\n').split('\0')
parents, subject, author, date = metadata
parents = parents.split()
# Compare every parent for merges, including files outside the mapped
# paths. No rename detection: source and destination both stay visible.
paths = set()
for parent in parents:
paths.update(os.fsdecode(p) for p in linux.git(
'diff', '--no-renames', '--name-only', '-z', parent, oid, '--').split(b'\0') if p)
entry = {'upstream': oid, 'parents': parents, 'subject': subject,
'author': author, 'author_date': date,
'kind': 'merge' if len(parents) > 1 else 'commit',
'paths': sorted(p for p in paths if destination(p)),
'outside_paths': sorted(p for p in paths if not destination(p)),
'scst_commits': imports.get(oid, []), 'status': 'pending', 'review': []}
if imports.get(oid):
entry['status'] = 'recorded'
if reverted.intersection(imports[oid]):
entry['status'] = 'review'
entry['review'].append('A recorded SCST import has a later revert; inspect its effect.')
if len(parents) > 1:
if len(parents) == 2:
entry['merge_resolution'] = bool(merge_diff(linux, oid))
entry['review'].append('Inspect merge resolution.' if entry['merge_resolution']
else 'Inspect merge; remerge diff is empty.')
else:
entry['merge_resolution'] = None
entry['review'].append('Octopus merge requires manual inspection of all parents.')
elif entry['outside_paths']:
entry['review'].append('Changes outside mapped paths require dependency review.')
if oid in decisions:
decision = decisions[oid]
entry['decision'] = decision
if decision['status'] == 'resolved' and decision['scst_commit'] in reverted:
raise ToolError(f'{oid}: resolution commit has been reverted; update the decision')
entry['status'] = decision['status']
entries.append(entry)
return {'version': 1, 'linux_target': target, 'linux_base': base,
'base_source': source, 'scst_head': head,
'mapping': {UPSTREAM: DEST, UPSTREAM + 'Makefile': DEST + 'Makefile_in-tree',
TRACE: DEST + TRACE},
'warnings': warnings, 'unresolved_markers': unresolved,
'counts': dict(Counter(e['status'] for e in entries)), 'commits': entries}
def mapped_diff(linux, parent, oid):
common = ('diff', '--no-ext-diff', '--no-textconv', '--no-color', '--no-renames',
'--full-index', '--binary', '--diff-algorithm=myers', '--unified=3')
driver = linux.git(*common, '--relative=' + UPSTREAM,
'--src-prefix=a/' + DEST, '--dst-prefix=b/' + DEST,
parent, oid, '--', UPSTREAM)
# Only patch headers change; never rewrite source text or binary payloads.
replacements = {
f'diff --git a/{DEST}Makefile b/{DEST}Makefile\n'.encode():
f'diff --git a/{DEST}Makefile_in-tree b/{DEST}Makefile_in-tree\n'.encode(),
f'--- a/{DEST}Makefile\n'.encode(): f'--- a/{DEST}Makefile_in-tree\n'.encode(),
f'+++ b/{DEST}Makefile\n'.encode(): f'+++ b/{DEST}Makefile_in-tree\n'.encode(),
}
lines = []
in_header = False
for line in driver.splitlines(keepends=True):
if line.startswith(b'diff --git '):
in_header = True
elif line.startswith((b'@@ ', b'GIT binary patch')):
in_header = False
lines.append(replacements.get(line, line) if in_header else line)
driver = b''.join(lines)
trace = linux.git(*common, '--relative=include/trace/events/',
'--src-prefix=a/' + DEST + 'include/trace/events/',
'--dst-prefix=b/' + DEST + 'include/trace/events/',
parent, oid, '--', TRACE)
return driver + trace
def import_patch(linux, entry):
oid = entry['upstream']
mail = linux.git('show', '-s', '--format=email', '--encoding=UTF-8', oid)
headers, body = mail.split(b'\n\n', 1)
subject = re.sub(r'^(?:scsi: )?qla2xxx: ', '', entry['subject'])
subject = '[PATCH] qla2x00t-32gbit: ' + subject
encoded = Header(subject, 'utf-8', header_name='Subject').encode(linesep='\n').encode('ascii')
headers, count = re.subn(rb'^Subject: [^\n]*(?:\n[ \t][^\n]*)*',
lambda _: b'Subject: ' + encoded, headers, flags=re.M)
if count != 1:
raise ToolError(f'{oid}: cannot construct mail subject')
diff = mapped_diff(linux, entry['parents'][0], oid)
if not diff:
raise ToolError(f'{oid}: selected commit has no mapped patch')
return (headers + b'\n\n' + body.rstrip() +
f'\n\n[ commit {oid} upstream ]\n\n---\n\n'.encode() + diff)
def export_history(report, linux, output):
"""Publish a complete new bundle; never clear or replace a previous run."""
output = output.absolute()
if output.exists() or output.is_symlink():
raise ToolError(f'Output already exists: {output}; choose a new directory')
if not output.parent.is_dir():
raise ToolError(f'Output parent does not exist: {output.parent}')
with tempfile.TemporaryDirectory(prefix='.qla2xxx-', dir=output.parent) as temporary:
bundle = Path(temporary) / 'bundle'
bundle.mkdir()
(bundle / 'review').mkdir()
series = []
artifacts = {}
def write(name, data):
(bundle / name).write_bytes(data)
artifacts[name] = hashlib.sha256(data).hexdigest()
for number, entry in enumerate(report['commits'], 1):
if entry['status'] in ('recorded', 'skip', 'resolved'):
continue
stem = f'{number:04d}-{entry["upstream"][:12]}'
if entry['kind'] == 'merge':
if len(entry['parents']) == 2:
name = f'review/{stem}-remerge.txt'
write(name, merge_diff(linux, entry['upstream']))
entry['artifact'] = name
continue
patch = import_patch(linux, entry)
if entry['outside_paths'] or entry['status'] == 'review':
name = f'review/{stem}-scst.patch'
write(f'review/{stem}-upstream.patch', linux.git(
'format-patch', '-1', '--stdout', '--no-signature', '--full-index',
'--binary', '--no-ext-diff', '--no-textconv', entry['upstream']))
else:
name = f'{stem}.patch'
series.append(name)
write(name, patch)
entry['artifact'] = name
write('series', (''.join(name + '\n' for name in series)).encode())
report['artifacts_sha256'] = artifacts
report['patch_count'] = len(series)
report['review_count'] = sum(bool(e['review']) for e in report['commits']
if e['status'] not in ('recorded', 'skip', 'resolved'))
(bundle / 'manifest.json').write_text(json.dumps(report, indent=2) + '\n', encoding='utf-8')
# mkdir reserves the destination without replacing even an empty directory
# created by someone else. All content is prepared before publication.
output.mkdir()
for item in bundle.iterdir():
item.rename(output / item.name)
return report
def compare(linux, target, scst, head, paths):
upstream = linux.files(target, *PATHS)
local = scst.files(head, DEST)
mapped = {destination(path): (path, metadata) for path, metadata in upstream.items()}
if paths:
selected = set()
for path in paths:
if path.startswith('/') or any(part in ('', '.', '..') for part in path.split('/')):
raise ToolError('--path must be relative to qla2x00t-32gbit/')
selected.add(DEST + path)
unknown = selected - set(mapped) - set(local)
if unknown:
raise ToolError(f'Paths absent from both trees: {", ".join(sorted(unknown))}')
else:
selected = set(mapped) | set(local)
entries = []
for name in sorted(selected):
origin, before = mapped.get(name, (None, None))
after = local.get(name)
if before is None:
status = 'scst-only'
elif after is None:
status = 'linux-only'
elif before == after:
status = 'identical'
else:
status = 'different'
entries.append({'path': name, 'upstream_path': origin, 'status': status,
'linux': before, 'scst': after})
return {'version': 1, 'linux_target': target, 'scst_head': head,
'counts': dict(Counter(e['status'] for e in entries)), 'files': entries}
def print_tree_patch(report, linux, scst):
for entry in report['files']:
if entry['status'] == 'identical':
continue
name = entry['path']
before = linux.git('cat-file', 'blob', entry['linux']['oid']) if entry['linux'] else b''
after = scst.git('cat-file', 'blob', entry['scst']['oid']) if entry['scst'] else b''
print(f'diff {entry["status"]} {name}')
print(f'mode {entry["linux"]["mode"] if entry["linux"] else "absent"} -> '
f'{entry["scst"]["mode"] if entry["scst"] else "absent"}')
if b'\0' in before or b'\0' in after:
print('Binary contents differ')
continue
for line in difflib.unified_diff(
before.decode('utf-8', errors='replace').splitlines(keepends=True),
after.decode('utf-8', errors='replace').splitlines(keepends=True),
fromfile='linux/' + (entry['upstream_path'] or name), tofile='scst/' + name):
sys.stdout.write(line)
if not line.endswith('\n'):
print('\n\\ No newline at end of file')
def print_history(report):
print(f'Linux: {report["linux_base"]}..{report["linux_target"]}')
print(f'SCST: {report["scst_head"]}')
for warning in report['warnings']:
print(f'Note: {warning}')
for entry in report['commits']:
review = ' [review]' if entry['review'] else ''
print(f'{entry["status"]:8} {entry["upstream"][:12]} '
f'{entry["kind"]:6}{review} {entry["subject"]}')
if entry.get('decision'):
print(f' {entry["decision"]["reason"]}')
for reason in entry['review']:
print(f' {reason}')
print('Counts: ' + json.dumps(report['counts'], sort_keys=True))
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
commands = parser.add_subparsers(dest='command', required=True)
for name in ('history', 'diff', 'audit', 'export'):
command = commands.add_parser(name)
command.add_argument('--linux', type=Path, required=True, help='local Linux Git repository')
command.add_argument('--scst', type=Path, default=ROOT, help='SCST repository (default: script root)')
command.add_argument('--to', required=True, help='Linux tag/commit, resolved once to a full SHA')
command.add_argument('--scst-ref', default='HEAD', help='committed SCST snapshot (default: HEAD)')
command.add_argument('--json', action='store_true', help='machine-readable output')
if name in ('history', 'export'):
command.add_argument('--since', help='exclusive upstream base; otherwise infer latest marker')
command.add_argument('--decisions', type=Path, help='reviewed skip/resolved decisions JSON')
if name == 'export':
command.add_argument('--output', type=Path, required=True, help='new bundle directory')
if name in ('diff', 'audit'):
command.add_argument('--path', action='append', default=[], help='SCST driver-relative file; repeatable')
command.add_argument('--patch', action='store_true', help='textual review diff, not an import patch')
if name == 'audit':
command.add_argument('--kernel-version', help='project known conditions for MAJOR.MINOR[.PATCH]')
command.add_argument('--define', action='append', default=[], help='assume NAME=integer at uses; repeatable')
command.add_argument('--undef', action='append', default=[], help='assume NAME undefined at uses; repeatable')
command.add_argument('--blame', action='store_true', help='include last editors of changed SCST lines')
args = parser.parse_args(argv)
if args.command in ('diff', 'audit') and args.patch and args.json:
parser.error('--patch and --json are mutually exclusive')
try:
linux, scst = Repo(args.linux), Repo(args.scst)
target, head = linux.resolve(args.to), scst.resolve(args.scst_ref)
if not linux.files(target, UPSTREAM) or not scst.files(head, DEST):
raise ToolError('Selected snapshots must contain Linux qla2xxx and SCST qla2x00t-32gbit')
if args.command == 'audit':
report = qla2xxx_audit.audit(
compare(linux, target, scst, head, args.path), linux, scst,
qla2xxx_audit.profile(args.kernel_version, args.define, args.undef), args.blame)
if not args.json:
qla2xxx_audit.print_audit(report, args.patch)
elif args.command == 'diff':
report = compare(linux, target, scst, head, args.path)
if args.patch:
print_tree_patch(report, linux, scst)
elif not args.json:
print(f'Linux: {target}\nSCST: {head}\nCommitted snapshots only.')
for entry in report['files']:
print(f'{entry["status"]:10} {entry["path"]}')
print('Counts: ' + json.dumps(report['counts'], sort_keys=True))
else:
report = make_history(args, linux, scst, target, head)
if args.command == 'export':
report = export_history(report, linux, args.output)
if not args.json:
print_history(report)
if args.command == 'export':
print(f'Exported {report["patch_count"]} candidate patches to {args.output}; '
f'{report["review_count"]} entries require separate review.')
if args.json:
print(json.dumps(report, indent=2))
except (ToolError, OSError, ValueError) as error:
print(f'qla2xxx: {error}', file=sys.stderr)
return 1
return 0
if __name__ == '__main__':
sys.exit(main())
scripts/kernel/qla2xxx_audit.py
+446
View File
@@ -0,0 +1,446 @@
# SPDX-License-Identifier: GPL-2.0-only
"""Conservative, review-oriented comparison; not a C compiler or equivalence proof."""
from collections import Counter
import difflib
import re
# Strings precede comments: comment markers inside literals are data. Line
# splicing is done first, as in C translation phase 2, preserving source lines.
LEX = re.compile(
r'(?P<literal>(?:u8|[LuU])?(?:"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\'))'
r'|(?P<comment>/\*.*?\*/|//[^\n]*)'
r'|(?P<word>[A-Za-z_]\w*|(?:\d|\.\d)(?:[eEpP][+-]|[\w.])*)'
r'|(?P<space>\s+)'
r'|(?P<operator>\%:\%:|>>=|<<=|\.\.\.|->|\+\+|--|<<|>>|<=|>=|==|!='
r'|&&|\|\||\*=|/=|%=|\+=|-=|&=|\^=|\|=|##|<:|:>|<%|%>|%:|.)', re.S)
DIRECTIVE = re.compile(r'^\s*(?:#|%:)\s*(\w+)\b(.*)$', re.S)
IDENTIFIER = r'[A-Za-z_]\w*'
def source_lines(source):
"""Logical, comment-free lines paired with original physical line numbers."""
logical, numbers, pending, start = [], [], None, 1
for number, line in enumerate(source.splitlines(keepends=True), 1):
if pending is None:
start = number
pending = ''
if line.endswith('\\\n'):
pending += line[:-2]
continue
logical.append(pending + line)
numbers.append(start)
pending = None
if pending is not None:
logical.append(pending)
numbers.append(start)
joined = ''.join(logical)
cleaned = LEX.sub(lambda m: re.sub(r'[^\n]', ' ', m.group())
if m.lastgroup == 'comment' else m.group(), joined)
return list(zip(cleaned.splitlines(keepends=True), numbers))
def tokens(lines):
"""C preprocessing tokens plus opaque directives with their source locations.
Keep directive whitespace: macro function/object distinction and #/##
stringification can make it significant. This deliberately under-normalizes.
"""
result = []
for line, number in lines:
if re.match(r'^\s*(?:#|%:)', line):
result.append((line.strip(), number))
else:
result.extend((m.group(), number) for m in LEX.finditer(line)
if m.lastgroup not in ('space', 'comment'))
return result
class Condition:
"""Small integer/boolean subset, with unknown values propagated explicitly."""
def __init__(self, expression, defines, kernel):
self.items = [m.group() for m in LEX.finditer(expression)
if m.lastgroup not in ('space', 'comment')] + ['<end>']
self.index = 0
self.defines = defines
self.kernel = kernel
def take(self):
token = self.items[self.index]
self.index += 1
return token
def accept(self, token):
if self.items[self.index] == token:
self.index += 1
return True
return False
def require(self, token):
if not self.accept(token):
raise ValueError('Unsupported preprocessor expression')
def atom(self):
token = self.take()
if token == '(':
value = self.expression()
self.require(')')
return value
if token in ('!', '+', '-'):
value = self.atom()
if value is None:
return None
return int(not value) if token == '!' else value if token == '+' else -value
if token == 'defined':
bracket = self.accept('(')
name = self.take()
if not re.fullmatch(IDENTIFIER, name):
raise ValueError('Invalid defined operand')
if bracket:
self.require(')')
return int(self.defines[name] is not None) if name in self.defines else None
if re.fullmatch(r'0[xX][0-9a-fA-F]+|0|[1-9][0-9]*', token):
value = int(token, 16 if token.lower().startswith('0x') else 10)
return value if value <= 0x7fffffff else None
if re.fullmatch(IDENTIFIER, token):
if self.accept('('):
args = []
if not self.accept(')'):
while True:
args.append(self.expression())
if self.accept(')'):
break
self.require(',')
if (token == 'KERNEL_VERSION' and self.kernel and len(args) == 3
and all(v is not None and 0 <= v <= 255 for v in args)):
return (args[0] << 16) + (args[1] << 8) + args[2]
return None
return (self.defines[token] or 0) if token in self.defines else None
raise ValueError('Unsupported preprocessor token')
def expression(self, minimum=0):
precedence = {'||': 1, '&&': 2, '==': 3, '!=': 3,
'<': 4, '>': 4, '<=': 4, '>=': 4, '+': 5, '-': 5}
left = self.atom()
while self.items[self.index] in precedence:
op = self.items[self.index]
level = precedence[op]
if level < minimum:
break
self.take()
right = self.expression(level + 1)
if op == '&&' and (left == 0 or right == 0):
left = 0
elif op == '||' and ((left is not None and left != 0)
or (right is not None and right != 0)):
left = 1
elif left is None or right is None:
left = None
else:
left = {'||': lambda: int(bool(left or right)),
'&&': lambda: int(bool(left and right)),
'==': lambda: int(left == right), '!=': lambda: int(left != right),
'<': lambda: int(left < right), '>': lambda: int(left > right),
'<=': lambda: int(left <= right), '>=': lambda: int(left >= right),
'+': lambda: left + right, '-': lambda: left - right}[op]()
if not -0x7fffffff <= left <= 0x7fffffff:
left = None
return left
def value(self):
try:
value = self.expression()
return value if self.items[self.index] == '<end>' else None
except (ValueError, IndexError, RecursionError):
return None
def profile(version, definitions, undefined):
values = {}
if version:
if not re.fullmatch(r'\d+\.\d+(?:\.\d+)?', version):
raise ValueError('--kernel-version requires MAJOR.MINOR[.PATCH]')
parts = [int(v) for v in version.split('.')]
if len(parts) == 2:
parts.append(0)
if any(p > 255 for p in parts):
raise ValueError('--kernel-version components must be in 0..255')
values['LINUX_VERSION_CODE'] = (parts[0] << 16) + (parts[1] << 8) + parts[2]
for definition in definitions:
name, _, value = definition.partition('=')
if (not re.fullmatch(IDENTIFIER, name) or not value
or not re.fullmatch(r'0[xX][0-9a-fA-F]+|0|[1-9][0-9]*', value)):
raise ValueError('--define requires NAME=nonnegative-integer')
number = int(value, 16 if value.lower().startswith('0x') else 10)
if name in values or name == 'KERNEL_VERSION' or number > 0x7fffffff:
raise ValueError(f'Conflicting or unsupported definition: {definition}')
values[name] = number
for name in undefined:
if not re.fullmatch(IDENTIFIER, name) or name in values or name == 'KERNEL_VERSION':
raise ValueError(f'Conflicting or invalid --undef: {name}')
values[name] = None
return {'kernel_version': version, 'defines': values,
'enabled': bool(version or definitions or undefined)}
def project(lines, selected):
"""Evaluate only explicit profile facts; never infer absent macros as zero.
Local redefinitions invalidate external assumptions even in conditional
branches. Includes are not expanded: profile facts must hold at their uses.
Unknown branches remain, with independently decidable nested groups reduced.
"""
if not selected['enabled']:
return lines, [], []
values = dict(selected['defines'])
redefined = set()
for line, _ in lines:
match = re.match(r'^\s*(?:#|%:)\s*(?:define|undef)\s+(' + IDENTIFIER + ')', line)
if match:
redefined.add(match[1])
for name in redefined:
values.pop(name, None)
kernel = bool(selected['kernel_version']) and 'KERNEL_VERSION' not in redefined
decisions = []
def condition(command, expression):
if command in ('ifdef', 'ifndef'):
value = values.get(expression.strip())
known = expression.strip() in values
return int((value is not None) == (command == 'ifdef')) if known else None
return Condition(expression, values, kernel).value()
def block(index):
output = []
while index < len(lines):
line, number = lines[index]
directive = DIRECTIVE.match(line)
command = directive[1] if directive else ''
if command in ('elif', 'else', 'endif'):
return output, index
if command not in ('if', 'ifdef', 'ifndef'):
output.append(lines[index])
index += 1
continue
branches = []
saw_else = False
while True:
header = lines[index]
match = DIRECTIVE.match(header[0])
kind, expression = match.groups()
if saw_else:
raise ValueError(f'Unexpected #{kind} after #else at line {header[1]}')
saw_else = kind == 'else'
value = 1 if saw_else else condition(kind, expression)
content, end = block(index + 1)
branches.append((header, value, content))
if end >= len(lines):
raise ValueError(f'Unclosed conditional at line {number}')
end_kind = DIRECTIVE.match(lines[end][0])[1]
if end_kind == 'endif':
break
index = end
chosen, known = None, True
for branch_number, (_, value, _) in enumerate(branches):
if value is None:
known = False
break
if value:
chosen = branch_number
break
decisions.append({'line': number, 'condition': line.strip(),
'result': 'selected' if known else 'unknown',
'selected_line': branches[chosen][0][1]
if known and chosen is not None else None})
if known:
if chosen is not None:
output.extend(branches[chosen][2])
else:
for header, _, content in branches:
output.append(header)
output.extend(content)
output.append(lines[end])
index = end + 1
return output, index
output, end = block(0)
if end != len(lines):
raise ValueError(f'Unmatched conditional at line {lines[end][1]}')
assumptions = set(selected['defines']) | ({'KERNEL_VERSION'} if selected['kernel_version'] else set())
return output, sorted(decisions, key=lambda d: d['line']), sorted(redefined & assumptions)
def line_delta(before, after, matcher=None):
if matcher is None:
matcher = difflib.SequenceMatcher(None, before, after, autojunk=False)
added = removed = 0
for kind, a, b, c, d in matcher.get_opcodes():
if kind != 'equal':
removed += b - a
added += d - c
return {'added': added, 'removed': removed}
def unified(before, after, name):
diff = difflib.unified_diff(before.splitlines(keepends=True),
after.splitlines(keepends=True),
fromfile='linux/' + name, tofile='scst/' + name)
return ''.join(line if line.endswith('\n') else line + '\n\\ No newline at end of file\n'
for line in diff)
def classify(before, after, path):
"""Review hints, never a claim that a hunk is a safe backport."""
joined = before + '\n' + after
if '/qla2x00-target/' in path or re.search(r'\b(?:scst_\w*|SCST_\w*)', joined):
return 'scst-integration', 'SCST adapter path or SCST symbol in the changed code'
if re.search(r'\b(?:LINUX_VERSION_CODE|KERNEL_VERSION|RHEL_\w*|HAVE_\w*|QLA_ENABLE_PI'
r'|NEW_LIBFC_API|BSG_JOB_TYPE|bsg_job_sense|se_cmd_tag|MAP_QUEUES_RET)\b', joined):
return 'compatibility-candidate', 'Compatibility wrapper or guard remains; inspect its definition and branches'
if re.search(r'(?<![=!<>])=(?!=)|\b(?:if|return|goto)\b', joined):
return 'behavior-candidate', 'Assignment or control flow changed; inspect values and callers'
calls = r'\b([A-Za-z_]\w*)\s*\('
if set(re.findall(calls, before)) != set(re.findall(calls, after)):
return 'behavior-candidate', 'Called or declared function names differ; inspect API semantics'
return 'needs-review', 'No automatic explanation for the remaining token change'
def scope_hint(lines, number):
# A column-zero kernel-style function definition, not a C parser. This is
# only navigation context; line spans and original diff are authoritative.
text = ''.join(line for line, original in lines if original <= number)
matches = list(re.finditer(r'^(?:[A-Za-z_]\w*[\t *]+)*([A-Za-z_]\w*)\([^;{}]*\)\s*\{', text, re.M))
return matches[-1][1] if matches else None
def residual_hunks(before, after, original_before, original_after, path, matcher):
result = []
for group in matcher.get_grouped_opcodes(6):
changes = [op for op in group if op[0] != 'equal']
entry = {}
changed_text = []
for side, stream, source, lower, upper in (
('linux', before, original_before, 1, 2),
('scst', after, original_after, 3, 4)):
ids = sorted({stream[i][1] for op in changes for i in range(op[lower], op[upper])})
# A pure insertion/deletion still has an anchor in the other tree.
anchor = min(changes[0][lower], len(stream) - 1)
start = ids[0] if ids else stream[anchor][1] if stream else 1
stop = ids[-1] if ids else start
entry[side] = {'start': start, 'end': stop, 'changed_lines': ids,
'text': '\n'.join(source.splitlines()[start - 1:stop]),
'scope_hint': scope_hint(source_lines(source), start)}
physical = source.splitlines()
changed_text.append('\n'.join(physical[n - 1] for n in ids))
entry['category'], entry['reason'] = classify(*changed_text, path)
entry['requires_review'] = True
result.append(entry)
return result
def audit(report, linux, scst, selected, blame=False):
report['audit_version'] = 1
report['profile'] = selected
report['warnings'] = [
'Only committed snapshots are inspected; working-tree changes are excluded.',
'Token equality and conditional projection are not semantic equivalence.',
'Includes/macros are not expanded. Unknown conditions remain; profile facts '
'must hold at their uses. Classification and scope names are review hints.',
'Blame identifies last editors of SCST lines, not necessarily the origin of a difference.']
for entry in report['files']:
entry['hunks'] = []
if entry['status'] == 'identical':
entry['assessment'] = 'identical'
continue
if any(entry[side] and entry[side]['mode'] not in ('100644', '100755')
for side in ('linux', 'scst')):
entry['assessment'] = 'special-file-review'
continue
blobs = [repo.git('cat-file', 'blob', entry[key]['oid']) if entry[key] else b''
for repo, key in ((linux, 'linux'), (scst, 'scst'))]
if any(b'\0' in blob for blob in blobs):
entry['assessment'] = 'binary-review'
continue
try:
old, new = [blob.decode('utf-8') for blob in blobs]
except UnicodeDecodeError:
entry['assessment'] = 'encoding-review'
continue
entry['raw_lines'] = line_delta(old.splitlines(), new.splitlines())
entry['raw_diff'] = unified(old, new, entry['path'])
if not entry['path'].endswith(('.c', '.h')):
entry['assessment'] = 'non-c-review'
continue
streams = [source_lines(source) for source in (old, new)]
lexical = [tokens(lines) for lines in streams]
projections = [project(lines, selected) for lines in streams]
projected = [tokens(p[0]) for p in projections]
for side, (_, decisions, invalidated) in zip(('linux', 'scst'), projections):
entry[side + '_conditions'] = decisions
entry[side + '_invalidated_defines'] = invalidated
old_tokens, new_tokens = [[t for t, _ in stream] for stream in projected]
# Matching large C files is the expensive step. Reuse its opcodes for
# both counts and hunks instead of matching the same streams twice.
matcher = difflib.SequenceMatcher(None, old_tokens, new_tokens, autojunk=False)
entry['residual_tokens'] = line_delta(old_tokens, new_tokens, matcher)
entry['hunks'] = residual_hunks(*projected, old, new, entry['path'], matcher)
if entry['status'] != 'different':
entry['assessment'] = entry['status'] + '-review'
elif entry['linux']['mode'] != entry['scst']['mode']:
entry['assessment'] = 'mode-review'
elif [t for t, _ in lexical[0]] == [t for t, _ in lexical[1]]:
entry['assessment'] = 'lexical-only'
elif not entry['hunks']:
entry['assessment'] = 'equal-under-profile'
else:
entry['assessment'] = 'residual-review'
if blame and entry['scst'] and entry['hunks']:
history = scst.text('blame', '--line-porcelain', report['scst_head'], '--', entry['path'])
origins = {}
commit, line = None, None
for record in history.splitlines():
match = re.fullmatch(r'([0-9a-f]{40}) \d+ (\d+)(?: \d+)?', record)
if match:
commit, line = match[1], int(match[2])
elif record.startswith('summary ') and line is not None:
origins[line] = {'commit': commit, 'subject': record[8:]}
for hunk in entry['hunks']:
evidence = {origins[n]['commit']: origins[n] for n in hunk['scst']['changed_lines']
if n in origins}
hunk['scst_last_editors'] = sorted(evidence.values(), key=lambda e: e['commit'])
report['assessments'] = dict(Counter(e['assessment'] for e in report['files']))
report['hunk_categories'] = dict(Counter(h['category'] for e in report['files'] for h in e['hunks']))
report['shared_c_totals'] = {
metric: {direction: sum(e.get(metric, {}).get(direction, 0) for e in report['files']
if e['status'] == 'different' and e['path'].endswith(('.c', '.h')))
for direction in ('added', 'removed')}
for metric in ('raw_lines', 'residual_tokens')}
return report
def print_audit(report, patch=False):
print(f'Linux: {report["linux_target"]}\nSCST: {report["scst_head"]}')
print(f'Profile: {report["profile"]}')
for warning in report['warnings']:
print('Note: ' + warning)
for entry in report['files']:
if entry['assessment'] == 'identical':
continue
print(f'\n{entry["assessment"]}: {entry["path"]} {entry.get("raw_lines", {})}')
if patch and not entry['hunks'] and entry['assessment'].endswith('-review'):
print(entry.get('raw_diff', '(non-text difference)'))
for hunk in entry['hunks']:
old, new = hunk['linux'], hunk['scst']
print(f' {hunk["category"]}: Linux {old["start"]}:{old["end"]} -> '
f'SCST {new["start"]}:{new["end"]} '
f'({new["scope_hint"] or old["scope_hint"] or "global/declaration"})')
if patch:
print(' ' + hunk['reason'])
for prefix, part in (('-', old), ('+', new)):
if part['changed_lines']:
for line in part['text'].splitlines():
print(prefix + line)
print('\nAssessments: ' + str(report['assessments']))
print('Residual hunks: ' + str(report['hunk_categories']))
print('Shared C sources: ' + str(report['shared_c_totals']))
scripts/kernel/test_qla2xxx.py
+477
View File
@@ -0,0 +1,477 @@
# SPDX-License-Identifier: GPL-2.0-only
"""Integration tests with disposable Git repositories, without kernel builds."""
from contextlib import redirect_stderr, redirect_stdout
import io
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
import qla2xxx
import qla2xxx_audit as audit
class QlaTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory(prefix='qla test ')
self.addCleanup(self.temporary.cleanup)
self.root = Path(self.temporary.name)
self.linux = self.root / 'linux repo'
self.scst = self.root / 'scst repo'
self.env = dict(os.environ, GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL='/dev/null',
GIT_AUTHOR_NAME='Upstream Author', GIT_AUTHOR_EMAIL='author@example.com',
GIT_COMMITTER_NAME='Test Maintainer', GIT_COMMITTER_EMAIL='test@example.com',
GIT_AUTHOR_DATE='2025-01-02T03:04:05+00:00',
GIT_COMMITTER_DATE='2025-01-02T03:04:05+00:00')
self.environment = patch.dict(os.environ, self.env, clear=True)
self.environment.start()
self.addCleanup(self.environment.stop)
for repo in (self.linux, self.scst):
repo.mkdir()
self.git(repo, 'init', '-q', '--initial-branch=main')
self.base = self.commit(self.linux, 'Base', {
qla2xxx.UPSTREAM + 'qla.c': 'base\n',
qla2xxx.UPSTREAM + 'Makefile': 'obj-m := old.o\n',
qla2xxx.TRACE: 'trace old\n',
})
self.scst_base = self.commit(self.scst, 'qla2x00t-32gbit: Base\n\n'
f'[ commit {self.base[:12]} upstream ]', {
qla2xxx.DEST + 'qla.c': 'base\n',
qla2xxx.DEST + 'Makefile_in-tree': 'obj-m := old.o\n',
qla2xxx.DEST + 'Makefile': 'SCST out-of-tree rules\n',
qla2xxx.DEST + qla2xxx.TRACE: 'trace old\n',
qla2xxx.DEST + 'qla2x00-target/adapter.c': 'SCST adapter\n',
})
def git(self, repo, *args, check=True):
result = subprocess.run(['git', '-C', str(repo), *args], env=self.env,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if check and result.returncode:
self.fail(result.stderr.decode(errors='replace'))
return result
def commit(self, repo, message, files):
for name, content in files.items():
path = repo / name
if content is None:
path.unlink()
else:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(content.encode() if isinstance(content, str) else content)
self.git(repo, 'add', '-A')
self.git(repo, 'commit', '-q', '-m', message)
return self.git(repo, 'rev-parse', 'HEAD').stdout.decode().strip()
def invoke(self, command, *options, ok=True):
stdout, stderr = io.StringIO(), io.StringIO()
with redirect_stdout(stdout), redirect_stderr(stderr):
status = qla2xxx.main([command, '--linux', str(self.linux), '--scst', str(self.scst),
'--to', 'HEAD', *map(str, options)])
if ok:
self.assertEqual(status, 0, stderr.getvalue())
else:
self.assertNotEqual(status, 0, stdout.getvalue())
return stdout.getvalue(), stderr.getvalue()
def history(self, *options):
return json.loads(self.invoke('history', '--json', *options)[0])
def output(self, name='export'):
return self.root / name
def export(self, *options):
return json.loads(self.invoke('export', '--json', '--output', self.output(), *options)[0])
def test_infers_marker_ignoring_later_local_change_and_detects_recorded_import(self):
first = self.commit(self.linux, 'scsi: qla2xxx: First', {qla2xxx.UPSTREAM + 'qla.c': 'one\n'})
self.commit(self.scst, f'qla2x00t-32gbit: First\n\n[ commit {first} upstream ]',
{qla2xxx.DEST + 'qla.c': 'one\n'})
self.commit(self.scst, 'qla2x00t-32gbit: Local adaptation',
{qla2xxx.DEST + 'local.c': 'local\n'})
second = self.commit(self.linux, 'scsi: qla2xxx: Second',
{qla2xxx.UPSTREAM + 'qla.c': 'two\n'})
report = self.history()
self.assertEqual(report['linux_base'], first)
self.assertEqual([e['upstream'] for e in report['commits']], [second])
report = self.history('--since', self.base)
self.assertEqual([e['status'] for e in report['commits']], ['recorded', 'pending'])
def test_export_applies_paths_binary_rename_author_date_and_provenance(self):
first = self.commit(self.linux, 'scsi: qla2xxx: Change all mappings\n\n'
'Keep the explanation.\n\nSigned-off-by: Upstream Author <author@example.com>', {
qla2xxx.UPSTREAM + 'qla.c': 'new\n',
qla2xxx.UPSTREAM + 'Makefile': 'obj-m := new.o\n',
qla2xxx.TRACE: 'trace new\n',
qla2xxx.UPSTREAM + 'firmware.bin': b'\0\xffbinary\0',
})
second = self.commit(self.linux, 'scsi: qla2xxx: Rename file', {
qla2xxx.UPSTREAM + 'qla.c': None,
qla2xxx.UPSTREAM + 'new name.c': 'new\n',
})
self.commit(self.linux, 'scsi: qla2xxx: Trace only', {qla2xxx.TRACE: 'trace latest\n'})
before = [self.git(repo, 'status', '--porcelain').stdout for repo in (self.linux, self.scst)]
report = self.export()
self.assertEqual(report['patch_count'], 3)
self.assertEqual(before, [self.git(repo, 'status', '--porcelain').stdout
for repo in (self.linux, self.scst)])
self.assertEqual(self.git(self.scst, 'rev-parse', 'HEAD').stdout.decode().strip(), self.scst_base)
for filename in (self.output() / 'series').read_text().splitlines():
self.git(self.scst, 'am', str(self.output() / filename))
self.assertEqual((self.scst / (qla2xxx.DEST + 'Makefile')).read_text(), 'SCST out-of-tree rules\n')
self.assertEqual((self.scst / (qla2xxx.DEST + 'Makefile_in-tree')).read_text(), 'obj-m := new.o\n')
self.assertEqual((self.scst / (qla2xxx.DEST + qla2xxx.TRACE)).read_text(), 'trace latest\n')
self.assertEqual((self.scst / (qla2xxx.DEST + 'firmware.bin')).read_bytes(), b'\0\xffbinary\0')
self.assertFalse((self.scst / (qla2xxx.DEST + 'qla.c')).exists())
self.assertEqual((self.scst / (qla2xxx.DEST + 'new name.c')).read_text(), 'new\n')
log = self.git(self.scst, 'log', '--format=%an <%ae>%n%aI%n%B', self.scst_base + '..HEAD').stdout.decode()
self.assertIn('Upstream Author <author@example.com>', log)
self.assertIn(self.git(self.linux, 'show', '-s', '--format=%aI', first).stdout.decode().strip(), log)
self.assertIn('qla2x00t-32gbit: Change all mappings', log)
self.assertIn(f'[ commit {first} upstream ]', log)
self.assertIn(f'[ commit {second} upstream ]', log)
self.assertIn('Signed-off-by: Upstream Author <author@example.com>', log)
self.assertEqual(self.history('--since', self.base)['counts'], {'recorded': 3})
def test_unicode_long_subject_and_author_round_trip(self):
self.env['GIT_AUTHOR_NAME'] = 'Автор Патча'
subject = 'Исправить ' + 'длинное описание ' * 10
self.commit(self.linux, 'scsi: qla2xxx: ' + subject + '\n\nОписание изменения.',
{qla2xxx.UPSTREAM + 'qla.c': 'unicode\n'})
self.export()
self.git(self.scst, 'am', str(next(self.output().glob('*.patch'))))
self.assertEqual(self.git(self.scst, 'show', '-s', '--format=%an').stdout.decode().strip(),
'Автор Патча')
self.assertEqual(self.git(self.scst, 'show', '-s', '--format=%s').stdout.decode().strip(),
'qla2x00t-32gbit: ' + subject.strip())
self.assertIn('Описание изменения.', self.git(self.scst, 'show', '-s', '--format=%B').stdout.decode())
def test_patch_header_mapping_never_rewrites_hunk_content(self):
# Deleted source text can look exactly like a --- patch header.
old = '-- a/qla2x00t-32gbit/Makefile\n'
self.commit(self.linux, 'scsi: qla2xxx: Add header-like text',
{qla2xxx.UPSTREAM + 'qla.c': old})
self.commit(self.linux, 'scsi: qla2xxx: Remove header-like text',
{qla2xxx.UPSTREAM + 'qla.c': 'ordinary text\n'})
self.export()
for filename in (self.output() / 'series').read_text().splitlines():
self.git(self.scst, 'am', str(self.output() / filename))
self.assertEqual((self.scst / (qla2xxx.DEST + 'qla.c')).read_text(), 'ordinary text\n')
def test_real_merge_resolution_is_exported_for_review(self):
self.git(self.linux, 'checkout', '-q', '-b', 'side')
side = self.commit(self.linux, 'scsi: qla2xxx: Side', {qla2xxx.UPSTREAM + 'qla.c': 'side\n'})
self.git(self.linux, 'checkout', '-q', 'main')
main = self.commit(self.linux, 'scsi: qla2xxx: Main', {qla2xxx.UPSTREAM + 'qla.c': 'main\n'})
result = self.git(self.linux, 'merge', '--no-ff', 'side', '-m', 'Merge side', check=False)
self.assertNotEqual(result.returncode, 0)
merge = self.commit(self.linux, 'Merge resolved', {qla2xxx.UPSTREAM + 'qla.c': 'main and side\n'})
report = self.export()
self.assertEqual({e['upstream'] for e in report['commits']}, {side, main, merge})
entry = report['commits'][-1]
self.assertTrue(entry['merge_resolution'])
self.assertEqual(report['review_count'], 1)
self.assertEqual(report['patch_count'], 2)
self.assertIn(b'main and side', (self.output() / entry['artifact']).read_bytes())
self.assertNotIn('remerge', (self.output() / 'series').read_text())
def test_clean_merge_is_still_reported(self):
self.git(self.linux, 'checkout', '-q', '-b', 'side')
self.commit(self.linux, 'scsi: qla2xxx: Side', {qla2xxx.UPSTREAM + 'other.c': 'side\n'})
self.git(self.linux, 'checkout', '-q', 'main')
self.commit(self.linux, 'scsi: qla2xxx: Main', {qla2xxx.UPSTREAM + 'qla.c': 'main\n'})
self.git(self.linux, 'merge', '--no-ff', 'side', '-m', 'Merge side')
merges = [e for e in self.history()['commits'] if e['kind'] == 'merge']
# Full-history may omit a TREESAME merge with no relevant contribution;
# this merge changes the selected subtree relative to both parents.
self.assertEqual(len(merges), 1)
self.assertFalse(merges[0]['merge_resolution'])
def test_merge_that_discards_a_side_change_remains_visible(self):
self.git(self.linux, 'checkout', '-q', '-b', 'side')
side = self.commit(self.linux, 'scsi: qla2xxx: Side', {qla2xxx.UPSTREAM + 'qla.c': 'side\n'})
self.git(self.linux, 'checkout', '-q', 'main')
main = self.commit(self.linux, 'scsi: qla2xxx: Main', {qla2xxx.UPSTREAM + 'qla.c': 'main\n'})
self.git(self.linux, 'merge', '--no-ff', 'side', '-m', 'Merge side', check=False)
merge = self.commit(self.linux, 'Keep main version', {qla2xxx.UPSTREAM + 'qla.c': 'main\n'})
report = self.history()
self.assertEqual({e['upstream'] for e in report['commits']}, {side, main, merge})
self.assertTrue(report['commits'][-1]['merge_resolution'])
def test_mixed_commit_is_not_in_automatic_series_and_keeps_original(self):
self.commit(self.linux, 'scsi: Shared API change', {
qla2xxx.UPSTREAM + 'qla.c': 'new API\n', 'include/scsi/api.h': 'external dependency\n'})
report = self.export()
self.assertEqual(report['patch_count'], 0)
entry = report['commits'][0]
self.assertEqual(entry['outside_paths'], ['include/scsi/api.h'])
self.assertIn('review/', entry['artifact'])
self.assertNotIn(b'include/scsi/api.h', (self.output() / entry['artifact']).read_bytes())
original = next((self.output() / 'review').glob('*-upstream.patch'))
self.assertIn(b'external dependency', original.read_bytes())
def test_diff_maps_makefile_and_classifies_committed_tree_only(self):
self.commit(self.linux, 'scsi: qla2xxx: Add', {qla2xxx.UPSTREAM + 'new.c': 'new\n'})
(self.scst / (qla2xxx.DEST + 'qla.c')).write_text('uncommitted\n')
report = json.loads(self.invoke('diff', '--json')[0])
statuses = {e['path']: e['status'] for e in report['files']}
self.assertEqual(statuses[qla2xxx.DEST + 'Makefile_in-tree'], 'identical')
self.assertEqual(statuses[qla2xxx.DEST + 'Makefile'], 'scst-only')
self.assertEqual(statuses[qla2xxx.DEST + 'qla.c'], 'identical')
self.assertEqual(statuses[qla2xxx.DEST + 'new.c'], 'linux-only')
limited = json.loads(self.invoke('diff', '--json', '--path', 'new.c')[0])
self.assertEqual(len(limited['files']), 1)
output = self.invoke('diff', '--patch', '--path', 'new.c')[0]
self.assertIn('-new\n', output)
self.invoke('diff', '--path', '../escape', ok=False)
self.invoke('diff', '--path', 'not-present.c', ok=False)
def test_decisions_require_reason_and_reachable_resolution(self):
oid = self.commit(self.linux, 'scsi: qla2xxx: Optional change',
{qla2xxx.UPSTREAM + 'qla.c': 'new\n'})
path = self.root / 'decisions.json'
document = {'version': 1, 'decisions': {oid: {'status': 'skip', 'reason': 'Not applicable'}}}
path.write_text(json.dumps(document))
report = self.export('--decisions', path)
self.assertEqual(report['counts'], {'skip': 1})
self.assertEqual(report['patch_count'], 0)
document['decisions'][oid] = {'status': 'resolved', 'reason': 'Adapted separately',
'scst_commit': self.scst_base}
path.write_text(json.dumps(document))
self.assertEqual(self.history('--decisions', path)['counts'], {'resolved': 1})
document['decisions'][oid]['reason'] = ''
path.write_text(json.dumps(document))
self.invoke('history', '--decisions', path, ok=False)
unmerged = self.commit(self.scst, 'Unmerged adaptation',
{qla2xxx.DEST + 'qla.c': 'unmerged\n'})
self.git(self.scst, 'checkout', '-q', self.scst_base)
document['decisions'][oid].update(reason='Unmerged', scst_commit=unmerged)
path.write_text(json.dumps(document))
self.invoke('history', '--decisions', path, ok=False)
def test_reverted_import_is_not_silently_skipped(self):
oid = self.commit(self.linux, 'scsi: qla2xxx: Fix', {qla2xxx.UPSTREAM + 'qla.c': 'fix\n'})
self.commit(self.scst, f'qla2x00t-32gbit: Fix\n\n[ commit {oid} upstream ]',
{qla2xxx.DEST + 'qla.c': 'fix\n'})
self.git(self.scst, 'revert', '--no-edit', 'HEAD')
self.invoke('history', ok=False)
report = self.export('--since', self.base)
self.assertEqual(report['commits'][0]['status'], 'review')
self.assertEqual(report['patch_count'], 0)
self.assertTrue((self.output() / report['commits'][0]['artifact']).is_file())
def test_output_is_never_replaced_even_when_empty_or_symlink(self):
self.output().mkdir()
self.invoke('export', '--output', self.output(), ok=False)
(self.output() / 'keep').write_text('previous run')
self.invoke('export', '--output', self.output(), ok=False)
self.assertEqual((self.output() / 'keep').read_text(), 'previous run')
link = self.root / 'broken-link'
link.symlink_to(self.root / 'missing')
self.invoke('export', '--output', link, ok=False)
self.assertTrue(link.is_symlink())
def test_empty_range_is_distinct_from_invalid_ref(self):
self.assertEqual(self.history()['commits'], [])
self.invoke('history', '--to', 'not-a-ref', ok=False)
self.invoke('history', '--since', 'not-a-ref', ok=False)
self.git(self.linux, 'checkout', '-q', '--orphan', 'unrelated')
self.commit(self.linux, 'Unrelated root', {qla2xxx.UPSTREAM + 'qla.c': 'other\n'})
self.invoke('history', '--since', self.base, ok=False)
def test_unresolved_latest_marker_requires_explicit_base(self):
self.commit(self.scst, 'qla2x00t-32gbit: Missing source\n\n'
'[ commit 0000000000000000000000000000000000000001 upstream ]',
{qla2xxx.DEST + 'qla.c': 'missing\n'})
self.invoke('history', ok=False)
report = self.history('--since', self.base)
self.assertEqual(len(report['unresolved_markers']), 1)
def test_shallow_history_is_rejected(self):
clone = self.root / 'shallow'
self.git(self.root, 'clone', '-q', '--depth=1', self.linux.as_uri(), str(clone))
self.invoke('history', '--linux', clone, ok=False)
def test_partial_clone_is_rejected_without_attempting_network_access(self):
self.git(self.linux, 'config', 'remote.origin.promisor', 'true')
_, error = self.invoke('history', ok=False)
self.assertIn('partial clone', error)
def test_latest_import_with_multiple_markers_needs_an_explicit_base(self):
oid = self.commit(self.linux, 'scsi: qla2xxx: Fix', {qla2xxx.UPSTREAM + 'qla.c': 'fix\n'})
self.commit(self.scst, 'qla2x00t-32gbit: Combined import\n\n'
f'[ commit {self.base} upstream ]\n[ commit {oid} upstream ]',
{qla2xxx.DEST + 'qla.c': 'fix\n'})
self.invoke('history', ok=False)
self.assertEqual(self.history('--since', self.base)['counts'], {'recorded': 1})
def test_mode_and_missing_final_newline_survive_export(self):
self.commit(self.linux, 'scsi: qla2xxx: No final newline',
{qla2xxx.UPSTREAM + 'qla.c': 'no newline'})
path = self.linux / (qla2xxx.UPSTREAM + 'qla.c')
path.chmod(0o755)
self.commit(self.linux, 'scsi: qla2xxx: Change mode', {})
self.export()
for filename in (self.output() / 'series').read_text().splitlines():
self.git(self.scst, 'am', str(self.output() / filename))
path = self.scst / (qla2xxx.DEST + 'qla.c')
self.assertEqual(path.read_bytes(), b'no newline')
self.assertTrue(path.stat().st_mode & 0o111)
def test_failure_does_not_publish_incomplete_bundle(self):
self.commit(self.linux, 'scsi: qla2xxx: Fix', {qla2xxx.UPSTREAM + 'qla.c': 'new\n'})
with patch.object(qla2xxx, 'import_patch', side_effect=qla2xxx.ToolError('test failure')):
self.invoke('export', '--output', self.output(), ok=False)
self.assertFalse(self.output().exists())
self.assertFalse(list(self.root.glob('.qla2xxx-*')))
def test_export_can_run_from_an_unrelated_directory(self):
result = subprocess.run([
sys.executable, str(Path(qla2xxx.__file__).resolve()), 'export',
'--linux', str(self.linux), '--scst', str(self.scst), '--to', 'HEAD',
'--output', str(self.output()), '--json'], cwd=self.root, env=self.env,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
self.assertEqual(result.returncode, 0, result.stderr.decode())
self.assertEqual(json.loads(result.stdout)['patch_count'], 0)
def test_audit_keeps_raw_diff_literals_behavior_and_blame(self):
self.commit(self.linux, 'Base source', {qla2xxx.UPSTREAM + 'qla.c':
'int mode = 1;\nchar *name = "a b";\n'})
change = self.commit(self.scst, 'Change policy', {qla2xxx.DEST + 'qla.c':
'int mode = 0;\nchar *name = "ab";\n'})
report = json.loads(self.invoke('audit', '--blame', '--json', '--path', 'qla.c')[0])
entry = report['files'][0]
self.assertEqual(entry['assessment'], 'residual-review')
self.assertIn('-int mode = 1;', entry['raw_diff'])
self.assertIn('+int mode = 0;', entry['raw_diff'])
self.assertGreater(entry['residual_tokens']['added'], 0)
self.assertTrue(all(h['requires_review'] for h in entry['hunks']))
self.assertIn(change, {c['commit'] for h in entry['hunks'] for c in h['scst_last_editors']})
self.assertEqual(self.git(self.scst, 'status', '--porcelain').stdout, b'')
def test_audit_distinguishes_comment_noise_from_profile_equality(self):
self.commit(self.linux, 'Base source', {qla2xxx.UPSTREAM + 'qla.c': 'int f(void) { return 1; }\n'})
self.commit(self.scst, 'Reformat', {qla2xxx.DEST + 'qla.c':
'/* comment */\nint f(void)\n{\n\treturn 1; // explanation\n}\n'})
entry = json.loads(self.invoke('audit', '--json', '--path', 'qla.c')[0])['files'][0]
self.assertEqual(entry['assessment'], 'lexical-only')
self.assertEqual(entry['hunks'], [])
self.commit(self.scst, 'Backport', {qla2xxx.DEST + 'qla.c':
'#if LINUX_VERSION_CODE >= KERNEL_VERSION(7, 0, 0)\n'
'int f(void) { return 1; }\n#else\nint f(void) { return 2; }\n#endif\n'})
entry = json.loads(self.invoke('audit', '--json', '--kernel-version', '7.3',
'--path', 'qla.c')[0])['files'][0]
self.assertEqual(entry['assessment'], 'equal-under-profile')
self.assertIn('return 2;', entry['raw_diff'])
entry = json.loads(self.invoke('audit', '--json', '--kernel-version', '6.1',
'--path', 'qla.c')[0])['files'][0]
self.assertEqual(entry['assessment'], 'residual-review')
self.assertEqual(entry['hunks'][0]['scst']['start'], 4)
def test_audit_preserves_unknown_vendor_and_config_branches(self):
self.commit(self.linux, 'Base source', {qla2xxx.UPSTREAM + 'qla.c': 'int value;\n'})
self.commit(self.scst, 'Conditional', {qla2xxx.DEST + 'qla.c':
'#if LINUX_VERSION_CODE >= KERNEL_VERSION(7, 0, 0) && defined(VENDOR)\n'
'int other;\n#else\nint value;\n#endif\n'})
entry = json.loads(self.invoke('audit', '--json', '--kernel-version', '7.3',
'--path', 'qla.c')[0])['files'][0]
self.assertEqual(entry['assessment'], 'residual-review')
self.assertEqual(entry['scst_conditions'][0]['result'], 'unknown')
entry = json.loads(self.invoke('audit', '--json', '--kernel-version', '7.3',
'--undef', 'VENDOR', '--path', 'qla.c')[0])['files'][0]
self.assertEqual(entry['assessment'], 'equal-under-profile')
def test_audit_rejects_conflicting_profiles_and_excludes_worktree(self):
self.invoke('audit', '--kernel-version', '7.3', '--define', 'LINUX_VERSION_CODE=1', ok=False)
self.invoke('audit', '--define', 'VENDOR=1', '--undef', 'VENDOR', ok=False)
self.invoke('audit', '--kernel-version', 'v7.3', ok=False)
with self.assertRaises(SystemExit):
self.invoke('audit', '--json', '--patch')
(self.scst / (qla2xxx.DEST + 'qla.c')).write_text('uncommitted\n')
entry = json.loads(self.invoke('audit', '--json', '--path', 'qla.c')[0])['files'][0]
self.assertEqual(entry['assessment'], 'identical')
def test_audit_never_hides_modes_binary_or_non_c_files(self):
self.commit(self.scst, 'Add binary and documentation', {
qla2xxx.DEST + 'binary.h': b'\0\xff', qla2xxx.DEST + 'README': 'SCST\n',
qla2xxx.DEST + 'invalid.c': b'\xff'})
(self.scst / (qla2xxx.DEST + 'qla.c')).chmod(0o755)
self.commit(self.scst, 'Mode', {})
entries = {e['path'][len(qla2xxx.DEST):]: e for e in
json.loads(self.invoke('audit', '--json')[0])['files']}
self.assertEqual(entries['qla.c']['assessment'], 'mode-review')
self.assertEqual(entries['binary.h']['assessment'], 'binary-review')
self.assertEqual(entries['invalid.c']['assessment'], 'encoding-review')
self.assertEqual(entries['README']['assessment'], 'non-c-review')
class ProjectionTests(unittest.TestCase):
def project(self, source, *definitions):
return audit.project(audit.source_lines(source), audit.profile('7.3', list(definitions), []))
def test_strings_macro_spacing_and_token_boundaries_survive(self):
def value(source):
return [text for text, _ in audit.tokens(audit.source_lines(source))]
self.assertEqual(value('char *s = "/* a b */"; // outside\n'),
value('char*s="/* a b */"; /* outside */\n'))
self.assertNotEqual(value('char *s="/* a b */";'), value('char *s="/* ab */";'))
self.assertNotEqual(value('x + +y;'), value('x ++ y;'))
self.assertNotEqual(value('#define F(x) x\n'), value('#define F (x) x\n'))
self.assertEqual(value('double n = 1e+3;'), ['double', 'n', '=', '1e+3', ';'])
def test_splicing_precedes_line_comments_and_preserves_line_numbers(self):
source = '// ignored \\\n#if 0\nint value;\n'
projected, _, _ = self.project(source)
self.assertEqual(audit.tokens(projected), [('int', 3), ('value', 3), (';', 3)])
self.assertEqual(audit.tokens(audit.source_lines('\\\nint x;\n'))[0], ('int', 1))
self.assertIn('\\ No newline at end of file\n+new\n', audit.unified('old', 'new', 'qla.c'))
def test_unknown_outer_condition_keeps_decidable_nested_content(self):
source = ('#ifndef HEADER\n#define HEADER\n'
'#if LINUX_VERSION_CODE > KERNEL_VERSION(6, 0, 0)\nnew_api();\n'
'#else\nold_api();\n#endif\n#endif\n')
output, decisions, _ = self.project(source)
text = ''.join(line for line, _ in output)
self.assertIn('#ifndef HEADER', text)
self.assertIn('new_api();', text)
self.assertNotIn('old_api();', text)
self.assertEqual([d['result'] for d in decisions], ['unknown', 'selected'])
def test_local_redefinition_invalidates_external_facts(self):
output, decisions, invalid = self.project('#define VENDOR 0\n#if VENDOR\na();\n'
'#else\nb();\n#endif\n', 'VENDOR=1')
self.assertEqual(invalid, ['VENDOR'])
self.assertEqual(decisions[0]['result'], 'unknown')
self.assertIn('b();', ''.join(line for line, _ in output))
def test_elif_and_short_circuit_do_not_discard_uncertainty(self):
cases = [('#if 0\na();\n#elif 1\nb();\n#else\nc();\n#endif\n', 'b();'),
('#if 0 && UNKNOWN(x)\na();\n#else\nb();\n#endif\n', 'b();'),
('#if UNKNOWN\na();\n#elif 1\nb();\n#endif\n', '#if UNKNOWN')]
for source, expected in cases:
with self.subTest(source=source):
output, _, _ = self.project(source)
self.assertIn(expected, ''.join(line for line, _ in output))
for expr in ('UNKNOWN + 2', '1U < -1', '2147483647 + 1', '0xffffffffffffffff',
'__import__("os")', '1 << 10', 'KERNEL_VERSION(7, 3, 300)'):
self.assertIsNone(audit.Condition(expr, {}, True).value(), expr)
def test_bad_nesting_is_an_error_not_an_empty_success(self):
for source in ('#endif\n', '#if 1\nvalue\n', '#if 0\n#else\n#else\n#endif\n'):
with self.assertRaises(ValueError):
self.project(source)
def test_api_changes_remain_review_candidates(self):
self.assertEqual(audit.classify('timer_shutdown(&t);', 'timer_delete(&t);',
'qla_edif.c')[0], 'behavior-candidate')
self.assertEqual(audit.classify('struct bsg_job *job;', 'BSG_JOB_TYPE *job;',
'qla_bsg.c')[0], 'compatibility-candidate')
self.assertEqual(audit.scope_hint(audit.source_lines(
'static int f(void)\n{\nreturn 1;\n}\n'), 3), 'f')
if __name__ == '__main__':
unittest.main()