import copy
import json
import tempfile
import threading
import unittest
import xml.etree.ElementTree as ET
from urllib.error import HTTPError
from urllib.request import Request,urlopen
from quality import Quality,Descriptors,parse_mpd,filtered_manifest,available_heights,local
from provider import ProviderError
from server import make_server
from catalog import Addon

MPD=b'''<?xml version="1.0"?><MPD xmlns="urn:mpeg:dash:schema:mpd:2011" type="static"><Period>
<AdaptationSet contentType="video" mimeType="video/mp4" maxHeight="1080">
<Representation id="v1080" width="1920" height="1080"><SegmentList><Initialization sourceURL="1080/init.m4s"/><SegmentURL media="1080/1.m4s"/></SegmentList></Representation>
<Representation id="v720" width="1280" height="720"><SegmentList><Initialization sourceURL="720/init.m4s"/><SegmentURL media="720/1.m4s"/></SegmentList></Representation>
<Representation id="v480" width="854" height="480"><SegmentList><Initialization sourceURL="480/init.m4s"/><SegmentURL media="480/1.m4s"/></SegmentList></Representation>
</AdaptationSet><AdaptationSet contentType="audio" lang="en"><Representation id="audio" bandwidth="128000"/></AdaptationSet>
<AdaptationSet contentType="text" lang="en"><Representation id="sub"/></AdaptationSet></Period></MPD>'''
ORIGIN='https://cdn.example.invalid/dash/title/index.mpd'
SOURCE={'name':'MovieBox','title':'Example (2000)\n1080,720,480 | hevc','url':ORIGIN,'behaviorHints':{'filename':'Example (2000)','proxyHeaders':{'request':{'Cookie':'signed=example','User-Agent':'test'}}}}


class QualityTests(unittest.TestCase):
    def test_filters_actual_tracks_for_each_height(self):
        for height in (1080,720,480):
            root=parse_mpd(filtered_manifest(MPD,ORIGIN,height))
            video=[e for e in root.iter() if local(e.tag)=='Representation' and e.get('height')]
            self.assertEqual([int(e.get('height')) for e in video],[height])
            self.assertTrue(any(e.get('id')=='audio' for e in root.iter()))
            self.assertTrue(any(e.get('id')=='sub' for e in root.iter()))
            self.assertEqual(next(e.text for e in root if local(e.tag)=='BaseURL'),'https://cdn.example.invalid/dash/title/')
            media=[e.get('media') for e in root.iter() if local(e.tag)=='SegmentURL']
            self.assertEqual(media,[str(height)+'/1.m4s'])

    def test_missing_height_is_rejected_not_silently_substituted(self):
        with self.assertRaises(LookupError): filtered_manifest(MPD,ORIGIN,2160)

    def test_existing_relative_base_and_location(self):
        root=parse_mpd(MPD)
        base=ET.Element('{urn:mpeg:dash:schema:mpd:2011}BaseURL'); base.text='../media/'; root.insert(0,base)
        location=ET.SubElement(root,'{urn:mpeg:dash:schema:mpd:2011}Location'); location.text='https://cdn.example.invalid/adaptive.mpd'
        result=parse_mpd(filtered_manifest(ET.tostring(root),ORIGIN,720))
        self.assertEqual(next(e.text for e in result if local(e.tag)=='BaseURL'),'https://cdn.example.invalid/dash/media/')
        self.assertFalse(any(local(e.tag)=='Location' for e in result))

    def test_multi_period_only_common_resolutions_exposed(self):
        root=parse_mpd(MPD); period=copy.deepcopy(next(e for e in root if local(e.tag)=='Period'))
        for a in period:
            for e in list(a):
                if e.get('height')=='1080': a.remove(e)
        root.append(period)
        self.assertEqual(available_heights(root),[720,480])

    def test_hosted_choices_are_distinct_and_preserve_headers(self):
        calls=[]
        def fetch(url,headers): calls.append(1); return MPD,ORIGIN
        quality=Quality(fetch=fetch)
        choices=quality.expand([SOURCE,SOURCE],'https://addon.example/private')
        self.assertEqual([s['name'] for s in choices],['MovieBox Auto up to 1080p'])
        self.assertEqual(len({s['url'] for s in choices}),1)
        self.assertEqual(len(calls),1)
        for source in choices:
            self.assertNotIn('signed=example',source['url'])
            self.assertEqual(source['behaviorHints']['proxyHeaders'],SOURCE['behaviorHints']['proxyHeaders'])
            body=quality.manifest(source['url'].replace('https://addon.example/private',''))
            selected=int(source['name'].split()[-1][:-1])
            self.assertEqual(available_heights(parse_mpd(body)),[h for h in (1080,720,480) if h<=selected] if 'Auto up to' in source['name'] else [selected])

    def test_adaptive_cap_retains_audio_and_excludes_1080(self):
        root=parse_mpd(filtered_manifest(MPD,ORIGIN,720,adaptive=True))
        self.assertEqual(available_heights(root),[720,480])
        self.assertTrue(any(e.get('id')=='audio' for e in root.iter()))
        self.assertTrue(any(e.get('id')=='sub' for e in root.iter()))
        adaptation=next(e for e in root.iter() if e.get('contentType')=='video')
        self.assertEqual((adaptation.get('minHeight'),adaptation.get('maxHeight')),('480','720'))
        self.assertNotIn('1080/1.m4s',str(ET.tostring(root)))

    def test_no_adaptive_choice_without_lower_fallback(self):
        root=parse_mpd(MPD)
        for adaptation in root.iter():
            for rep in list(adaptation):
                if rep.get('height') in ('1080','480'): adaptation.remove(rep)
        data=ET.tostring(root)
        quality=Quality(fetch=lambda url,headers:(data,ORIGIN))
        self.assertEqual([s['name'] for s in quality.expand([SOURCE],'https://addon.example/private')],['MovieBox Auto up to 720p'])
        with self.assertRaises(LookupError): filtered_manifest(data,ORIGIN,720,adaptive=True)

    def test_auto1080_cap_excludes_4k_and_keeps_lower_tracks(self):
        root=parse_mpd(MPD)
        adaptation=next(e for e in root.iter() if e.get('contentType')=='video')
        ultra=copy.deepcopy(next(e for e in adaptation if e.get('height')=='1080'))
        ultra.set('id','v2160'); ultra.set('height','2160'); adaptation.append(ultra)
        quality=Quality(fetch=lambda url,headers:(ET.tostring(root),ORIGIN))
        choices=quality.expand([SOURCE],'https://addon.example/private')
        auto=next(s for s in choices if s['name']=='MovieBox Auto up to 1080p')
        body=quality.manifest(auto['url'].replace('https://addon.example/private',''))
        self.assertEqual(available_heights(parse_mpd(body)),[1080,720,480])
        self.assertIn(b'audio',body)

    def test_descriptors_survive_restart_and_cookie_is_not_in_id(self):
        with tempfile.TemporaryDirectory() as temp:
            payload={'url':ORIGIN,'headers':{'Cookie':'signed=private'}}
            ident=Descriptors(temp).put(payload)
            self.assertEqual(Descriptors(temp).get(ident),payload)
            self.assertEqual(len(ident),32)
            with self.assertRaises(LookupError): Descriptors(temp).get('f'*32)

    def test_rate_limited_manifests_are_not_retried_on_every_listing(self):
        calls=[]
        def fetch(url,headers): calls.append(1); raise ProviderError('Media HTTP 429',429,120)
        quality=Quality(fetch=fetch)
        for _ in range(2):
            choices=quality.expand([SOURCE],'https://addon.example/private')
            self.assertEqual(choices[0]['name'],'MovieBox Auto')
        self.assertEqual(len(calls),1)

    def test_dynamic_and_entity_manifests_rejected(self):
        with self.assertRaises(ProviderError): parse_mpd(MPD.replace(b'type="static"',b'type="dynamic"'))
        with self.assertRaises(ProviderError): parse_mpd(b'<!DOCTYPE MPD><MPD/>')

    def test_authenticated_http_manifest_has_adaptive_tracks(self):
        token='quality-unit-'+'x'*32
        addon=Addon(playback_base='https://addon.example/private',manifest_fetch=lambda url,headers:(MPD,ORIGIN))
        choice=next(s for s in addon.quality.expand([SOURCE],'https://addon.example/private') if s['name']=='MovieBox Auto up to 1080p')
        route=choice['url'].replace('https://addon.example/private','')
        server=make_server(0,token,addon); thread=threading.Thread(target=server.serve_forever,daemon=True); thread.start()
        base='http://127.0.0.1:'+str(server.server_address[1])
        try:
            with urlopen(base+'/'+token+route) as response:
                self.assertEqual(response.headers.get_content_type(),'application/dash+xml')
                self.assertEqual(available_heights(parse_mpd(response.read())),[1080,720,480])
            with urlopen(Request(base+'/'+token+route,method='HEAD')) as response:
                self.assertEqual(response.read(),b'')
            with self.assertRaises(HTTPError) as caught: urlopen(base+'/wrong'+route)
            self.assertEqual(caught.exception.code,404)
        finally:
            server.shutdown(); server.server_close(); thread.join(2)


if __name__=='__main__': unittest.main(verbosity=2)