import unittest
from datetime import datetime, timezone

import fakeredis.aioredis
import httpx
from pwo import async_test

from kaya.core import KayaApp, HttpContext
from kaya.session import Session, SessionMixin
from kaya.session.redis import RedisSessionStore


class RedisSessionStoreTest(unittest.TestCase):
    client: fakeredis.aioredis.FakeRedis
    store: RedisSessionStore

    def setUp(self) -> None:
        self.client = fakeredis.aioredis.FakeRedis()
        self.store = RedisSessionStore(self.client)

    @async_test
    async def test_save_and_load_round_trip(self) -> None:
        session = Session('abc', {'foo': 'bar', 'n': 42})
        await self.store.save('abc', session)
        loaded = await self.store.load('abc')
        self.assertIsNotNone(loaded)
        assert loaded is not None
        self.assertEqual('abc', loaded.id)
        self.assertEqual({'foo': 'bar', 'n': 42}, dict(loaded))

    @async_test
    async def test_load_unknown_session_returns_none(self) -> None:
        self.assertIsNone(await self.store.load('missing'))

    @async_test
    async def test_delete_removes_session(self) -> None:
        await self.store.save('abc', Session('abc', {'foo': 'bar'}))
        await self.store.delete('abc')
        self.assertIsNone(await self.store.load('abc'))

    @async_test
    async def test_save_sets_ttl_when_max_age_given(self) -> None:
        await self.store.save('abc', Session('abc', {'foo': 'bar'}), max_age=60)
        ttl = await self.client.ttl('kaya:session:abc')
        self.assertGreater(ttl, 0)
        self.assertLessEqual(ttl, 60)

    @async_test
    async def test_save_without_max_age_has_no_ttl(self) -> None:
        await self.store.save('abc', Session('abc', {'foo': 'bar'}))
        self.assertEqual(-1, await self.client.ttl('kaya:session:abc'))

    @async_test
    async def test_load_slides_expiry_when_max_age_given(self) -> None:
        await self.store.save('abc', Session('abc', {'foo': 'bar'}), max_age=60)
        await self.client.expire('kaya:session:abc', 10)
        loaded = await self.store.load('abc', max_age=60)
        self.assertIsNotNone(loaded)
        ttl = await self.client.ttl('kaya:session:abc')
        self.assertGreater(ttl, 10)
        self.assertLessEqual(ttl, 60)

    @async_test
    async def test_pickle_round_trip_of_non_json_values(self) -> None:
        now = datetime(2026, 7, 20, 12, 0, 0, tzinfo=timezone.utc)
        session = Session('abc', {'when': now, 'blob': b'\x00\x01', 'items': {1, 2, 3}})
        await self.store.save('abc', session)
        loaded = await self.store.load('abc')
        self.assertIsNotNone(loaded)
        assert loaded is not None
        self.assertEqual(now, loaded['when'])
        self.assertEqual(b'\x00\x01', loaded['blob'])
        self.assertEqual({1, 2, 3}, loaded['items'])

    @async_test
    async def test_custom_prefix(self) -> None:
        store = RedisSessionStore(self.client, prefix='myapp:sess:')
        await store.save('abc', Session('abc', {'foo': 'bar'}))
        self.assertIsNotNone(await self.client.get('myapp:sess:abc'))
        self.assertIsNone(await self.client.get('kaya:session:abc'))

    @async_test
    async def test_custom_serializer(self) -> None:
        import json

        store = RedisSessionStore(
            self.client,
            dumps=lambda d: json.dumps(d).encode('utf-8'),
            loads=lambda b: json.loads(b.decode('utf-8')),
        )
        await store.save('abc', Session('abc', {'foo': 'bar'}))
        self.assertEqual(b'{"foo": "bar"}', await self.client.get('kaya:session:abc'))
        loaded = await store.load('abc')
        self.assertIsNotNone(loaded)
        assert loaded is not None
        self.assertEqual({'foo': 'bar'}, dict(loaded))


class RedisSessionIntegrationTest(unittest.TestCase):
    app: KayaApp

    def setUp(self) -> None:
        store = RedisSessionStore(fakeredis.aioredis.FakeRedis())
        self.app = KayaApp(mixins=[SessionMixin(store)])

        @self.app.GET('/')
        async def home(ctx: HttpContext) -> None:
            n = ctx.session.get('visits', 0) + 1
            ctx.session['visits'] = n
            await ctx.send_str(200, f'visits: {n}')

    @async_test
    async def test_session_persists_across_requests(self) -> None:
        transport = httpx.ASGITransport(app=self.app)
        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
            r = await client.get('/')
            self.assertEqual(200, r.status_code)
            self.assertEqual('visits: 1', r.text)
            self.assertIn('Set-Cookie', r.headers)

            r = await client.get('/')
            self.assertEqual(200, r.status_code)
            self.assertEqual('visits: 2', r.text)