import io
import json
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace as S
from unittest.mock import patch
import full_collect as f

class Tests(unittest.TestCase):
    def test_no_1100_cap_and_zero_votes(self):
        class API:
            def kernels_list_with_response(self,**kw):
                i=(kw['page']-1)*100
                return S(kernels=[S(ref='u/n'+str(n),title='t',total_votes=0) for n in range(i,min(i+100,1201))],next_page_token=None)
        with patch.object(f.old,'call_sdk',side_effect=lambda fn,what:fn()):r,c=f.old.list_kernels(API(),'test')
        self.assertEqual(len(r),1201);self.assertFalse(c);self.assertEqual(r[0]['votes'],0)
    def test_repeated_kernel_page_not_complete(self):
        api=S(kernels_list_with_response=lambda **kw:S(kernels=[S(ref='u/'+str(n),title='t',total_votes=1) for n in range(100)],next_page_token=None))
        with patch.object(f.old,'call_sdk',side_effect=lambda fn,what:fn()):
            with self.assertRaises(f.old.Incomplete):f.old.list_kernels(api,'test')
    def test_board_association_and_full_count(self):
        raw=dict(publicLeaderboard=[dict(teamId=n,rank=n,displayScore='1',medal='GOLD') for n in range(1,151)],privateLeaderboard=[],teams=[dict(teamId=1,writeUpForumTopicId=99,solutionWriteUpUrl='/writeup')])
        r=f.normalize_board(raw,150);self.assertEqual(r['status'],'ok');self.assertEqual(len(r['publicLeaderboard']),150)
        self.assertEqual(r['publicLeaderboard'][0]['solution_topic_id'],99)
        self.assertEqual(f.normalize_board(raw,151)['status'],'partial')
    def test_existing_source_never_downloaded_or_removed(self):
        with tempfile.TemporaryDirectory() as t:
            dest=Path(t);p=dest/'source.py';p.write_text('print(1)')
            class API:
                def kernels_pull(self,*a,**kw):raise AssertionError('downloaded twice')
            f.safe_pull(API(),'u/source',dest)
            self.assertEqual(p.read_text(),'print(1)')
    def test_invalid_original_not_overwritten(self):
        with tempfile.TemporaryDirectory() as t:
            root=Path(t);dest=root/'folder';dest.mkdir();original=dest/'code.ipynb';original.write_text('<html>wrong file</html>')
            src=root/'source.ipynb';src.write_text(json.dumps({'cells':[]}))
            with patch.object(f.old,'local_kernel_file',return_value=src):f.safe_pull(None,'u/code',dest,'v1')
            self.assertEqual(original.read_text(),'<html>wrong file</html>');self.assertEqual(len(f.sources(dest)),1)
    def test_disk_reserve_before_body_write(self):
        with tempfile.TemporaryDirectory() as t:
            p=Path(t)/'.repair-download'/'model.tar.gz'
            response=S(headers={'Content-Length':'10'},iter_content=lambda n:iter([b'1234567890']))
            with patch.object(f,'room',return_value=False):
                with self.assertRaises(f.old.DiskLimit):f.safe_sdk_download(None,response,str(p),None)
            self.assertFalse(p.exists())
    def test_html_notebook_not_valid_source(self):
        with tempfile.TemporaryDirectory() as t:
            p=Path(t)/'bad.ipynb';p.write_text('<html>login</html>');self.assertFalse(f.source_ok(p))
    def test_stream_reserve_keeps_partial(self):
        with tempfile.TemporaryDirectory() as t:
            p=Path(t)/'.repair-download'/'model.tar.gz'
            response=S(headers={'Content-Length':'2','ETag':'same-object'},iter_content=lambda n:iter([b'a',b'b']),close=lambda:None,raise_for_status=lambda:None)
            with patch.object(f,'room',side_effect=[True,True,False]):
                with self.assertRaises(f.old.DiskLimit):f.safe_sdk_download(None,response,str(p),None)
            self.assertEqual(p.read_bytes(),b'a')
            self.assertEqual(f.read(p.with_name(p.name+'.resume.json'))['validator'],'same-object')
    def test_empty_envelopes_not_complete(self):
        with self.assertRaises(f.old.Incomplete):f.old.collect_models(lambda token:(200,'{}'),sleep=lambda _:None)
        with self.assertRaises(f.old.Incomplete):f.old.collect_topics(lambda page:(200,'{}'),sleep=lambda _:None)
    def test_attachment_url_entities_and_github(self):
        self.assertEqual(f.asset_url('https://www.googleapis.com/x.png?generation=1&amp;alt=media'),'https://www.googleapis.com/x.png?generation=1&alt=media')
        self.assertEqual(f.asset_url('https://github.com/u/r/blob/main/file.py#L10'),'https://raw.githubusercontent.com/u/r/main/file.py')
        self.assertEqual(f.asset_url('https://storage.cloud.google.com/b/file.jpg'),'https://storage.googleapis.com/b/file.jpg')
    def test_model_version_uses_exact_registered_archive(self):
        import tarfile
        with tempfile.TemporaryDirectory() as t:
            dest=Path(t)
            for name,data in [('old.tar.gz',b'old'),('new.tar.gz',b'new')]:
                with tarfile.open(dest/name,'w:gz') as archive:
                    info=tarfile.TarInfo('weights');info.size=len(data);archive.addfile(info,io.BytesIO(data))
            ref='u/model/PyTorch/default/2'
            (dest/'.ok').write_text(ref)
            (dest/'.model-files.json').write_text(json.dumps({ref:{'file':'new.tar.gz','bytes':(dest/'new.tar.gz').stat().st_size}}))
            self.assertEqual(f.safe_model(None,ref,dest),dest/'new.tar.gz')
    def test_decoded_http_body_not_compared_with_compressed_length(self):
        with tempfile.TemporaryDirectory() as t:
            p=Path(t)/'.repair-download'/'code.py'
            response=S(headers={'Content-Length':'4','Content-Encoding':'gzip','ETag':'x'},iter_content=lambda n:iter([b'print(1)']),close=lambda:None,raise_for_status=lambda:None)
            with patch.object(f,'room',return_value=True):f.safe_sdk_download(None,response,str(p),None,heavy=False)
            self.assertEqual(p.read_bytes(),b'print(1)')
if __name__=='__main__':unittest.main()