import asyncio
import time
import json

import pytest

from api.utils import official_gold as gold
from api.core.Gzip import gzip_zip
from api.utils.gold_time import refresh_slot


class FakeResponse:
    status = 200
    def __init__(self, region):
        self.region = region

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc, traceback):
        return False

    def raise_for_status(self):
        return None

    async def json(self, content_type=None):
        await asyncio.sleep(0)
        return {
            'data': {
                'godDealRank': [{'price': value} for value in ['100.5', '101', '102', '103', '104', '105']],
                'godTopDealRank': [{'price': value} for value in ['200.25', '201', '202', '203', '204', '205']],
            }
        }


class FakeSession:
    def __init__(self, calls, **kwargs):
        self.calls = calls
        self.kwargs = kwargs

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc, traceback):
        return False

    def post(self, url, data):
        self.calls.append((url, data))
        return FakeResponse(data['serverNo'])


def test_fetch_official_gold_uses_async_http_with_isolated_forms(monkeypatch):
    calls = []
    sessions = []

    def create_session(**kwargs):
        session = FakeSession(calls, **kwargs)
        sessions.append(session)
        return session

    monkeypatch.setattr(gold.aiohttp, 'ClientSession', create_session)

    result = asyncio.run(gold.fetch_gold())

    assert [data['serverNo'] for _, data in calls] == list(gold.REGIONS)
    assert all(url == gold.URL for url, _ in calls)
    assert len({id(data) for _, data in calls}) == len(gold.REGIONS)
    assert [item['region'] for item in result] == [f'跨{region}' for region in gold.REGIONS]
    assert [item['code'] for item in result] == ['k1', 'k2', 'k3a', 'k3b', 'k4', 'k5', 'k6', 'k7']
    assert all(isinstance(item['updataTime'], str) and item['updataTime'].isdigit() for item in result)
    assert int(result[0]['updataTime']) == refresh_slot()
    assert result[0]['goldDealRank'] == [100.5, 101, 102, 103, 104]
    assert result[0]['goldTopDealRank'] == [200.25, 201, 202, 203, 204]
    assert result[0]['max'] == round(10000 / 100.5, 2)
    assert result[0]['min'] == round(10000 / 204, 2)
    assert sessions[0].kwargs['timeout'].total == 10


def test_official_batch_timestamp_comes_from_start_not_fetch_completion(monkeypatch):
    started_at = 1793766300  # 与真实结束时间不同的固定开始时间。
    monkeypatch.setattr(gold.aiohttp, 'ClientSession', lambda **kwargs: FakeSession([], **kwargs))
    rows = asyncio.run(gold.fetch_gold(started_at=started_at))
    assert {row['updataTime'] for row in rows} == {str(refresh_slot(started_at))}


def test_preloaded_next_slot_cache_is_servable_before_the_half_hour(monkeypatch):
    from datetime import datetime, timedelta, timezone
    china = timezone(timedelta(hours=8))
    now = datetime(2026, 11, 4, 1, 25, tzinfo=china).timestamp()
    monkeypatch.setattr(gold.time, 'time', lambda: now)
    rows = cached_rows()
    redis = FakeRedis()
    redis.values[gold.CACHE_KEY] = gzip_zip(json.dumps(rows))
    assert gold.is_fresh(asyncio.run(gold.read_cache(redis)))
    target = int(rows[0]['updataTime'])
    assert target > now
    assert target - now == 300
    monkeypatch.setattr(gold.time, 'time', lambda: now + 1800)
    assert not gold.is_fresh(rows)


def test_official_route_uses_new_cache_contract_and_numeric_rank_arrays(monkeypatch, client):
    calls = []

    async def cached(redis):
        calls.append(redis)
        return [{'updataTime': '1793711716', 'region': '跨3B', 'code': 'k3b',
                 'goldDealRank': [100.5, 101], 'goldTopDealRank': [200.25, 201],
                 'max': round(10000 / 100.5, 2), 'min': round(10000 / 201, 2)}]

    monkeypatch.setattr(gold, 'get_data', cached)
    body = client.get('/api/open/gold/official/').json()
    assert set(body) == {'code', 'message', 'data'}
    assert body['code'] == 200
    assert len(calls) == 1
    row = body['data'][0]
    assert set(row) == {'updataTime', 'region', 'code', 'goldDealRank', 'goldTopDealRank', 'max', 'min'}
    assert isinstance(row['updataTime'], str) and row['updataTime'].isdigit()
    assert row['code'] == 'k3b'
    assert row['goldDealRank'] == [100.5, 101]
    assert row['goldTopDealRank'] == [200.25, 201]
    assert row['max'] == round(10000 / 100.5, 2)
    assert row['min'] == round(10000 / 201, 2)
    for rank in (row['goldDealRank'], row['goldTopDealRank']):
        assert len(rank) <= 5
        assert all(isinstance(value, (int, float)) for value in rank)


def test_gold_openapi_matches_timestamp_types_and_required_region_codes(client):
    schemas = client.app.openapi()['components']['schemas']
    realtime = schemas['RealTimeGold']
    official = schemas['OfficialGold']
    assert realtime['properties']['updataTime']['type'] == 'integer'
    assert official['properties']['updataTime']['type'] == 'string'
    assert 'code' in realtime['required'] and 'code' in official['required']
    assert realtime['properties']['goldPrice']['type'] == 'number'
    for name in ('goldDealRank', 'goldTopDealRank'):
        assert official['properties'][name]['maxItems'] == 5
        assert official['properties'][name]['items']['type'] == 'number'


class FakeRedis:
    def __init__(self):
        self.values = {}
        self.ttls = {}

    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.ttls[key] = ex
        return True

    async def delete(self, key):
        return int(self.values.pop(key, None) is not None)

    async def eval(self, script, number, key, owner):
        if self.values.get(key) == owner:
            return await self.delete(key)
        return 0


def cached_rows(age=0):
    return [{'updataTime': str(refresh_slot(time.time() - age)), 'region': f'跨{region}',
             'code': f'k{region.lower()}', 'goldDealRank': [100.5], 'goldTopDealRank': [200.25]}
            for region in gold.REGIONS]


def test_official_converts_point_prices_to_gold_per_100_points():
    row = {**cached_rows()[0], 'goldDealRank': [125, 150], 'goldTopDealRank': [200, 250]}
    result = gold.normalize_row(row)
    assert result['max'] == 80
    assert result['min'] == 40
    assert result['goldDealRank'] == [125, 150]
    assert result['goldTopDealRank'] == [200, 250]


def test_conversion_ignores_nonpositive_invalid_prices_and_uses_null_when_empty():
    row = {**cached_rows()[0], 'goldDealRank': [0, -1, float('nan'), float('inf')],
           'goldTopDealRank': []}
    result = gold.normalize_row(row)
    assert result['max'] is None and result['min'] is None
    row['goldTopDealRank'] = [125]
    result = gold.normalize_row(row)
    assert result['max'] == result['min'] == 80


def test_old_official_cache_gets_max_and_min_without_refetch():
    redis = FakeRedis()
    redis.values[gold.CACHE_KEY] = gzip_zip(json.dumps(cached_rows()))
    rows = asyncio.run(gold.read_cache(redis))
    assert len(rows) == 8
    assert rows[0]['max'] == round(10000 / 100.5, 2)
    assert rows[0]['min'] == round(10000 / 200.25, 2)


@pytest.mark.parametrize('payload', [
    {'returnCode': -30003, 'returnMsg': 'auth verification failed'},
    {'result': '-30003'}, {'message': 'token expired'}, {'msg': '请重新登录'},
])
def test_expired_credentials_are_classified(payload):
    assert gold.is_auth_failure(payload)


@pytest.mark.parametrize('payload', [
    {'returnCode': 500, 'message': 'server error'}, {'message': 'network timeout'}, {'data': {}},
])
def test_regular_upstream_failures_do_not_trigger_expiry_alert(payload):
    assert not gold.is_auth_failure(payload)


def test_fetch_detects_actual_auth_failure_response(monkeypatch):
    class Response(FakeResponse):
        async def json(self, **kwargs):
            return {'result': -30003, 'returnCode': -30003, 'returnMsg': 'auth verification failed'}

    class Session(FakeSession):
        def post(self, url, data):
            return Response(data['serverNo'])

    monkeypatch.setattr(gold.aiohttp, 'ClientSession', lambda **kwargs: Session([], **kwargs))
    with pytest.raises(gold.TokenExpired):
        asyncio.run(gold.fetch_gold())


def test_refresh_updates_cache_without_http_request_and_skips_when_fresh(monkeypatch):
    redis = FakeRedis()
    calls = []

    async def fetch(*, started_at=None):
        calls.append(1)
        return cached_rows()

    monkeypatch.setattr(gold, 'fetch_gold', fetch)
    assert asyncio.run(gold.refresh_cache(redis)) is True
    assert len(asyncio.run(gold.read_cache(redis))) == 8
    assert redis.ttls[gold.CACHE_KEY] == gold.RETENTION_SECONDS
    assert gold.LOCK_KEY not in redis.values
    assert asyncio.run(gold.refresh_cache(redis)) is False
    assert calls == [1]


def test_failed_refresh_preserves_cache_and_notifies_once_per_fetch(monkeypatch):
    redis = FakeRedis()
    old = gzip_zip(json.dumps(cached_rows(age=1900)))
    redis.values[gold.CACHE_KEY] = old
    alerts = []

    async def fetch(*, started_at=None):
        raise gold.TokenExpired()

    async def notify(redis):
        alerts.append(1)

    monkeypatch.setattr(gold, 'fetch_gold', fetch)
    monkeypatch.setattr(gold, 'notify_token_expired', notify)
    assert asyncio.run(gold.refresh_cache(redis)) is False
    assert redis.values[gold.CACHE_KEY] == old
    assert alerts == [1]
    assert gold.LOCK_KEY not in redis.values
    assert redis.ttls[gold.COOLDOWN_KEY] == gold.ERROR_COOLDOWN_SECONDS
    assert asyncio.run(gold.refresh_cache(redis)) is False
    assert alerts == [1]


def test_network_failure_preserves_cache_without_token_alert(monkeypatch):
    redis = FakeRedis()
    old = gzip_zip(json.dumps(cached_rows(age=1900)))
    redis.values[gold.CACHE_KEY] = old

    async def fetch(*, started_at=None):
        raise TimeoutError('upstream unavailable')

    async def notify(redis):
        raise AssertionError('network failures are not credential failures')

    monkeypatch.setattr(gold, 'fetch_gold', fetch)
    monkeypatch.setattr(gold, 'notify_token_expired', notify)
    assert asyncio.run(gold.refresh_cache(redis)) is False
    assert redis.values[gold.CACHE_KEY] == old


@pytest.mark.parametrize('url', [None, '', '  '])
def test_unconfigured_bark_never_creates_http_session(monkeypatch, url):
    monkeypatch.setattr(gold.config, 'BARK_URL', url)

    def session(**kwargs):
        raise AssertionError('BARK must be completely skipped when not configured')

    monkeypatch.setattr(gold.aiohttp, 'ClientSession', session)
    redis = FakeRedis()
    assert asyncio.run(gold.notify_token_expired(redis)) is False
    assert not redis.values


def test_bark_alert_is_deduplicated_and_does_not_include_credentials(monkeypatch):
    redis = FakeRedis()
    posts = []
    monkeypatch.setattr(gold.config, 'BARK_URL', 'https://bark.example/device-secret')
    monkeypatch.setattr(gold.config, 'HELPER_TOKEN', 'private-token')

    class Response(FakeResponse):
        async def json(self, **kwargs):
            return {'code': 200}

    class Session(FakeSession):
        def post(self, url, json, **kwargs):
            posts.append((url, json))
            return Response('push')

    monkeypatch.setattr(gold.aiohttp, 'ClientSession', lambda **kwargs: Session([], **kwargs))
    assert asyncio.run(gold.notify_token_expired(redis)) is True
    assert asyncio.run(gold.notify_token_expired(redis)) is False
    assert len(posts) == 1
    assert posts[0][0] == gold.config.BARK_URL
    assert 'private-token' not in json.dumps(posts[0][1])
    assert 'device-secret' not in json.dumps(posts[0][1])
    assert all('private-token' not in key for key in redis.values)
    assert next(iter(redis.ttls.values())) == gold.ALERT_RETENTION_SECONDS
    monkeypatch.setattr(gold.config, 'HELPER_TOKEN', 'new-token')
    assert asyncio.run(gold.notify_token_expired(redis)) is True
    assert len(posts) == 2


def test_failed_bark_push_can_retry_without_leaking_device_url(monkeypatch, caplog):
    monkeypatch.setattr(gold.config, 'BARK_URL', 'https://bark.example/device-secret')
    redis = FakeRedis()

    class Session(FakeSession):
        def post(self, url, **kwargs):
            raise RuntimeError(url)

    monkeypatch.setattr(gold.aiohttp, 'ClientSession', lambda **kwargs: Session([], **kwargs))
    assert asyncio.run(gold.notify_token_expired(redis)) is False
    assert not redis.values
    assert 'device-secret' not in caplog.text