mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-11-01 21:35:25 +08:00
test(core): strengthen handler contracts and report source quality
This commit is contained in:
huanghuoguoguo
@@ -0,0 +1,400 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Report Ruff's McCabe C901 scores, optionally against an explicit Git revision.
|
||||
|
||||
Scope: every Python file in src/langbot, including libs, migrations and templates.
|
||||
No complexity threshold is enforced. Ruff is the sole complexity calculator; AST
|
||||
parsing supplies qualified names, source ranges and change fingerprints only.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import ast
|
||||
from collections import Counter
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path, PurePosixPath
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tarfile
|
||||
import tempfile
|
||||
|
||||
SOURCE = 'src/langbot'
|
||||
|
||||
|
||||
def command(args: list[str], root: Path) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(args, cwd=root, capture_output=True, check=False)
|
||||
|
||||
|
||||
def git(root: Path, *args: str) -> bytes:
|
||||
result = command(['git', *args], root)
|
||||
if result.returncode:
|
||||
raise ValueError(result.stderr.decode(errors='replace').strip())
|
||||
return result.stdout
|
||||
|
||||
|
||||
def index_functions(path: Path) -> dict[int, dict]:
|
||||
"""Identify functions without trying to reimplement Ruff's complexity metric."""
|
||||
functions = {}
|
||||
occurrences = Counter()
|
||||
|
||||
def visit(node: ast.AST, parents: tuple[str, ...] = ()) -> None:
|
||||
if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
parents = (*parents, node.name)
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
name = '.'.join(parents)
|
||||
occurrences[name] += 1
|
||||
functions[node.lineno] = {
|
||||
'name': name,
|
||||
'occurrence': occurrences[name],
|
||||
'line': node.lineno,
|
||||
'end_line': node.end_lineno,
|
||||
'fingerprint': hashlib.sha256(ast.dump(node).encode()).hexdigest(),
|
||||
}
|
||||
for child in ast.iter_child_nodes(node):
|
||||
visit(child, parents)
|
||||
|
||||
visit(ast.parse(path.read_bytes(), filename=str(path)))
|
||||
return functions
|
||||
|
||||
|
||||
def measure(root: Path) -> list[dict]:
|
||||
source = root / SOURCE
|
||||
if not source.is_dir():
|
||||
return []
|
||||
result = command(
|
||||
[
|
||||
sys.executable,
|
||||
'-m',
|
||||
'ruff',
|
||||
'check',
|
||||
'--isolated',
|
||||
'--no-cache',
|
||||
'--select',
|
||||
'C901',
|
||||
'--config',
|
||||
'lint.mccabe.max-complexity=0',
|
||||
'--config',
|
||||
'include=["*.py"]',
|
||||
'--target-version',
|
||||
'py312',
|
||||
'--ignore-noqa',
|
||||
'--no-respect-gitignore',
|
||||
'--exclude',
|
||||
'',
|
||||
'--output-format',
|
||||
'json',
|
||||
SOURCE,
|
||||
],
|
||||
root,
|
||||
)
|
||||
# Exit 1 is expected: max-complexity=0 intentionally reports every function.
|
||||
if result.returncode not in (0, 1):
|
||||
raise ValueError(f'Ruff failed: {result.stderr.decode(errors="replace").strip()}')
|
||||
diagnostics = json.loads(result.stdout)
|
||||
functions = []
|
||||
indexes = {}
|
||||
for diagnostic in diagnostics:
|
||||
match = re.fullmatch(r'`.+` is too complex \((\d+) > 0\)', diagnostic['message'])
|
||||
if diagnostic['code'] != 'C901' or not match:
|
||||
raise ValueError(f'Unexpected Ruff diagnostic: {diagnostic}')
|
||||
path = Path(diagnostic['filename'])
|
||||
if path not in indexes:
|
||||
indexes[path] = index_functions(path)
|
||||
line = diagnostic['location']['row']
|
||||
if line not in indexes[path]:
|
||||
raise ValueError(f'Cannot identify Ruff function at {path}:{line}')
|
||||
functions.append(
|
||||
{
|
||||
'path': path.relative_to(root).as_posix(),
|
||||
**indexes[path][line],
|
||||
'complexity': int(match[1]),
|
||||
}
|
||||
)
|
||||
return sorted(functions, key=lambda item: (item['path'], item['line']))
|
||||
|
||||
|
||||
def baseline(root: Path, ref: str) -> tuple[str, list[dict]]:
|
||||
revision = git(root, 'rev-parse', '--verify', '--end-of-options', f'{ref}^{{commit}}').decode().strip()
|
||||
# Snapshot source only. Never execute baseline code or load baseline config.
|
||||
entries = git(root, 'ls-tree', '--name-only', revision, '--', SOURCE)
|
||||
if not entries:
|
||||
return revision, []
|
||||
archive = git(root, 'archive', '--format=tar', revision, '--', SOURCE)
|
||||
with tempfile.TemporaryDirectory(prefix='langbot-complexity-') as directory:
|
||||
snapshot = Path(directory)
|
||||
with tarfile.open(fileobj=io.BytesIO(archive)) as contents:
|
||||
for member in contents:
|
||||
path = PurePosixPath(member.name)
|
||||
if not member.isfile() or path.suffix != '.py':
|
||||
continue
|
||||
if path.is_absolute() or '..' in path.parts or not path.is_relative_to(SOURCE):
|
||||
raise ValueError(f'Unsafe source archive path: {member.name}')
|
||||
destination = snapshot / path
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
with contents.extractfile(member) as handle:
|
||||
destination.write_bytes(handle.read())
|
||||
return revision, measure(snapshot)
|
||||
|
||||
|
||||
def file_renames(root: Path, revision: str) -> dict[str, str]:
|
||||
parts = git(root, 'diff', '--name-status', '-z', '--find-renames', revision, '--', SOURCE).split(b'\0')
|
||||
renames = {}
|
||||
index = 0
|
||||
while index < len(parts) and parts[index]:
|
||||
status = parts[index].decode()
|
||||
index += 1
|
||||
old_path = parts[index].decode()
|
||||
index += 1
|
||||
if status.startswith(('R', 'C')):
|
||||
new_path = parts[index].decode()
|
||||
index += 1
|
||||
if status.startswith('R'):
|
||||
renames[new_path] = old_path
|
||||
return renames
|
||||
|
||||
|
||||
def function_key(function: dict, path: str | None = None) -> tuple:
|
||||
return path or function['path'], function['name'], function['occurrence']
|
||||
|
||||
|
||||
def compare(before: list[dict], after: list[dict], renames: dict[str, str]) -> list[dict]:
|
||||
"""Do not infer function renames; show unmatched identities as added/deleted."""
|
||||
remaining = {function_key(item): item for item in before}
|
||||
changes = []
|
||||
for current in after:
|
||||
key = function_key(current, renames.get(current['path']))
|
||||
previous = remaining.pop(key, None)
|
||||
if previous is None:
|
||||
status = 'added'
|
||||
elif current['path'] != previous['path']:
|
||||
status = 'renamed_file'
|
||||
elif current['fingerprint'] != previous['fingerprint']:
|
||||
status = 'modified'
|
||||
else:
|
||||
continue
|
||||
changes.append(
|
||||
{
|
||||
'status': status,
|
||||
'before': previous,
|
||||
'after': current,
|
||||
'delta': current['complexity'] - previous['complexity'] if previous else None,
|
||||
}
|
||||
)
|
||||
changes.extend({'status': 'deleted', 'before': item, 'after': None, 'delta': None} for item in remaining.values())
|
||||
return changes
|
||||
|
||||
|
||||
def percentage(covered: int, total: int) -> float | None:
|
||||
return round(100 * covered / total, 2) if total else None
|
||||
|
||||
|
||||
def attach_coverage(functions: list[dict], coverage: dict, root: Path) -> dict:
|
||||
files = {}
|
||||
for filename, data in coverage['files'].items():
|
||||
path = Path(filename)
|
||||
if not path.is_absolute():
|
||||
path = root / path
|
||||
try:
|
||||
relative = path.relative_to(root).as_posix()
|
||||
except ValueError:
|
||||
continue
|
||||
if PurePosixPath(relative).is_relative_to(SOURCE):
|
||||
files[relative] = data
|
||||
branch_enabled = coverage.get('meta', {}).get('branch_coverage', False)
|
||||
for function in functions:
|
||||
data = files.get(function['path'])
|
||||
if data is None:
|
||||
function['coverage'] = None
|
||||
continue
|
||||
start, end = function['line'], function['end_line']
|
||||
excluded = set(data.get('excluded_lines', []))
|
||||
executed = {line for line in data['executed_lines'] if start <= line <= end} - excluded
|
||||
missing = {line for line in data['missing_lines'] if start <= line <= end} - excluded
|
||||
executed_branches = {tuple(arc) for arc in data.get('executed_branches', []) if start <= arc[0] <= end}
|
||||
missing_branches = {tuple(arc) for arc in data.get('missing_branches', []) if start <= arc[0] <= end}
|
||||
function['coverage'] = {
|
||||
'line_percent': percentage(len(executed), len(executed | missing)),
|
||||
'branch_percent': percentage(len(executed_branches), len(executed_branches | missing_branches))
|
||||
if branch_enabled
|
||||
else None,
|
||||
'covered_lines': len(executed),
|
||||
'missing_lines': len(missing),
|
||||
'covered_branches': len(executed_branches) if branch_enabled else None,
|
||||
'missing_branches': len(missing_branches) if branch_enabled else None,
|
||||
}
|
||||
summaries = [data['summary'] for data in files.values()]
|
||||
covered = sum(item['covered_lines'] for item in summaries)
|
||||
statements = sum(item['num_statements'] for item in summaries)
|
||||
covered_branches = sum(item.get('covered_branches', 0) for item in summaries)
|
||||
branches = sum(item.get('num_branches', 0) for item in summaries)
|
||||
return {
|
||||
'matched_source_files': len(files),
|
||||
'functions_without_coverage': sum(item.get('coverage') is None for item in functions),
|
||||
'line_percent': percentage(covered, statements),
|
||||
'branch_percent': percentage(covered_branches, branches) if branch_enabled else None,
|
||||
'covered_lines': covered,
|
||||
'statements': statements,
|
||||
'covered_branches': covered_branches if branch_enabled else None,
|
||||
'branches': branches if branch_enabled else None,
|
||||
'note': 'Coverage covers matching source files only. Function coverage uses inclusive def-to-end source '
|
||||
'ranges; nested functions overlap. Unmatched coverage is unknown, not zero. '
|
||||
'No-branch functions have null branch_percent. Coverage must come from this same checkout.',
|
||||
}
|
||||
|
||||
|
||||
def statistics(functions: list[dict]) -> dict:
|
||||
scores = [item['complexity'] for item in functions]
|
||||
return {
|
||||
'function_count': len(scores),
|
||||
'mean': round(sum(scores) / len(scores), 2) if scores else None,
|
||||
'maximum': max(scores, default=None),
|
||||
'distribution': {
|
||||
'1-5': sum(score <= 5 for score in scores),
|
||||
'6-10': sum(6 <= score <= 10 for score in scores),
|
||||
'11-20': sum(11 <= score <= 20 for score in scores),
|
||||
'21-50': sum(21 <= score <= 50 for score in scores),
|
||||
'51+': sum(score > 50 for score in scores),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def hotspot_key(function: dict) -> tuple:
|
||||
coverage = function.get('coverage') or {}
|
||||
line = coverage.get('line_percent')
|
||||
return -function['complexity'], line if line is not None else 101, function['path'], function['line']
|
||||
|
||||
|
||||
def format_percent(value: float | None) -> str:
|
||||
return f'{value:.2f}%' if value is not None else 'n/a'
|
||||
|
||||
|
||||
def describe(function: dict) -> str:
|
||||
text = f'`{function["path"]}:{function["line"]}` {function["name"]} CC={function["complexity"]}'
|
||||
coverage = function.get('coverage')
|
||||
if coverage:
|
||||
text += (
|
||||
f'; lines {format_percent(coverage["line_percent"])}, branches {format_percent(coverage["branch_percent"])}'
|
||||
)
|
||||
return text
|
||||
|
||||
|
||||
def render_summary(report: dict, limit: int) -> str:
|
||||
stats = report['current']
|
||||
lines = [
|
||||
'# Source complexity report',
|
||||
'',
|
||||
f'Analyzer: {report["analyzer"]} (C901, report only; no complexity gate)',
|
||||
f'Scope: {SOURCE}/**/*.py, including libs, migrations and templates',
|
||||
'',
|
||||
f'Functions: {stats["function_count"]}; mean CC: {stats["mean"]}; maximum CC: {stats["maximum"]}',
|
||||
'Distribution: ' + ', '.join(f'{key}: {value}' for key, value in stats['distribution'].items()),
|
||||
'',
|
||||
]
|
||||
if 'coverage' in report:
|
||||
coverage = report['coverage']
|
||||
lines.extend(
|
||||
[
|
||||
f'Source line coverage: {format_percent(coverage["line_percent"])}; '
|
||||
f'source branch coverage: {format_percent(coverage["branch_percent"])}',
|
||||
f'Matched coverage files: {coverage["matched_source_files"]}; functions with unknown coverage: '
|
||||
f'{coverage["functions_without_coverage"]}',
|
||||
coverage['note'],
|
||||
'',
|
||||
]
|
||||
)
|
||||
if 'baseline' in report:
|
||||
base = report['baseline']
|
||||
lines.extend(
|
||||
[
|
||||
f'## Changes against {report["base_revision"]}',
|
||||
'',
|
||||
f'Baseline functions: {base["function_count"]}; mean CC: {base["mean"]}; maximum CC: {base["maximum"]}',
|
||||
'Baseline distribution: ' + ', '.join(f'{key}: {value}' for key, value in base['distribution'].items()),
|
||||
'Statuses: '
|
||||
+ ', '.join(
|
||||
f'{key}: {value}' for key, value in Counter(c['status'] for c in report['changes']).items()
|
||||
),
|
||||
'Matches use file path, qualified name and duplicate-definition occurrence. Git-detected file renames '
|
||||
'are matched; function renames appear as added/deleted. Changes are AST changes, excluding comments/formatting.',
|
||||
'',
|
||||
]
|
||||
)
|
||||
changes = sorted(report['changes'], key=lambda change: hotspot_key(change['after'] or change['before']))
|
||||
for change in changes[:limit]:
|
||||
delta = f'; delta {change["delta"]:+d}' if change['delta'] is not None else ''
|
||||
lines.append(f'- {change["status"]}: {describe(change["after"] or change["before"])}{delta}')
|
||||
if not changes:
|
||||
lines.append('No function changes in scope.')
|
||||
lines.append('')
|
||||
lines.extend(
|
||||
[f'## Top {limit} hotspots', '', 'Ordered by complexity, then lower known line coverage for ties.', '']
|
||||
)
|
||||
lines.extend(f'- {describe(function)}' for function in report['hotspots'])
|
||||
lines.extend(
|
||||
[
|
||||
'',
|
||||
'Scores are Ruff McCabe values, not a composite quality score. Nested functions are included '
|
||||
'as reported by Ruff. Full function records and changes are in the JSON artifact.',
|
||||
'',
|
||||
]
|
||||
)
|
||||
return '\n'.join(lines)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--root', type=Path, default=Path(__file__).resolve().parents[1])
|
||||
parser.add_argument('--base-ref', help='Explicit commit/ref to compare with the current working tree')
|
||||
parser.add_argument('--coverage-json', type=Path, help='coverage.py JSON generated from this checkout')
|
||||
parser.add_argument('--json', type=Path, default=Path('complexity.json'))
|
||||
parser.add_argument('--summary', type=Path, default=Path('complexity.md'))
|
||||
parser.add_argument('--top', type=int, default=20)
|
||||
args = parser.parse_args(argv)
|
||||
if args.top < 1:
|
||||
parser.error('--top must be positive')
|
||||
root = args.root.resolve()
|
||||
try:
|
||||
if not (root / SOURCE).is_dir():
|
||||
raise ValueError(f'Missing source directory: {root / SOURCE}')
|
||||
version = command([sys.executable, '-m', 'ruff', '--version'], root)
|
||||
if version.returncode:
|
||||
raise ValueError('Ruff is required; run uv sync --dev first')
|
||||
functions = measure(root)
|
||||
report = {
|
||||
'schema_version': 1,
|
||||
'analyzer': version.stdout.decode().strip(),
|
||||
'metric': 'Ruff C901 McCabe complexity',
|
||||
'scope': f'{SOURCE}/**/*.py (including libs)',
|
||||
'current': statistics(functions),
|
||||
'functions': functions,
|
||||
}
|
||||
if args.coverage_json:
|
||||
report['coverage'] = attach_coverage(functions, json.loads(args.coverage_json.read_text()), root)
|
||||
report['hotspots'] = sorted(functions, key=hotspot_key)[: args.top]
|
||||
if args.base_ref:
|
||||
revision, previous = baseline(root, args.base_ref)
|
||||
report.update(
|
||||
{
|
||||
'base_ref': args.base_ref,
|
||||
'base_revision': revision,
|
||||
'baseline': statistics(previous),
|
||||
'changes': compare(previous, functions, file_renames(root, revision)),
|
||||
}
|
||||
)
|
||||
summary = render_summary(report, args.top)
|
||||
for output in (args.json, args.summary):
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.json.write_text(json.dumps(report, indent=2) + '\n')
|
||||
args.summary.write_text(summary)
|
||||
print(summary)
|
||||
return 0
|
||||
except (OSError, ValueError, SyntaxError, KeyError, tarfile.TarError) as error:
|
||||
print(f'Complexity report failed: {error}', file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Resolve the latest official LocalAgent from Space and prepare its E2E fixture.
|
||||
|
||||
Resolution always contacts Space. Cache only the version/checksum-resolved
|
||||
archive, preserving its certification envelope for normal Runtime admission.
|
||||
The saved resolution JSON can also replay the exact package used by a CI run.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import zipfile
|
||||
|
||||
import httpx
|
||||
import yaml
|
||||
|
||||
SPACE_URL = 'https://space.langbot.app'
|
||||
PLUGIN_AUTHOR = 'langbot-team'
|
||||
PLUGIN_NAME = 'LocalAgent'
|
||||
VERSIONS_URL = f'{SPACE_URL}/api/v1/marketplace/plugins/{PLUGIN_AUTHOR}/{PLUGIN_NAME}/versions'
|
||||
MAX_PACKAGE_BYTES = 16 * 1024 * 1024
|
||||
MAX_METADATA_BYTES = 1024 * 1024
|
||||
|
||||
|
||||
def download_package(url: str, *, max_bytes: int = MAX_PACKAGE_BYTES) -> bytes:
|
||||
"""Use Core's HTTP client with bounded responses and network retries."""
|
||||
for attempt in range(3):
|
||||
try:
|
||||
with httpx.stream('GET', url, timeout=30, follow_redirects=True) as response:
|
||||
response.raise_for_status()
|
||||
body = bytearray()
|
||||
for chunk in response.iter_bytes():
|
||||
body.extend(chunk)
|
||||
if len(body) > max_bytes:
|
||||
raise ValueError(f'Space response exceeds the {max_bytes}-byte limit')
|
||||
return bytes(body)
|
||||
except httpx.HTTPError as exc:
|
||||
if attempt == 2:
|
||||
raise RuntimeError(f'Cannot fetch LocalAgent fixture from Space: {url}: {exc}') from exc
|
||||
time.sleep(attempt + 1)
|
||||
raise AssertionError('Unreachable download retry state')
|
||||
|
||||
|
||||
def resolve_latest() -> dict:
|
||||
"""Select the first published version, matching Core's marketplace resolver."""
|
||||
payload = json.loads(download_package(VERSIONS_URL, max_bytes=MAX_METADATA_BYTES))
|
||||
if payload.get('code') != 0:
|
||||
raise ValueError('Space did not return a successful plugin version response')
|
||||
versions = payload.get('data', {}).get('versions', [])
|
||||
if not versions or not isinstance(versions, list) or not isinstance(versions[0], dict):
|
||||
raise ValueError('Space returned no LocalAgent versions')
|
||||
latest = versions[0]
|
||||
version = latest.get('version', '')
|
||||
checksum = latest.get('checksum', '')
|
||||
if not isinstance(version, str) or not re.fullmatch(r'[A-Za-z0-9._+-]+', version):
|
||||
raise ValueError('Space returned an invalid LocalAgent version')
|
||||
if not isinstance(checksum, str) or not re.fullmatch(r'[a-fA-F0-9]{64}', checksum):
|
||||
raise ValueError('Space did not provide a valid archive SHA256 checksum')
|
||||
if latest.get('plugin_id') != f'{PLUGIN_AUTHOR}/{PLUGIN_NAME}' or latest.get('status') != 'live':
|
||||
raise ValueError('Latest Space entry is not the published LocalAgent plugin')
|
||||
return {
|
||||
'author': PLUGIN_AUTHOR,
|
||||
'name': PLUGIN_NAME,
|
||||
'version': version,
|
||||
'url': f'{SPACE_URL}/api/v1/marketplace/plugins/download/{PLUGIN_AUTHOR}/{PLUGIN_NAME}/{version}',
|
||||
'sha256': checksum.lower(),
|
||||
'resolved_at': datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
||||
'versions_url': VERSIONS_URL,
|
||||
}
|
||||
|
||||
|
||||
def validate_package(package: bytes, resolution: dict) -> None:
|
||||
"""Check the original archive bytes and manifest identity before installation."""
|
||||
actual = hashlib.sha256(package).hexdigest()
|
||||
if actual != resolution['sha256']:
|
||||
raise ValueError(f'LocalAgent SHA256 mismatch: expected {resolution["sha256"]}, got {actual}')
|
||||
with zipfile.ZipFile(io.BytesIO(package)) as archive:
|
||||
manifest = yaml.safe_load(archive.read('manifest.yaml'))
|
||||
metadata = manifest['metadata']
|
||||
for field in ('author', 'name', 'version'):
|
||||
if str(metadata[field]) != resolution[field]:
|
||||
raise ValueError(f'LocalAgent manifest {field} does not match the resolved artifact')
|
||||
|
||||
|
||||
def fetch_package(destination: Path, resolution: dict) -> Path:
|
||||
"""Reuse only a verified resolved archive or download it atomically."""
|
||||
if destination.is_file():
|
||||
try:
|
||||
validate_package(destination.read_bytes(), resolution)
|
||||
return destination
|
||||
except (ValueError, KeyError, zipfile.BadZipFile):
|
||||
print('Cached LocalAgent package does not match this run; downloading resolved artifact', file=sys.stderr)
|
||||
|
||||
package = download_package(resolution['url'])
|
||||
validate_package(package, resolution)
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = destination.with_suffix('.tmp')
|
||||
temporary.write_bytes(package)
|
||||
temporary.replace(destination)
|
||||
return destination
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
commands = parser.add_subparsers(dest='command', required=True)
|
||||
resolve = commands.add_parser('resolve', help='Query Space latest and save replay metadata; never cached')
|
||||
resolve.add_argument('resolution', type=Path)
|
||||
resolve.add_argument('--github-output', type=Path, help='Write the resolved artifact cache key for Actions')
|
||||
fetch = commands.add_parser('fetch', help='Download/verify the artifact described by saved resolution metadata')
|
||||
fetch.add_argument('resolution', type=Path)
|
||||
fetch.add_argument('destination', type=Path)
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
if args.command == 'resolve':
|
||||
resolution = resolve_latest()
|
||||
args.resolution.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.resolution.write_text(json.dumps(resolution, indent=2) + '\n', encoding='utf-8')
|
||||
if args.github_output:
|
||||
with args.github_output.open('a', encoding='utf-8') as output:
|
||||
output.write(f'cache_key={resolution["version"]}-{resolution["sha256"]}\n')
|
||||
print(json.dumps(resolution, indent=2))
|
||||
else:
|
||||
resolution = json.loads(args.resolution.read_text(encoding='utf-8'))
|
||||
package = fetch_package(args.destination.resolve(), resolution)
|
||||
print(f'{resolution["author"]}/{resolution["name"]}:{resolution["version"]} verified at {package}')
|
||||
print(f'Archive SHA256: {resolution["sha256"]}')
|
||||
except (OSError, ValueError, KeyError, RuntimeError, zipfile.BadZipFile) as exc:
|
||||
parser.exit(1, f'LocalAgent fixture provisioning failed: {exc}\n')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user