publication_tests/test_publisher.py

"""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()