import concurrent.futures
import json
import threading
import time
import unittest
from unittest.mock import patch
from urllib.error import HTTPError
from urllib.request import Request, urlopen
from catalog import Addon, Cache, episode_numbers, select_candidates, stream_ttl
from provider import ProviderError, adapt_streams
from server import make_server, RateLimiter


class FakeProvider:
    token = None
    def __init__(self):
        self.search_calls = 0
        self.stream_calls = []
    def search(self, title, kind):
        self.search_calls += 1
        return [{'subjectId': '123456', 'subjectType': 1 if kind == 'movie' else 2, 'title': title, 'releaseDate': '2000-01-01', 'season': 1 if kind == 'series' else 0}]
    def seasons(self, subject):
        return {'seasons': [{'se': 1, 'allEp': '1,3', 'maxEp': 3}]}
    def streams(self, subject, season=0, episode=0):
        self.stream_calls.append((subject,season,episode))
        return [{'name':'test','url':'https://example.invalid/movie.mp4','behaviorHints':{'proxyHeaders':{'request':{'Cookie':'abc=def'}}}}]


def metadata(url):
    parts = url.split('/')
    kind, ident = parts[-2], parts[-1][:-5]
    if '/catalog/' in url:
        return {'metas': [{'id':'tt1234567','type':'movie','name':'Example'}]}
    return {'meta': {'id': ident, 'type': kind, 'name': 'Example', 'year': '2000', 'videos': [{'season':1,'episode':1},{'season':1,'episode':2},{'season':1,'episode':3}]}}


class AddonTests(unittest.TestCase):
    def setUp(self):
        self.provider = FakeProvider()
        self.addon = Addon(self.provider, metadata)

    def test_remakes_require_year_type_and_title(self):
        items = [{'subjectId':'1','subjectType':1,'title':'Example','releaseDate':'1990-01-01'}, {'subjectId':'2','subjectType':1,'title':'Example','releaseDate':'2000-01-01'}, {'subjectId':'3','subjectType':2,'title':'Example','releaseDate':'2000-01-01'}, {'subjectId':'4','subjectType':1,'title':'Example 2','releaseDate':'2000-01-01'}]
        self.assertEqual(select_candidates(items,{'name':'Example','year':'2000'},'movie'),['2'])

    def test_ambiguous_identity_returns_no_match(self):
        items=[{'subjectId':str(i),'subjectType':1,'title':'Example','releaseDate':'2000-01-01'} for i in (1,2)]
        self.assertEqual(select_candidates(items,{'name':'Example','year':'2000'},'movie'),[])

    def test_complete_original_series_beats_dub(self):
        items=[]
        for subject,title in [('1','Example'),('2','Example [Hindi]')]:
            for season in (1,2): items.append({'subjectId':subject,'subjectType':2,'title':title+' S'+str(season),'releaseDate':str(1999+season)+'-01-01','season':season})
        self.assertEqual(select_candidates(items,{'name':'Example','year':'2000'},'series')[0],'1')

    def test_missing_provider_episode_not_requested(self):
        self.assertEqual(self.addon.dispatch('/stream/series/tt1234567:1:2.json'),{'streams':[]})
        self.assertEqual(self.provider.stream_calls,[])

    def test_out_of_range_episode_rejected_before_search(self):
        self.assertEqual(self.addon.dispatch('/stream/series/tt1234567:5:1.json'),{'streams':[]})
        self.assertEqual(self.provider.search_calls,0)

    def test_encoded_episode_and_cached_source(self):
        for _ in range(2):
            self.assertEqual(len(self.addon.dispatch('/stream/series/tt1234567%3A1%3A3.json')['streams']),1)
        self.assertEqual(self.provider.stream_calls,[('123456',1,3)])
        self.assertEqual(self.provider.search_calls,1)

    def test_aio_name_content_and_episode_labels(self):
        from catalog import label_source
        original={'url':'https://example.invalid/index.mpd','name':'old','title':'MovieBox | 1080,720 | h265','behaviorHints':{'proxyHeaders':{'request':{'Cookie':'example=secret'}}}}
        movie=label_source(original,{'name':'Interstellar','year':'2014'},'movie')
        self.assertEqual(movie['name'],'MovieBox')
        self.assertTrue(movie['title'].startswith('Interstellar (2014)\n'))
        self.assertEqual(movie['behaviorHints']['filename'],'Interstellar (2014)')
        episode=label_source(original,{'name':'Sherlock','videos':[{'season':1,'episode':1,'title':'A Study in Pink'}]},'series',1,1)
        self.assertTrue(episode['title'].startswith('Sherlock S01E01 - A Study in Pink'))
        named_episode=label_source(original,{'name':'Sherlock','videos':[{'season':1,'episode':1,'name':'A Study in Pink'}]},'series',1,1)
        self.assertTrue(named_episode['title'].startswith('Sherlock S01E01 - A Study in Pink'))
        self.assertEqual(episode['behaviorHints']['proxyHeaders']['request']['Cookie'],'example=secret')
        self.assertEqual(original['name'],'old')

    def test_catalog_pagination_and_encoded_search(self):
        seen=[]
        def fetch(url): seen.append(url); return {'metas':[]}
        addon=Addon(self.provider,fetch)
        addon.dispatch('/catalog/movie/top/search=Night%20of%20the%20Living%20Dead&skip=100.json')
        self.assertIn('search=Night%20of%20the%20Living%20Dead&skip=100.json',seen[0])
        with self.assertRaises(ValueError): addon.dispatch('/catalog/movie/top/skip=-1.json')
        with self.assertRaises(ValueError): addon.dispatch('/catalog/movie/top/skip=100001.json')
        with self.assertRaises(ValueError): addon.dispatch('/catalog/movie/top/url=https%3A%2F%2Flocalhost.json')

    def test_invalid_identifiers_cannot_fetch_arbitrary_urls(self):
        self.assertEqual(self.addon.dispatch('/stream/movie/http%3A%2F%2Flocalhost.json'),{'streams':[]})
        self.assertEqual(self.provider.search_calls,0)

    def test_signed_cookie_expiry_caps_ttl(self):
        stream={'behaviorHints':{'proxyHeaders':{'request':{'Cookie':'Edge-Cache-Cookie=prefix:sign=abc:t=1000000045'}}}}
        self.assertEqual(stream_ttl([stream],1000000000),15)
        self.assertEqual(stream_ttl([stream],1000000050),0)
        self.assertEqual(stream_ttl([stream],999999000),300)
        self.assertEqual(stream_ttl([{}],1000000000),20)
        self.assertEqual(stream_ttl([stream,{}],999999000),20)
        other={'behaviorHints':{'proxyHeaders':{'request':{'Cookie':'Edge-Cache-Cookie=prefix:sign=abc:t=1000000040000'}}}}
        self.assertEqual(stream_ttl([stream,other],1000000000),10)
        class SignedProvider(FakeProvider):
            def streams(self, subject, season=0, episode=0):
                result=super().streams(subject,season,episode)
                result[0]['behaviorHints']['proxyHeaders']['request']['Cookie']='Edge-Cache-Cookie=prefix:sign=abc:t='+str(int(time.time())+3600)
                return result
        provider=SignedProvider(); addon=Addon(provider,metadata); clock=time.monotonic()
        with patch('catalog.time.monotonic',return_value=clock):
            first=addon.dispatch('/stream/movie/tt1234567.json')
        with patch('catalog.time.monotonic',return_value=clock+25):
            second=addon.dispatch('/stream/movie/tt1234567.json')
        self.assertEqual(first,second)
        self.assertEqual(len(provider.stream_calls),1)

    def test_notice_and_header_injection_filtered(self):
        for source in [{'url':'https://example.org/notice.mp4'}, {'url':'https://example.org/video.mp4','signCookie':'a=b\r\nX-Foo: bad'}]:
            self.assertEqual(adapt_streams({'streams':[source]},'test'),[])

    def test_concurrent_cache_misses_coalesce(self):
        cache=Cache(2); calls=[]
        def load(): calls.append(1); time.sleep(.02); return {'ok':True}
        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
            values=list(pool.map(lambda _:cache.remember('same',10,load),range(16)))
        self.assertEqual(len(calls),1)
        self.assertTrue(all(v=={'ok':True} for v in values))
        self.assertEqual(cache.locks,{})
        cache.put('second',{},10); cache.put('third',{},10)
        self.assertFalse(cache.get('same')[0])

    def test_provider_429_enforces_cooldown(self):
        calls=[]
        def fail(): calls.append(1); raise ProviderError('Upstream HTTP 429',429,90)
        with self.assertRaises(ProviderError): self.addon.call_provider(fail)
        with self.assertRaises(ProviderError): self.addon.call_provider(fail)
        self.assertEqual(len(calls),1)

    def test_http_auth_head_cors_and_manifest(self):
        token='unit-test-token-'+'x'*32
        server=make_server(0,token,self.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 self.assertRaises(HTTPError) as caught: urlopen(base+'/wrong/manifest.json')
            self.assertEqual(caught.exception.code,404)
            with urlopen(base+'/'+token+'/manifest.json') as response:
                manifest=json.load(response)
                self.assertEqual(manifest['version'],'1.1.3')
                self.assertEqual(len(manifest['catalogs']),6)
                self.assertEqual(response.headers['Cache-Control'],'no-store')
            with urlopen(Request(base+'/'+token+'/manifest.json',method='HEAD')) as response: self.assertEqual(response.read(),b'')
            with urlopen(Request(base+'/'+token+'/manifest.json',method='OPTIONS')) as response: self.assertEqual(response.status,204)
        finally:
            server.shutdown(); server.server_close(); thread.join(2)

    def test_client_request_limit(self):
        rate=RateLimiter(2)
        self.assertTrue(rate.allowed('a')); self.assertTrue(rate.allowed('a')); self.assertFalse(rate.allowed('a')); self.assertTrue(rate.allowed('b'))


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