cesnimda
85 lines
4.7 KiB
Python
85 lines
4.7 KiB
Python
"""Actual pinned trained-checkpoint CPU compatibility; human quality remains separate."""
|
|
from pathlib import Path
|
|
import importlib.metadata
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import resource
|
|
import sys
|
|
import time
|
|
sys.path.insert(0, '/qualification')
|
|
from verify import verify_fixture
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
manifest_bytes = (HERE / 'ARTIFACTS.json').read_bytes()
|
|
manifests = json.loads(manifest_bytes)
|
|
manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest()
|
|
kind = sys.argv[1]
|
|
fixture = verify_fixture(Path(sys.argv[2]), manifests[kind])
|
|
assert os.environ['HF_HUB_OFFLINE'] == os.environ['TRANSFORMERS_OFFLINE'] == '1'
|
|
manifest = manifests[kind]
|
|
repository_name = 'sshleifer/distilbart-cnn-12-6'
|
|
assert manifest['fixture'] == 'actual-distilbart-cnn-12-6'
|
|
cache_home = Path('/tmp/qualified-hf-cache')
|
|
cache_home.mkdir(mode=0o700)
|
|
os.environ['HF_HOME'] = str(cache_home)
|
|
cache = cache_home / 'hub'
|
|
repository = cache / ('models--' + repository_name.replace('/', '--'))
|
|
snapshot = repository / 'snapshots' / manifest['sourceRevision']
|
|
snapshot.mkdir(parents=True)
|
|
(repository / 'blobs').mkdir()
|
|
for entry in manifest['files']:
|
|
identity = entry.get('gitBlobId') or entry['sha256']
|
|
blob = repository / 'blobs' / identity
|
|
blob.symlink_to(fixture / entry['file'])
|
|
(snapshot / entry['file']).symlink_to(Path('../../blobs') / identity)
|
|
os.environ.pop('AI_SERVICE_SKIP_MODEL_LOAD', None)
|
|
os.environ.pop('AI_SERVICE_EAGER_MODEL_LOAD', None)
|
|
os.environ['AI_SERVICE_TOKEN'] = 'p9-local-qualification-only'
|
|
sys.path.insert(0, '/app')
|
|
import torch
|
|
import app
|
|
from fastapi.testclient import TestClient
|
|
|
|
torch.set_num_threads(1)
|
|
started = time.monotonic()
|
|
from huggingface_hub import try_to_load_from_cache
|
|
assert app.MODEL_NAME == repository_name
|
|
assert app.MODEL_REVISION == manifest['sourceRevision']
|
|
for entry in manifest['files']:
|
|
cached = try_to_load_from_cache(app.MODEL_NAME, entry['file'], revision=app.MODEL_REVISION)
|
|
assert isinstance(cached, str) and Path(cached).resolve() == fixture / entry['file']
|
|
assert try_to_load_from_cache(app.MODEL_NAME, 'absent-qualification-file', revision=app.MODEL_REVISION) is None
|
|
assert app._ensure_runtime_loaded(), app.MODEL_LOAD_ERROR
|
|
assert app.model.__class__.__name__ == 'BartForConditionalGeneration'
|
|
assert (app.model.config.d_model, app.model.config.encoder_layers, app.model.config.decoder_layers, app.model.config.vocab_size) == (1024, 12, 6, 50264)
|
|
assert len(app.tokenizer) == 50265
|
|
assert app.model.training is False
|
|
assert str(app.device) == 'cpu' and not app.GPU_AVAILABLE
|
|
text = 'Åse Østby builds accessible Python and C# applications for reliable Norwegian career workflows.'
|
|
encoded = app.tokenizer(text, return_tensors='pt')
|
|
assert app.tokenizer.decode(encoded.input_ids[0], skip_special_tokens=True) == text
|
|
summary = app._model_summarize(text, 24, 8)
|
|
assert isinstance(summary, str) and summary
|
|
long_input = app.tokenizer(text * 400, return_tensors='pt', truncation=True, max_length=1024)
|
|
assert long_input.input_ids.shape == (1, 1024)
|
|
assert long_input.input_ids.dtype == torch.int64
|
|
with torch.no_grad():
|
|
logits = app.model(**long_input, decoder_input_ids=torch.tensor([[2, 0, 10]])).logits
|
|
assert torch.isfinite(logits).all()
|
|
assert app._model_summarize(text * 400, 24, 8)
|
|
client = TestClient(app.app)
|
|
assert client.get('/health').json()['model_loaded'] is True
|
|
assert client.post('/summarize', json={'text': text}).status_code == 401
|
|
headers = {'X-Ai-Service-Token': 'p9-local-qualification-only'}
|
|
body = {'text': text, 'max_length': 24, 'min_length': 8}
|
|
first = client.post('/summarize', json=body, headers=headers)
|
|
second = client.post('/summarize', json=body, headers=headers)
|
|
assert first.status_code == second.status_code == 200, first.text
|
|
assert first.json()['cached'] is False and second.json()['cached'] is True
|
|
assert first.json()['summary'] == second.json()['summary']
|
|
verify_fixture(fixture, manifests[kind])
|
|
elapsed = time.monotonic() - started
|
|
assert elapsed < 240
|
|
print(json.dumps({'status': 'PASS', 'kind': kind, 'scope': 'Actual pinned trained DistilBART CPU loading/tokenizer/generation and summary/auth/cache compatibility; no human semantic-quality/GPU/production claim.', 'manifestSha256': manifest_sha256, 'repositoryRevision': app.MODEL_REVISION, 'modelClass': app.model.__class__.__name__, 'device': str(app.device), 'cudaAvailable': torch.cuda.is_available(), 'seconds': elapsed, 'maxRssKiB': resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, 'versions': {name: importlib.metadata.version(name) for name in ['torch','transformers','filelock','huggingface-hub','tokenizers','safetensors']}, 'summary': summary, 'bounded1024Tokens': True, 'finiteLogits': True, 'cacheRepeatedRead': True, 'fixtureUnchanged': True}), flush=True)
|