"""后台暂时故障的分类、退避与停止条件，只使用临时任务和假响应。"""
import errno
from contextlib import closing
from pathlib import Path
import shutil
import sqlite3
import ssl
import tempfile
import unittest

import httpx

from peach.task_retry import RETRY_DELAYS, retry_delay, transient_delay
from peach.sources.base import FailureReason, SourceFailure
from peach import migrations


class RetryPolicyTests(unittest.TestCase):
    def test_network_and_temporary_file_errors_back_off_then_stop(self):
        for error in [TimeoutError(), ConnectionResetError(), OSError(errno.EBUSY, '文件暂时被占用'),
                      SourceFailure(FailureReason.NETWORK, '请求未取得')]:
            with self.subTest(error=type(error).__name__):
                self.assertEqual([retry_delay(error, index) for index in range(4)], [*RETRY_DELAYS, None])

    def test_credentials_integrity_disk_and_program_errors_do_not_retry(self):
        for error in [PermissionError(), FileNotFoundError(), OSError(errno.ENOSPC, '磁盘已满'),
                      sqlite3.IntegrityError(), ValueError(), RuntimeError(),
                      SourceFailure(FailureReason.AUTH_REQUIRED, '需要登录'),
                      SourceFailure(FailureReason.PARSE_ERROR, '页面已改版')]:
            with self.subTest(error=type(error).__name__):
                self.assertIsNone(transient_delay(error))

    def test_http_retry_after_is_respected_and_authentication_is_not_replayed(self):
        request = httpx.Request('GET', 'https://example.invalid/source')
        for status, expected in [(429, 1200.0), (503, 1200.0), (401, None), (403, None), (404, None)]:
            response = httpx.Response(status, headers={'Retry-After': '1200'}, request=request)
            error = httpx.HTTPStatusError('来源未取得', request=request, response=response)
            self.assertEqual(retry_delay(error, 0), expected)
        error = httpx.ConnectError('证书未通过校验')
        error.__cause__ = ssl.SSLCertVerificationError('证书不可信')
        self.assertIsNone(transient_delay(error))

    def test_invalid_retry_after_keeps_a_finite_schedule(self):
        request = httpx.Request('GET', 'https://example.invalid/source')
        for header in ('invalid', '9' * 400):
            with self.subTest(header=header[:20]):
                response = httpx.Response(429, headers={'Retry-After': header}, request=request)
                error = httpx.HTTPStatusError('来源限流', request=request, response=response)
                self.assertEqual(retry_delay(error, 0), 900)

    def test_sqlite_lock_codes_are_transient_but_missing_schema_is_not(self):
        with sqlite3.connect(':memory:') as db:
            try:
                db.execute('SELECT * FROM missing_table')
            except sqlite3.OperationalError as error:
                self.assertIsNone(transient_delay(error))
        error = sqlite3.OperationalError('数据库被占用')
        error.sqlite_errorcode = sqlite3.SQLITE_BUSY
        self.assertEqual(transient_delay(error), 0)


class RetryUpgradeTests(unittest.TestCase):
    def test_existing_task_keeps_its_remote_identity_path_and_failure_during_upgrade(self):
        directory = Path(__file__).resolve().parents[1] / 'migrations'
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary).resolve()
            previous = root / 'previous'
            previous.mkdir()
            for migration in migrations.discover(directory):
                if migration.version < '0051':
                    shutil.copyfile(migration.path, previous / migration.path.name)
            database = root / 'ledger.db'
            migrations.upgrade(database, previous)
            with closing(sqlite3.connect(database)) as connection:
                connection.row_factory = sqlite3.Row
                connection.execute("INSERT INTO download_task(provider,source_uri,target,state,updated_at,remote_id,"
                    "ledger_path,postprocess_state,postprocess_detail) VALUES(?,?,?,?,?,?,?,?,?)",
                    ('115', 'test-magnet', '/Movies', 'ingested', '2026-01-01T00:00:00Z', 'remote-1',
                     r'C:\Media\film.mp4', 'failed', '需要检查挂载'))
                connection.commit()
                before = dict(connection.execute('SELECT * FROM download_task').fetchone())
            self.assertEqual([item.version for item in migrations.upgrade(database, directory)], ['0051', '0052', '0053', '0054'])
            with closing(sqlite3.connect(database)) as connection:
                connection.row_factory = sqlite3.Row
                after = dict(connection.execute('SELECT * FROM download_task').fetchone())
            self.assertEqual(after.pop('postprocess_retries'), 0)
            self.assertEqual(after, before)
            self.assertEqual(migrations.upgrade(database, directory), [])