import asyncio
import hashlib
import json
import time

import pytest
from fastapi import BackgroundTasks

from api.routes import colg
from api.types.colg import GOLD_REGION_CODES
from api.utils import thirdpart_gold as gold
from api.utils.gold_time import refresh_slot


def offer(price, url='listing', platform=17, money=10):
    return {'Price': price, 'Money': money, 'Number': 1, 'PfId': platform, 'BuyUrl': url}


def cached_item(region='跨1', age=0):
    fetched_at = time.time() - age
    return {'region': region, 'goldPrice': 100.0, 'sampleCount': 7,
            'code': GOLD_REGION_CODES[region], 'updataTime': refresh_slot(fetched_at),
            'fetchedAt': fetched_at}


class FakeRedis:
    def __init__(self):
        self.values = {}
        self.writes = []

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        return False

    async def mget(self, keys):
        return [self.values.get(key) for key in keys]

    async def get(self, key):
        return self.values.get(key)

    async def set(self, key, value, nx=False, ex=None):
        if nx and key in self.values:
            return False
        self.values[key] = value
        self.writes.append((key, value, ex))
        return True

    async def eval(self, script, number, key, token):
        if self.values.get(key) == token:
            del self.values[key]
            return 1
        return 0


def test_average_uses_gold_per_yuan_ratios_at_indices_three_through_nine():
    rows = [offer(i / 100) for i in range(1, 12)]
    assert gold.first_page_average(rows) == (15.65, 7)
    invalid = [offer(0), offer(-1), offer(float('nan')), offer(float('inf')),
               offer(.01, money=-1), offer(None), offer(.01, money=float('nan'))]
    # 页面允许Money=-1的库存占位报价；另行排除不显示的低额订单。
    invalid.pop(4)
    assert gold.first_page_average(invalid + [offer(.01, money=.5)] + rows) == (15.65, 7)
    assert gold.first_page_average([offer(.016789)] * 10) == (round(1 / .016789, 2), 7)
    assert gold.first_page_average(rows[:4]) == (25.0, 1)
    with pytest.raises(ValueError, match='有效报价'):
        gold.first_page_average(invalid)
    with pytest.raises(ValueError, match='有效报价'):
        gold.first_page_average(rows[:3])


def test_page_deduplicates_before_cutoff_and_preserves_7881_variants():
    rows = [offer(.01, 'duplicate', 40), offer(.01, 'duplicate', 40)]
    rows += [offer(i / 100, str(i), 17) for i in range(2, 12)]
    assert gold.first_page_average(rows) == (15.65, 7)
    rows = [offer(.01, platform=44)] * 3 + [offer(.04, platform=44)] * 7
    assert gold.first_page_average(rows) == (25.0, 7)


def test_each_quote_is_inverted_before_averaging_without_early_rounding():
    rows = [offer(.008)] * 3 + [offer(.0095)] * 4 + [offer(.01)] * 3
    assert gold.first_page_average(rows) == (103.01, 7)
    assert round(1 / ((.0095 * 4 + .01 * 3) / 7), 2) == 102.94


def test_request_signature_matches_observed_browser_request():
    channel = {'data': '{"PfId":17,"CoinNo":"yxb"}',
               'sign': 'e33b1f23fdbe4677e5e6f8bc7bdcbf78'}
    seed = {'time': '639292586508765510', 'secret': 'a30ad49b1b2532c403de6319ec4f1c1a'}
    payload = gold.make_payload(channel, seed, {'dnfCross': 1})
    assert payload['secret'] == 'b7ac9fd4e4f15735dcbe1411b627bb4e'
    assert payload['data'] == channel['data']
    assert payload['time'] == seed['time']
    assert payload['price1'] == payload['price2'] == 0


class FakeResponse:
    status = 200

    def __init__(self, text):
        self.body = text

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        return False

    async def text(self):
        return self.body

    def raise_for_status(self):
        pass


def test_region_fetches_its_own_seed_config_and_all_initial_channels():
    seed = {'time': 'fresh-time', 'secret': 'fresh-seed'}
    channel = {'data': json.dumps({'PfId': 17, 'CoinNo': 'yxb', 'CoinUnit': '万金币'}),
               'sign': 'config-sign'}
    page = ('window.bijiaSecret=' + json.dumps(seed) + ';'
            'window.bijiaUserInfo={"dnfCross":1};'
            '<script src="/cate/1838/CoinSale_yxb_254_0.js?ver=935"></script>')
    posts, gets = [], []

    class Session:
        def get(self, url, **kwargs):
            gets.append(url)
            return FakeResponse(page if '/bijiaqi/' in url else
                                'window.bijiaqiChanels=' + json.dumps([channel]) + ';')

        def post(self, url, json, **kwargs):
            posts.append((url, json))
            return FakeResponse('{"success":1,"data":' + json_module.dumps([
                {'Price': .01, 'Money': 10, 'Number': 1} for _ in range(10)
            ]) + '}')

    json_module = json
    started_at = 1793766590  # 固定开始时间，与响应读取完成的当前时间不同。
    result = asyncio.run(gold.fetch_region(Session(), 'kua8', started_at=started_at))
    assert result['region'] == '跨8'
    assert result['code'] == 'k8'
    assert isinstance(result['updataTime'], int)
    assert result['goldPrice'] == 100.0
    assert result['sampleCount'] == 7
    assert result['updataTime'] == refresh_slot(started_at)
    assert result['fetchedAt'] == started_at
    assert gets == [gold.BASE_URL + '/bijiaqi/dnf/youxibi/kua8',
                    gold.BASE_URL + '/cate/1838/CoinSale_yxb_254_0.js?ver=935']
    assert posts[0][1]['secret'] == hashlib.md5(b'fresh-seed_17_config-sign').hexdigest()
    assert posts[0][1]['time'] == 'fresh-time'
    assert 'page' not in posts[0][1]


def test_rate_limited_html_does_not_become_empty_quotes():
    with pytest.raises(gold.UpstreamLimited):
        asyncio.run(gold._read_response(FakeResponse('<title>LimitDeny</title>')))


def test_cache_omits_missing_regions_marks_old_values_and_retains_order():
    redis = FakeRedis()
    redis.values[f'{gold.CACHE_PREFIX}:kua1'] = json.dumps(cached_item(age=1900))
    redis.values[f'{gold.CACHE_PREFIX}:kua2'] = json.dumps(cached_item('跨2', age=90000))
    redis.values[f'{gold.CACHE_PREFIX}:kua8'] = json.dumps(cached_item('跨8'))
    rows = asyncio.run(gold.read_cache(redis))
    assert [row['region'] for row in rows] == ['跨1', '跨8']
    assert [row['stale'] for row in rows] == [True, False]


def test_realtime_prefetches_at_25_and_remains_fresh_at_30(monkeypatch):
    base = 1793764800 // 1800 * 1800
    clock = {'now': base + 25 * 60}
    monkeypatch.setattr(gold.time, 'time', lambda: clock['now'])
    redis = FakeRedis()
    old = cached_item(age=25 * 60)
    redis.values[f'{gold.CACHE_PREFIX}:kua1'] = json.dumps(old)
    assert asyncio.run(gold.read_cache(redis))[0]['stale'] is True
    redis.values[f'{gold.CACHE_PREFIX}:kua1'] = json.dumps(cached_item())
    prefetched = asyncio.run(gold.read_cache(redis))[0]
    assert prefetched['updataTime'] == base + 30 * 60
    assert prefetched['stale'] is False
    clock['now'] = base + 30 * 60
    assert asyncio.run(gold.read_cache(redis))[0]['stale'] is False


def test_invalid_cached_average_is_not_served():
    redis = FakeRedis()
    redis.values[f'{gold.CACHE_PREFIX}:kua1'] = json.dumps({**cached_item(), 'goldPrice': -1})
    redis.values[f'{gold.CACHE_PREFIX}:kua8'] = json.dumps({**cached_item('跨8'), 'sampleCount': 0})
    assert asyncio.run(gold.read_cache(redis)) == []


def test_old_yuan_per_gold_cache_cannot_be_returned_as_new_ratio():
    redis = FakeRedis()
    redis.values['openapi:colg:thirdpart_gold:v2:kua1'] = json.dumps({**cached_item(), 'goldPrice': .01})
    assert asyncio.run(gold.read_cache(redis)) == []


def test_legacy_cache_gets_epoch_and_region_code_without_refetching():
    redis = FakeRedis()
    legacy = cached_item('跨3A')
    legacy.pop('code')
    legacy['updataTime'] = '2026-11-03 21:13:17'
    redis.values[f'{gold.CACHE_PREFIX}:kua3a'] = json.dumps(legacy)
    rows = asyncio.run(gold.read_cache(redis))
    assert len(rows) == 1
    assert rows[0]['code'] == 'k3a'
    assert rows[0]['updataTime'] == refresh_slot(legacy['fetchedAt'])
    assert not redis.writes


def test_cold_concurrent_requests_schedule_only_one_refresh():
    redis = FakeRedis()
    tasks = BackgroundTasks()

    async def run():
        return await asyncio.gather(*(gold.schedule_refresh(redis, tasks, []) for _ in range(10)))

    assert sum(asyncio.run(run())) == 1
    assert len(tasks.tasks) == 1
    assert tasks.tasks[0].func is gold.launch_refresh
    assert next(ttl for key, _, ttl in redis.writes if key == gold.LOCK_KEY) == gold.LOCK_SECONDS


def test_fresh_cache_and_failure_cooldown_do_not_schedule_upstream():
    redis = FakeRedis()
    tasks = BackgroundTasks()
    rows = [{**cached_item(name), 'stale': False} for name in gold.REGIONS.values()]
    assert asyncio.run(gold.schedule_refresh(redis, tasks, rows)) is False
    redis.values[gold.COOLDOWN_KEY] = '1'
    assert asyncio.run(gold.schedule_refresh(redis, tasks, [])) is False
    assert not tasks.tasks
    assert gold.LOCK_KEY not in redis.values


@pytest.mark.parametrize('limited', [False, True])
def test_refresh_preserves_failed_region_and_releases_its_lock(monkeypatch, limited):
    redis = FakeRedis()
    redis.values[gold.LOCK_KEY] = 'owner'
    old = json.dumps(cached_item(age=1900))
    redis.values[f'{gold.CACHE_PREFIX}:kua1'] = old
    monkeypatch.setattr(gold, 'create_redis', lambda: redis)
    monkeypatch.setattr(gold, 'REGIONS', {'kua1': '跨1', 'kua8': '跨8'})
    monkeypatch.setattr(gold, 'REGION_INTERVAL', 0)
    calls = []

    async def fetch(session, region, *, started_at=None):
        calls.append(region)
        if region == 'kua1':
            raise gold.UpstreamLimited() if limited else ValueError('channel failed')
        return cached_item('跨8')

    monkeypatch.setattr(gold, 'fetch_region', fetch)
    asyncio.run(gold.refresh_cache('owner'))
    assert redis.values[f'{gold.CACHE_PREFIX}:kua1'] == old
    assert gold.LOCK_KEY not in redis.values
    assert redis.values[gold.COOLDOWN_KEY] == '1'
    if limited:
        assert calls == ['kua1']
    else:
        assert calls == ['kua1', 'kua8']
        assert f'{gold.CACHE_PREFIX}:kua8' in redis.values


def test_refresh_cannot_delete_another_owners_lock(monkeypatch):
    redis = FakeRedis()
    redis.values[gold.LOCK_KEY] = 'replacement'
    monkeypatch.setattr(gold, 'create_redis', lambda: redis)
    asyncio.run(gold.refresh_cache('expired-owner'))
    assert redis.values[gold.LOCK_KEY] == 'replacement'


def test_route_serializes_existing_contract_and_hides_internal_timestamp(monkeypatch, client):
    async def read(redis):
        return [{**cached_item(name), 'stale': False} for name in gold.REGIONS.values()]

    async def schedule(*args):
        return False

    monkeypatch.setattr(gold, 'read_cache', read)
    monkeypatch.setattr(gold, 'schedule_refresh', schedule)
    body = client.get('/api/open/gold/thirdpart/').json()
    assert body['code'] == 200
    assert len(body['data']) == 9
    for row in body['data']:
        assert set(row) == {'updataTime', 'region', 'code', 'goldPrice'}
    schema = client.app.openapi()
    assert schema['paths']['/api/open/gold/thirdpart/']['get']['operationId'] == 'thirdpartGold'
    assert 'ThirdPartGoldCache' not in schema['components']['schemas']


def test_realtime_uses_thirdpart_cache_and_returns_only_defined_fields(monkeypatch, client):
    async def read(redis):
        return [{**cached_item(name), 'stale': False} for name in gold.REGIONS.values()]

    schedules = []

    async def schedule(redis, tasks, data):
        schedules.append(data)
        return False

    def old_colg_request(*args, **kwargs):
        raise AssertionError('realtime must no longer request COLG')

    monkeypatch.setattr(gold, 'read_cache', read)
    monkeypatch.setattr(gold, 'schedule_refresh', schedule)
    monkeypatch.setattr(colg.requests, 'post', old_colg_request)
    body = client.get('/api/open/gold/realtime/').json()
    assert set(body) == {'code', 'message', 'data'}
    assert body['code'] == 200
    assert len(body['data']) == 9
    assert len(schedules) == 1
    for name, row in zip(gold.REGIONS.values(), body['data']):
        assert set(row) == {'updataTime', 'region', 'code', 'goldPrice'}
        assert row['region'] == name
        assert row['code'] == GOLD_REGION_CODES[name]
        assert isinstance(row['updataTime'], int)
        assert isinstance(row['goldPrice'], (int, float))


def test_realtime_initialization_returns_empty_array_instead_of_partial_regions(monkeypatch, client):
    async def read(redis):
        return [{**cached_item('跨2'), 'stale': False}]

    async def schedule(*args):
        return False

    monkeypatch.setattr(gold, 'read_cache', read)
    monkeypatch.setattr(gold, 'schedule_refresh', schedule)
    body = client.get('/api/open/gold/realtime/').json()
    assert body['code'] == 202
    assert body['data'] == []
    assert '1/9' in body['message']


def test_route_cold_start_and_redis_failure_do_not_block_or_scrape(monkeypatch):
    redis = FakeRedis()
    tasks = BackgroundTasks()
    result = asyncio.run(colg.get_thirdpart_gold(redis, tasks))
    assert result.code == 202 and result.data == []
    assert len(tasks.tasks) == 1

    async def unavailable(redis):
        raise ConnectionError('Redis unavailable')

    monkeypatch.setattr(gold, 'read_cache', unavailable)
    tasks = BackgroundTasks()
    result = asyncio.run(colg.get_thirdpart_gold(redis, tasks))
    assert result.code == 503 and result.data == []
    assert not tasks.tasks


def test_route_partial_cache_reports_initialization_progress(monkeypatch):
    async def read(redis):
        return [{**cached_item(), 'stale': False}]

    async def schedule(*args):
        return False

    monkeypatch.setattr(gold, 'read_cache', read)
    monkeypatch.setattr(gold, 'schedule_refresh', schedule)
    result = asyncio.run(colg.get_thirdpart_gold(None, BackgroundTasks()))
    assert result.code == 202
    assert '1/9' in result.message
    assert result.data == []


@pytest.mark.parametrize(('count', 'expected_code'), [(1, 503), (9, 200)])
def test_refresh_schedule_failure_serves_only_complete_cache(monkeypatch, count, expected_code):
    data = [{**cached_item(name), 'stale': False} for name in list(gold.REGIONS.values())[:count]]

    async def read(redis):
        return data

    async def schedule(*args):
        raise ConnectionError('Redis unavailable')

    monkeypatch.setattr(gold, 'read_cache', read)
    monkeypatch.setattr(gold, 'schedule_refresh', schedule)
    result = asyncio.run(colg.get_thirdpart_gold(None, BackgroundTasks()))
    assert result.code == expected_code
    assert result.data == (data if count == 9 else [])


def test_background_launcher_finishes_before_refresh_completes(monkeypatch):
    async def run():
        started, finish = asyncio.Event(), asyncio.Event()

        async def refresh(token):
            started.set()
            await finish.wait()

        monkeypatch.setattr(gold, 'refresh_cache', refresh)
        tasks = BackgroundTasks()
        tasks.add_task(gold.launch_refresh, 'owner')
        try:
            await asyncio.wait_for(tasks(), .5)
            await asyncio.wait_for(started.wait(), .5)
            assert len(gold._refresh_tasks) == 1
            assert not next(iter(gold._refresh_tasks)).done()
        finally:
            finish.set()
            await gold.close_refresh_tasks()
        assert not gold._refresh_tasks

    asyncio.run(run())


def test_shutdown_cancels_refresh_and_releases_redis_lock(monkeypatch):
    redis = FakeRedis()
    redis.values[gold.LOCK_KEY] = 'owner'
    monkeypatch.setattr(gold, 'create_redis', lambda: redis)

    async def run():
        started = asyncio.Event()

        async def fetch(session, region, *, started_at=None):
            started.set()
            await asyncio.Event().wait()

        monkeypatch.setattr(gold, 'fetch_region', fetch)
        try:
            await gold.launch_refresh('owner')
            await asyncio.wait_for(started.wait(), .5)
        finally:
            await gold.close_refresh_tasks()
        assert gold.LOCK_KEY not in redis.values
        assert not gold._refresh_tasks

    asyncio.run(run())


def test_retry_only_fetches_missing_regions_instead_of_repeating_fresh_pages(monkeypatch):
    redis = FakeRedis()
    redis.values[gold.LOCK_KEY] = 'owner'
    for region, name in gold.REGIONS.items():
        if region != 'kua8':
            redis.values[f'{gold.CACHE_PREFIX}:{region}'] = json.dumps(cached_item(name))
    monkeypatch.setattr(gold, 'create_redis', lambda: redis)
    calls = []

    async def fetch(session, region, *, started_at=None):
        calls.append(region)
        return cached_item(gold.REGIONS[region])

    monkeypatch.setattr(gold, 'fetch_region', fetch)
    asyncio.run(gold.refresh_cache('owner'))
    assert calls == ['kua8']
    assert len(asyncio.run(gold.read_cache(redis))) == 9


def test_realtime_regions_share_batch_start_even_when_fetch_crosses_slot_boundary(monkeypatch):
    clock = {'now': 1793766590.0}
    started_at = clock['now']
    redis = FakeRedis()
    redis.values[gold.LOCK_KEY] = 'owner'
    monkeypatch.setattr(gold, 'create_redis', lambda: redis)
    monkeypatch.setattr(gold, 'REGIONS', {'kua1': '跨1', 'kua8': '跨8'})
    monkeypatch.setattr(gold, 'REGION_INTERVAL', 0)
    monkeypatch.setattr(gold.time, 'time', lambda: clock['now'])
    starts = []

    async def fetch(session, region, *, started_at):
        starts.append(started_at)
        clock['now'] += 1800
        return {**cached_item(gold.REGIONS[region]), 'fetchedAt': started_at,
                'updataTime': refresh_slot(started_at)}

    monkeypatch.setattr(gold, 'fetch_region', fetch)
    asyncio.run(gold.refresh_cache('owner'))
    assert starts == [started_at, started_at]
    timestamps = [json.loads(redis.values[f'{gold.CACHE_PREFIX}:{region}'])['updataTime']
                  for region in gold.REGIONS]
    assert timestamps == [refresh_slot(started_at)] * 2


def test_unresponsive_redis_returns_prompt_error_instead_of_hanging(monkeypatch):
    async def run():
        connections = []

        async def blackhole(reader, writer):
            connections.append(writer)
            await reader.read()

        server = await asyncio.start_server(blackhole, '127.0.0.1', 0)
        port = server.sockets[0].getsockname()[1]
        monkeypatch.setattr(gold.config, 'REDIS_URL', f'redis://127.0.0.1:{port}/0')
        try:
            async with gold.create_redis() as redis:
                result = await asyncio.wait_for(colg.get_thirdpart_gold(redis, BackgroundTasks()), 2.5)
            assert result.code == 503
            assert result.data == []
        finally:
            server.close()
            for writer in connections:
                writer.close()
                await writer.wait_closed()
            await server.wait_closed()

    asyncio.run(run())