import base64
import concurrent.futures
import io
import re
import tempfile
import unittest
from pathlib import Path
from urllib.parse import urlencode
import app


class ApplicationTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.previous = app.DB
        app.DB = Path(self.temp.name) / 'db.sqlite3'
        self.previous_uploads = app.UPLOADS
        app.UPLOADS = Path(self.temp.name) / 'images'
        app.SESSIONS.clear()
        app.ATTEMPTS.clear()
        app.initialize()
        with app.connect() as db:
            db.execute("INSERT INTO products (id,name,description,unit,stock) VALUES (1,'Hirschragout','500 g vakuumiert','Packungen',3)")

    def tearDown(self):
        app.DB = self.previous
        app.UPLOADS = self.previous_uploads
        self.temp.cleanup()

    def stock(self):
        with app.connect() as db:
            return db.execute('SELECT stock FROM products WHERE id=1').fetchone()[0]

    def test_reservation_transaction_rolls_back_all_items(self):
        with self.assertRaises(ValueError):
            app.reserve('Anna', 'anna@example.test', {1: 2, 99: 1})
        self.assertEqual(self.stock(), 3)
        with app.connect() as db:
            self.assertEqual(db.execute('SELECT count(*) FROM reservations').fetchone()[0], 0)

    def test_concurrent_reservations_never_oversell(self):
        def attempt(_):
            try:
                app.reserve('Anna', 'anna@example.test', {1: 2})
                return True
            except ValueError:
                return False
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
            self.assertEqual(sum(pool.map(attempt, range(2))), 1)
        self.assertEqual(self.stock(), 1)

    def test_cancellation_returns_stock_only_once(self):
        app.reserve('Anna', 'anna@example.test', {1: 2})
        app.change_status(1, 'storniert')
        app.change_status(1, 'storniert')
        self.assertEqual(self.stock(), 3)
        with self.assertRaises(ValueError):
            app.change_status(1, 'offen')

    def test_collected_stock_stays_deducted(self):
        app.reserve('Anna', 'anna@example.test', {1: 1})
        app.change_status(1, 'bestätigt')
        app.change_status(1, 'abgeholt')
        with self.assertRaises(ValueError):
            app.change_status(1, 'storniert')
        self.assertEqual(self.stock(), 2)

    def test_password_is_salted_and_minimum_length_enforced(self):
        with self.assertRaises(ValueError):
            app.set_password('short')
        app.set_password('Test-only-password!')
        with app.connect() as db:
            row = db.execute('SELECT * FROM admin').fetchone()
        self.assertNotEqual(row['hash'], 'Test-only-password!')
        self.assertEqual(row['hash'], app.password_hash('Test-only-password!', row['salt']))

    def test_http_auth_csrf_rotation_reservation_and_escaping(self):
        app.set_password('Test-only-password!')
        def request(method, path, fields=None, cookie=''):
            body = urlencode(fields) if fields else None
            payload = (body or '').encode()
            raw = (f'{method} {path} HTTP/1.0\r\nHost: localhost\r\nCookie: {cookie}\r\nContent-Length: {len(payload)}\r\nContent-Type: application/x-www-form-urlencoded\r\n\r\n').encode() + payload
            class Connection:
                def __init__(self):
                    self.output = io.BytesIO()
                def makefile(self, *args):
                    return io.BytesIO(raw)
                def sendall(self, data):
                    self.output.write(data)
            connection = Connection()
            app.Handler(connection, ('127.0.0.1', 1234), object())
            headers, page = connection.output.getvalue().decode().split('\r\n\r\n', 1)
            status = int(headers.split()[1])
            cookie_header = next((line.split(': ',1)[1] for line in headers.split('\r\n') if line.startswith('Set-Cookie:')), '')
            return status, page, cookie_header
        try:
            status, page, cookie = request('GET', '/?tier=Rotwild%20Hirsch&cut=Gulasch')
            self.assertEqual(status, 200)
            csrf = re.search('name="csrf" value="([^"]+)"', page)[1]
            self.assertIn('SameSite=Strict', cookie)
            self.assertEqual(request('POST','/product', {'csrf': csrf}, cookie)[0],403)
            self.assertEqual(request('POST','/login', {'csrf':'wrong','password':'Test-only-password!'},cookie)[0],403)
            status, _, admin_cookie = request('POST','/login',{'csrf':csrf,'password':'Test-only-password!'},cookie)
            self.assertEqual(status,303)
            self.assertNotEqual(cookie, admin_cookie)
            status, page, _ = request('GET','/admin',cookie=admin_cookie)
            self.assertIn('Bestand pflegen',page)
            self.assertNotIn('Bestand pflegen',request('GET','/admin',cookie=cookie)[1])
            status, page, public_cookie = request('GET','/?tier=Rotwild%20Hirsch&cut=Gulasch')
            public_csrf = re.search('name="csrf" value="([^"]+)"',page)[1]
            self.assertEqual(request('POST','/reserve',{'csrf':public_csrf,'name':'<script>alert(1)</script>','contact':'a@example.test','q_1':'1'},public_cookie)[0],200)
            admin_page = request('GET','/admin',cookie=admin_cookie)[1]
            self.assertIn('&lt;script&gt;',admin_page)
            self.assertNotIn('<script>',admin_page)
            self.assertEqual(self.stock(),2)
        finally:
            app.SESSIONS.clear()

    def test_invalid_quantities_do_not_change_stock(self):
        for quantities in ({1:-1},{1:4},{1:0},{}):
            with self.assertRaises(ValueError):
                app.reserve('Anna','anna@example.test',quantities)
        self.assertEqual(self.stock(),3)

    def test_upload_example_and_reject_foreign_files(self):
        for name in app.EXAMPLES:
            self.assertEqual(app.image_kind((app.EXAMPLES_DIR / name).read_bytes()), 'jpg')
        self.assertIsNone(app.image_path('beispiel:../app.py'))
        self.assertIsNone(app.image_path('../../app.py'))
        with self.assertRaises(ValueError):
            app.store_image(None, 'keep', b'<html>not an image</html>')
        png = base64.b64decode('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==')
        key, old = app.store_image(None, 'keep', png)
        self.assertTrue(key.endswith('.png'))
        self.assertEqual((app.UPLOADS / key).read_bytes(), png)
        example, _ = app.store_image(key, 'beispiel:ruecken.jpg', None)
        self.assertEqual(example, 'beispiel:ruecken.jpg')
        app.drop_upload(key)
        self.assertFalse((app.UPLOADS / key).exists())
        app.set_password('Test-only-password!')
        def request(method, path, body=b'', cookie='', content_type='application/x-www-form-urlencoded'):
            raw = (f'{method} {path} HTTP/1.0\r\nHost: localhost\r\nCookie: {cookie}\r\nContent-Length: {len(body)}\r\nContent-Type: {content_type}\r\n\r\n').encode() + body
            class Connection:
                def __init__(self):
                    self.output = io.BytesIO()
                def makefile(self, *args):
                    return io.BytesIO(raw)
                def sendall(self, data):
                    self.output.write(data)
            connection = Connection()
            app.Handler(connection, ('127.0.0.1', 1234), object())
            headers, page = connection.output.getvalue().split(b'\r\n\r\n', 1)
            text = headers.decode()
            status = int(text.split()[1])
            cookie_header = next((line.split(': ',1)[1] for line in text.split('\r\n') if line.startswith('Set-Cookie:')), '')
            return status, page, cookie_header
        try:
            _, page, cookie = request('GET', '/?tier=Rotwild%20Hirsch&cut=Gulasch')
            csrf = re.search(br'name="csrf" value="([^"]+)"', page)[1].decode()
            _, _, admin_cookie = request('POST', '/login', urlencode({'csrf': csrf, 'password': 'Test-only-password!'}).encode(), cookie)
            _, admin_page, _ = request('GET', '/admin', cookie=admin_cookie)
            admin_csrf = re.search(br'name="csrf" value="([^"]+)"', admin_page)[1].decode()
            boundary = '----wildtest'
            body = (f'--{boundary}\r\nContent-Disposition: form-data; name="csrf"\r\n\r\n{admin_csrf}\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="id"\r\n\r\n1\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="name"\r\n\r\nHirschragout\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="description"\r\n\r\n500 g vakuumiert\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="unit"\r\n\r\nPackungen\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="stock"\r\n\r\n3\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="price"\r\n\r\n18,50\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="animal"\r\n\r\nRotwild Hirsch\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="cut"\r\n\r\nGulasch\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="image_choice"\r\n\r\nkeep\r\n'
                    f'--{boundary}\r\nContent-Disposition: form-data; name="image"; filename="ware.png"\r\nContent-Type: image/png\r\n\r\n').encode() + png + f'\r\n--{boundary}--\r\n'.encode()
            status, _, _ = request('POST', '/product', body, admin_cookie, f'multipart/form-data; boundary={boundary}')
            self.assertEqual(status, 303)
            status, image, _ = request('GET', '/bild/1')
            self.assertEqual(status, 200)
            self.assertEqual(image, png)
            status, page, _ = request('GET', '/?tier=Rotwild%20Hirsch&cut=Gulasch')
            self.assertIn(b'/bild/1', page)
            self.assertIn('18,50 € / kg'.encode(), page)
            with self.assertRaises(ValueError):
                app.parse_price('gratis')
            status, _, _ = request('GET', '/beispiel/../app.py')
            self.assertEqual(status, 404)
            status, example_image, _ = request('GET', '/beispiel/schulter.jpg')
            self.assertEqual(status, 200)
            self.assertTrue(example_image.startswith(b'\xff\xd8\xff'))
        finally:
            app.SESSIONS.clear()

    def test_delete_product_and_reservation(self):
        app.reserve('Anna', 'anna@example.test', {1: 2})
        with self.assertRaises(ValueError):
            app.delete_product(1)
        self.assertEqual(self.stock(), 1)
        app.delete_reservation(1)
        self.assertEqual(self.stock(), 3)
        with app.connect() as db:
            self.assertEqual(db.execute('SELECT count(*) FROM reservations').fetchone()[0], 0)
        app.reserve('Anna', 'anna@example.test', {1: 1})
        app.change_status(1, 'bestätigt')
        app.delete_reservation(1)
        self.assertEqual(self.stock(), 3)
        app.reserve('Anna', 'anna@example.test', {1: 1})
        app.change_status(1, 'abgeholt')
        app.delete_reservation(1)
        self.assertEqual(self.stock(), 2)
        image, _ = app.store_image(None, 'keep', base64.b64decode('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=='))
        with app.connect() as db:
            db.execute('UPDATE products SET image=? WHERE id=1', (image,))
        app.drop_upload(app.delete_product(1))
        self.assertFalse((app.UPLOADS / image).exists())
        with app.connect() as db:
            self.assertEqual(db.execute('SELECT count(*) FROM products').fetchone()[0], 0)


    def test_create_several_packs_and_compact_admin(self):
        app.set_password('Test-only-password!')

        def request(method, path, body=b'', cookie='', content_type='application/x-www-form-urlencoded'):
            raw = (f'{method} {path} HTTP/1.0\r\nHost: localhost\r\nCookie: {cookie}\r\nContent-Length: {len(body)}\r\nContent-Type: {content_type}\r\n\r\n').encode() + body
            class Connection:
                def __init__(self):
                    self.output = io.BytesIO()
                def makefile(self, *args):
                    return io.BytesIO(raw)
                def sendall(self, data):
                    self.output.write(data)
            connection = Connection()
            app.Handler(connection, ('127.0.0.1', 1234), object())
            headers, page = connection.output.getvalue().split(b'\r\n\r\n', 1)
            text = headers.decode()
            status = int(text.split()[1])
            cookie_header = next((line.split(': ', 1)[1] for line in text.split('\r\n') if line.startswith('Set-Cookie:')), '')
            return status, page, cookie_header

        try:
            _, page, cookie = request('GET', '/?tier=Rotwild%20Hirsch&cut=Gulasch')
            csrf = re.search(br'name="csrf" value="([^"]+)"', page)[1].decode()
            _, _, admin_cookie = request('POST', '/login', urlencode({'csrf': csrf, 'password': 'Test-only-password!'}).encode(), cookie)
            status, admin_page, _ = request('GET', '/admin', cookie=admin_cookie)
            self.assertEqual(status, 200)
            text = admin_page.decode()
            self.assertIn('Packungen anlegen', text)
            self.assertEqual(text.count('name="weights"'), 1)
            self.assertEqual(text.count('Text und Bild'), 1)
            admin_csrf = re.search(br'name="csrf" value="([^"]+)"', admin_page)[1].decode()
            status, _, _ = request('POST', '/product', urlencode({
                'csrf': admin_csrf,
                'animal': 'Rotwild Hirsch',
                'cut': 'Hack',
                'name': '',
                'description': 'Hirsch, 3 Jahre alt',
                'price': '20,00',
                'weights': '740\n760\n785',
                'image_choice': 'auto',
            }).encode(), admin_cookie)
            self.assertEqual(status, 303)
            with app.connect() as db:
                rows = db.execute("SELECT name, unit, stock, price_cents, image, animal, cut FROM products WHERE cut='Hack' ORDER BY unit").fetchall()
            self.assertEqual([row['unit'] for row in rows], ['Packung à 740 g', 'Packung à 760 g', 'Packung à 785 g'])
            self.assertTrue(all(row['stock'] == 1 and row['name'] == 'Rotwild Hirsch Hack' and row['image'] == 'beispiel:hack.jpg' and row['price_cents'] == 2000 for row in rows))
            status, catalog, _ = request('GET', '/?tier=Rotwild%20Hirsch&cut=Hack')
            self.assertEqual(status, 200)
            self.assertEqual(catalog.decode().count('class="card pack"'), 3)
        finally:
            app.SESSIONS.clear()


    def test_taken_pack_offers_alternative_without_double_reservation(self):
        with app.connect() as db:
            db.execute("UPDATE products SET animal='Rotwild Hirsch', cut='Hack', unit='Packung à 740 g', stock=0 WHERE id=1")
            db.execute("INSERT INTO products (id,name,description,unit,stock,animal,cut,price_cents) VALUES (2,'Rotwild Hirsch Hack','Hirsch','Packung à 760 g',1,'Rotwild Hirsch','Hack',2000)")
        with self.assertRaises(app.StockConflict) as caught:
            app.reserve('Anna', 'anna@example.test', {1: 1})
        self.assertEqual([item['id'] for item in caught.exception.conflicts[0]['alternatives']], [2])
        with app.connect() as db:
            self.assertEqual(db.execute('SELECT count(*) FROM reservations').fetchone()[0], 0)
            self.assertEqual(db.execute('SELECT stock FROM products WHERE id=2').fetchone()[0], 1)
        app.reserve('Anna', 'anna@example.test', {2: 1})
        with self.assertRaises(app.StockConflict) as caught:
            app.reserve('Bea', 'bea@example.test', {2: 1})
        self.assertEqual(caught.exception.conflicts[0]['alternatives'], [])
        with app.connect() as db:
            self.assertEqual(db.execute('SELECT count(*) FROM reservations').fetchone()[0], 1)
            self.assertEqual(db.execute('SELECT stock FROM products WHERE id=2').fetchone()[0], 0)

        app.set_password('Test-only-password!')

        def request(method, path, body=b'', cookie=''):
            raw = (f'{method} {path} HTTP/1.0\r\nHost: localhost\r\nCookie: {cookie}\r\nContent-Length: {len(body)}\r\nContent-Type: application/x-www-form-urlencoded\r\n\r\n').encode() + body
            class Connection:
                def __init__(self):
                    self.output = io.BytesIO()
                def makefile(self, *args):
                    return io.BytesIO(raw)
                def sendall(self, data):
                    self.output.write(data)
            connection = Connection()
            app.Handler(connection, ('127.0.0.1', 1234), object())
            headers, page = connection.output.getvalue().split(b'\r\n\r\n', 1)
            text = headers.decode()
            cookie_header = next((line.split(': ', 1)[1] for line in text.split('\r\n') if line.startswith('Set-Cookie:')), '')
            return int(text.split()[1]), page, cookie_header

        _, page, cookie = request('GET', '/admin')
        csrf = re.search(br'name="csrf" value="([^"]+)"', page)[1].decode()
        status, offer, _ = request('POST', '/reserve', urlencode({'csrf': csrf, 'name': 'Bea', 'contact': 'bea@example.test', 'q_2': '1'}).encode(), cookie.split(';')[0])
        self.assertEqual(status, 200)
        text = offer.decode()
        self.assertIn('nicht reserviert', text)
        self.assertIn('Zurück und etwas anderes wählen', text)
        self.assertNotIn('Anfrage eingegangen', text)


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