File size: 10,498 Bytes
9d6c005 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | """Offline tests for publication boundaries; not rendering-quality tests."""
import contextlib
import hashlib
import importlib.util
import io
import json
import os
from pathlib import Path
import socket
import sys
import tempfile
from types import SimpleNamespace as NS
import unittest
from unittest.mock import Mock, patch
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('publisher', ROOT/'scripts/publish_hf.py')
p = importlib.util.module_from_spec(spec)
spec.loader.exec_module(p)
def sibling(name, data, lfs=False):
return NS(rfilename=name, size=len(data), lfs=NS(sha256=hashlib.sha256(data).hexdigest()) if lfs else None,
blob_id=hashlib.sha1(b'blob '+str(len(data)).encode()+b'\0'+data).hexdigest())
class Missing(Exception):
pass
class Service:
def __init__(self, existing=False, contents=None):
self.exists=existing;self.contents=contents or {};self.sha='b'*40
self.owner='PureOne';self.private=False;self.gated=False
self.created=[];self.commits=[];self.reads=[];self.auth_checks=[]
def whoami(self):return {'name':self.owner}
def repo_info(self, repo, **kw):
self.reads.append(kw)
if not self.exists:raise Missing()
return NS(private=self.private,gated=self.gated,sha=self.sha,
siblings=[sibling(n,b) for n,b in self.contents.items()])
def create_repo(self, **kw):
self.created.append(kw);self.exists=True
def auth_check(self, **kw):self.auth_checks.append(kw)
def create_commit(self, **kw):
assert kw['parent_commit']==self.sha
self.commits.append(kw)
for op in kw['operations']:self.contents[op.path_in_repo]=Path(op.path_or_fileobj).read_bytes()
self.sha='c'*40
return NS(oid=self.sha)
class PublicationTests(unittest.TestCase):
def setUp(self):
self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name)/'source';self.root.mkdir()
(self.root/'README.md').write_text('Release fixture.\n')
self.write_manifest();self.files=p.release_files(self.root)
self.work=Path(self.temp.name)/'work';self.work.mkdir()
def tearDown(self):self.temp.cleanup()
def write_manifest(self, entries=None):
entries=entries or [(p.sha256(self.root/'README.md'),'README.md')]
(self.root/'CHECKSUMS.sha256').write_text(''.join(h+' '+n+'\n' for h,n in entries))
def download(self, **kw):
self.assertIs(kw['token'],False);self.assertEqual(kw['endpoint'],p.ENDPOINT)
return str(self.root/'CHECKSUMS.sha256')
def run_publish(self, service, public=None, download=None):
return p.publish(self.files,p.DEFAULT_REPO,service,public or service,Missing,NS,download or self.download,self.work)
def test_modified_local_bytes_rejected(self):
(self.root/'README.md').write_text('changed')
with self.assertRaises(p.ReleaseError):p.release_files(self.root)
def test_unsafe_paths_rejected(self):
for name in ['../escape','/absolute','A/../B','a\\b','file:stream','.env','a.token','a//b','x/','publication_receipt.json','CHECKSUMS.sha256']:
with self.subTest(name=name):self.assertFalse(p.safe_name(name))
def test_case_colliding_manifest_rejected(self):
h=p.sha256(self.root/'README.md');self.write_manifest([(h,'README.md'),(h,'readme.md')])
with self.assertRaises(p.ReleaseError):p.release_files(self.root)
def test_symlink_rejected(self):
target=self.root/'target';target.write_text('abc');link=self.root/'linked'
try:link.symlink_to(target)
except OSError:self.skipTest('OS does not allow symlink creation')
self.write_manifest([(p.sha256(target),'linked')])
with self.assertRaises(p.ReleaseError):p.release_files(self.root)
def test_only_manifest_files_staged(self):
(self.root/'.env').write_text('a local private fixture')
(self.root/'publication_receipt.json').write_text('{}')
stage=self.work/'stage';copied=p.copy_release(self.files,stage)
self.assertEqual(set(copied),set(self.files));self.assertFalse((stage/'.env').exists())
def test_staging_race_rejected(self):
(self.root/'README.md').write_text('changed after preflight')
with self.assertRaises(p.ReleaseError):p.copy_release(self.files,self.work/'stage')
def test_dry_run_never_reads_token_or_network(self):
with patch.object(socket,'socket',side_effect=AssertionError('network')),patch.object(p.getpass,'getpass',side_effect=AssertionError('token')),contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(p.main([],root=self.root),0)
def test_owner_match_is_case_insensitive(self):p.check_owner({'name':'pureone'},p.DEFAULT_REPO)
def test_wrong_owner_stops_before_creation(self):
s=Service();s.owner='SomeoneElse'
with self.assertRaises(p.ReleaseError):self.run_publish(s)
self.assertFalse(s.created);self.assertFalse(s.reads)
def test_private_target_not_changed(self):
s=Service(existing=True);s.private=True
with self.assertRaises(p.ReleaseError):self.run_publish(s)
self.assertTrue(s.private);self.assertFalse(s.commits)
def test_gated_target_rejected(self):
s=Service(existing=True);s.gated='auto'
with self.assertRaises(p.ReleaseError):self.run_publish(s)
self.assertFalse(s.commits)
def test_unrelated_target_not_overwritten(self):
s=Service(existing=True,contents={'other.txt':b'other project'})
with self.assertRaises(p.ReleaseError):self.run_publish(s)
self.assertFalse(s.commits)
def test_changed_remote_bytes_not_overwritten(self):
s=Service(existing=True,contents={'README.md':b'different'})
with self.assertRaises(p.ReleaseError):self.run_publish(s)
self.assertFalse(s.commits)
def test_new_public_repo_parent_guard_and_pinned_verification(self):
s=Service();receipt=self.run_publish(s)
self.assertFalse(s.created[0]['private']);self.assertFalse(s.created[0]['exist_ok'])
self.assertEqual(s.commits[0]['parent_commit'],'b'*40)
self.assertEqual(s.reads[-1]['revision'],'c'*40)
self.assertTrue(receipt['publication_verified']);self.assertEqual(receipt['verified_release_files'],2)
def test_partial_matching_upload_resumes_missing_files(self):
s=Service(existing=True,contents={'README.md':(self.root/'README.md').read_bytes()})
self.run_publish(s)
self.assertEqual([op.path_in_repo for op in s.commits[0]['operations']],['CHECKSUMS.sha256'])
self.assertFalse(s.created)
def test_complete_upload_is_no_op_and_still_verified(self):
s=Service(existing=True,contents={n:f.read_bytes() for n,f in self.files.items()})
receipt=self.run_publish(s)
self.assertFalse(s.commits);self.assertEqual(receipt['revision'],'b'*40)
def test_wrong_public_revision_fails(self):
s=Service();public=Mock();public.repo_info.return_value=NS(sha='d'*40)
with self.assertRaises(p.ReleaseError):self.run_publish(s,public=public)
def test_missing_public_file_fails(self):
s=Service();public=Mock();public.repo_info.return_value=NS(sha='c'*40,private=False,gated=False,siblings=[])
with self.assertRaises(p.ReleaseError):self.run_publish(s,public=public)
def test_corrupted_public_hash_fails(self):
s=Service();public=Mock();public.repo_info.return_value=NS(sha='c'*40,private=False,gated=False,siblings=[sibling('README.md',b'bad')])
with self.assertRaises(p.ReleaseError):self.run_publish(s,public=public)
def test_lfs_and_git_hash_paths(self):
f=self.root/'README.md';data=f.read_bytes()
self.assertTrue(p.remote_matches(sibling('README.md',data),f))
self.assertTrue(p.remote_matches(sibling('README.md',data,lfs=True),f))
self.assertFalse(p.remote_matches(sibling('README.md',b'bad',lfs=True),f))
def test_public_manifest_download_mismatch_fails(self):
wrong=self.work/'wrong';wrong.write_text('not the manifest')
with self.assertRaises(p.ReleaseError):self.run_publish(Service(),download=lambda **kw:str(wrong))
def test_concurrent_commit_failure_is_not_success(self):
s=Service();s.create_commit=Mock(side_effect=RuntimeError('concurrent change'))
with self.assertRaises(RuntimeError):self.run_publish(s)
self.assertFalse((self.root/'publication_receipt.json').exists())
def test_server_attributes_are_allowed_without_upload(self):
s=Service(existing=True,contents={'.gitattributes':b'server generated'})
self.assertTrue(self.run_publish(s)['publication_verified'])
self.assertNotIn('.gitattributes',[op.path_in_repo for op in s.commits[0]['operations']])
def mock_sdk(self, service, factory):
return patch.dict(sys.modules,{
'huggingface_hub':NS(HfApi=factory,CommitOperationAdd=NS,hf_hub_download=self.download),
'huggingface_hub.errors':NS(RepositoryNotFoundError=Missing)})
def test_secure_input_warning_aborts_before_authenticated_calls(self):
import warnings
factory=Mock()
def unsafe_prompt(*args):warnings.warn('no secure console',p.getpass.GetPassWarning)
with self.mock_sdk(None,factory),patch.dict(os.environ,{'HF_TOKEN':''}),patch.object(p.getpass,'getpass',side_effect=unsafe_prompt),contextlib.redirect_stdout(io.StringIO()):
with self.assertRaises(p.getpass.GetPassWarning):p.main(['--publish'],root=self.root)
factory.assert_not_called()
def test_prompted_token_is_not_written_and_public_client_is_anonymous(self):
s=Service();factory=Mock(return_value=s);output=io.StringIO();secret='unit-test-private-input'
with self.mock_sdk(s,factory),patch.dict(os.environ,{'HF_TOKEN':''}),patch.object(p.getpass,'getpass',return_value=secret),contextlib.redirect_stdout(output):
self.assertEqual(p.main(['--publish'],root=self.root),0)
self.assertEqual(factory.call_args_list[0].kwargs['token'],secret)
self.assertIs(factory.call_args_list[1].kwargs['token'],False)
self.assertNotIn(secret,output.getvalue())
self.assertTrue(json.loads((self.root/'publication_receipt.json').read_text())['publication_verified'])
for file in self.root.rglob('*'):
if file.is_file():self.assertNotIn(secret.encode(),file.read_bytes())
if __name__=='__main__':unittest.main()
|