#!/usr/bin/env python3
"""Isolated HTTP integration checks for the cPanel package.

Run: python3 tests/integration.py
Requires PHP >= 8.2 with PDO SQLite/fileinfo and Python 3 stdlib.
The runner copies the app to a temporary directory and creates a new database.
It never reads or writes the package's live config.php, storage, or customers.
"""
from __future__ import annotations
import contextlib
import http.cookiejar
import json
import os
from pathlib import Path
import re
import shutil
import socket
import sqlite3
import subprocess
import tempfile
import time
import unittest
import urllib.error
import urllib.parse
import urllib.request

PACKAGE = Path(__file__).resolve().parents[1]
PASSWORD = 'Isolated-Integration-2026!'
EMAIL = 'integration@example.invalid'
SETUP_KEY = 'isolated-integration-setup-key-not-production'


class Client:
    def __init__(self, base):
        self.base = base
        self.jar = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar))
        self.csrf = None

    def raw(self, path, body=None, headers=None, method=None):
        req = urllib.request.Request(self.base + path, data=body, headers=headers or {}, method=method)
        try:
            response = self.opener.open(req, timeout=10)
        except urllib.error.HTTPError as error:
            response = error
        with response:
            return response.status, response.read(), response.headers

    def api(self, action, data=None, csrf=True, extra_headers=None):
        headers = dict(extra_headers or {})
        body = None
        if data is not None:
            body = json.dumps(data, ensure_ascii=False).encode()
            headers['Content-Type'] = 'application/json'
            if csrf and self.csrf:
                headers['X-CSRF-Token'] = self.csrf
        status, content, response_headers = self.raw('/api.php?action=' + action, body, headers)
        try:
            result = json.loads(content)
        except json.JSONDecodeError as exc:
            raise AssertionError(f'{action}: non-JSON HTTP {status}: {content[:500]!r}') from exc
        return status, result, response_headers

    def login(self):
        status, result, _ = self.api('login', {'email': EMAIL, 'password': PASSWORD}, csrf=False)
        if status != 200 or not result.get('ok'):
            raise AssertionError(f'Fixture login failed: {status} {result}')
        self.csrf = result['data']['csrf']


class Integration(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.tmp = tempfile.TemporaryDirectory(prefix='doramagacin-integration-')
        cls.root = Path(cls.tmp.name)
        for folder in ('app', 'public'):
            shutil.copytree(PACKAGE / folder, cls.root / folder)
        (cls.root / 'storage').mkdir()
        (cls.root / 'sessions').mkdir()
        cls.dbpath = cls.root / 'storage' / 'test.sqlite'
        with socket.socket() as sock:
            sock.bind(('127.0.0.1', 0))
            port = sock.getsockname()[1]
        cls.base = f'http://127.0.0.1:{port}'
        config = {'base_url': cls.base, 'db_path': str(cls.dbpath), 'setup_key': SETUP_KEY, 'mail_enabled': False}
        # PHP strings are generated from a restricted fixture dictionary only.
        values = ',\n'.join("'%s' => %s" % (k, 'false' if v is False else "'%s'" % str(v).replace("'", "\\'")) for k, v in config.items())
        (cls.root / 'config.php').write_text('<?php return [\n' + values + '\n];\n')
        cls.log = open(cls.root / 'server.log', 'w+')
        cls.server = subprocess.Popen([shutil.which('php') or 'php', '-d', 'session.save_path=' + str(cls.root / 'sessions'), '-S', f'127.0.0.1:{port}', '-t', str(cls.root / 'public')], stdout=cls.log, stderr=cls.log)
        cls.setup_client = Client(cls.base)
        for _ in range(80):
            if cls.server.poll() is not None:
                raise RuntimeError('PHP server exited before setup')
            try:
                status, page, _ = cls.setup_client.raw('/setup.php')
                break
            except OSError:
                time.sleep(.05)
        else:
            raise RuntimeError('PHP server did not start')
        match = re.search(rb'name="csrf" value="([a-f0-9]+)"', page)
        if not match:
            raise AssertionError('Installer unavailable: ' + page.decode(errors='replace')[:2000])
        data = urllib.parse.urlencode({'csrf': match[1].decode(), 'setup_key': SETUP_KEY, 'name': 'Integration Admin', 'email': EMAIL, 'password': PASSWORD, 'confirm_password': PASSWORD, 'seed_bardovci': '1'}).encode()
        status, page, _ = cls.setup_client.raw('/setup.php', data, {'Content-Type': 'application/x-www-form-urlencoded'})
        if status != 200 or b'Installation completed' not in page:
            raise AssertionError('Installer failed: ' + page.decode(errors='replace')[:2000])
        cls.today = time.strftime('%Y-%m-%d')
        cls.serial = 0

    @classmethod
    def tearDownClass(cls):
        cls.server.terminate()
        with contextlib.suppress(subprocess.TimeoutExpired):
            cls.server.wait(timeout=3)
        if cls.server.poll() is None:
            cls.server.kill()
        cls.log.close()
        cls.tmp.cleanup()

    def setUp(self):
        self.client = Client(self.base)
        self.client.login()
        self.anonymous = Client(self.base)

    def ok(self, action, payload=None, client=None):
        status, result, _ = (client or self.client).api(action, payload)
        self.assertEqual(status, 200, f'{action}: {result}')
        self.assertTrue(result.get('ok'), f'{action}: {result}')
        return result['data']

    def rejected(self, action, payload=None, expected=422, client=None, **kwargs):
        status, result, _ = (client or self.client).api(action, payload, **kwargs)
        self.assertEqual(status, expected, f'{action}: {result}')
        self.assertFalse(result.get('ok'), f'{action}: {result}')
        return result

    def sql(self, statement, values=()):
        with sqlite3.connect(self.dbpath) as conn:
            conn.row_factory = sqlite3.Row
            result = conn.execute(statement, values)
            return [dict(row) for row in result.fetchall()]

    def customer(self, name='Integration customer'):
        type(self).serial += 1
        return self.ok('customer.save', {'name': name + ' ' + str(self.serial), 'email': 'customer@example.invalid', 'phone': '+38970000000', 'address': 'Private fixture address', 'notes': 'PRIVATE-NOTES-NOT-PUBLIC'})

    def unit(self, maintenance=False):
        type(self).serial += 1
        return self.ok('unit.save', {'location_id': 2, 'number': 'TEST-' + str(self.serial), 'size_m2': 4.5, 'price_minor': 100000, 'maintenance': maintenance})

    def rental(self, unit=None, customer=None):
        unit, customer = unit or self.unit(), customer or self.customer()
        payload = {'unit_id': unit['id'], 'customer_id': customer['id'], 'start_date': self.today, 'monthly_price_minor': 100000, 'billing_day': 1}
        return self.ok('rental.save', payload)

    def draft(self, customer=None, rental=None, items=None):
        customer = customer or self.customer()
        payload = {'customer_id': customer['id'], 'issue_date': self.today, 'due_date': self.today, 'currency': 'MKD', 'notes': '<script>alert(1)</script>', 'items': items or [{'description': 'Storage <img src=x onerror=alert(1)>', 'quantity': 1, 'unit_price_minor': 10000, 'tax_percent': 18}]}
        if rental:
            payload['rental_id'] = rental['id']
        return self.ok('invoice.save', payload)

    def issued(self, **kwargs):
        draft = self.draft(**kwargs)
        return self.ok('invoice.issue', {'id': draft['id']})

    def test_01_seed_setup_and_public_privacy(self):
        data = self.ok('bootstrap')
        self.assertEqual(len(data['units']), 118)
        self.assertEqual(sorted(int(u['number']) for u in data['units']), list(range(1, 119)))
        self.assertTrue(all(u['location_id'] == 1 and u['status'] == 'maintenance' for u in data['units']))
        for key in ('customers', 'rentals', 'invoices', 'payments', 'contracts'):
            self.assertEqual(data[key], [], key)
        self.assertEqual(data['summary']['occupied'], 0)
        self.assertEqual(len(data['locations']), 2)
        status, page, _ = self.anonymous.raw('/setup.php')
        self.assertEqual(status, 200)
        self.assertIn(b'Setup is locked', page)
        public = self.ok('public', client=self.anonymous)
        self.assertEqual(set(public), {'settings', 'locations'})
        self.assertFalse({'bank_account', 'tax_id', 'invoice_prefix'} & set(public['settings']))
        for path in ('/config.php', '/app/schema.sql', '/storage/test.sqlite', '/../config.php'):
            status, body, _ = self.anonymous.raw(path)
            # PHP's built-in server may fall back to index.php (HTTP 200).
            # Verify confidentiality, rather than assuming Apache's status code.
            self.assertNotIn(SETUP_KEY.encode(), body)
            self.assertNotIn(b'SQLite format 3', body)
            self.assertNotIn(b'CREATE TABLE IF NOT EXISTS users', body)
        self.assertFalse((self.root / 'public' / 'config.php').exists())
        self.assertFalse((self.root / 'public' / 'storage').exists())

    def test_02_auth_csrf_origin_and_logout(self):
        for action in ('bootstrap', 'report', 'export&type=customers'):
            self.rejected(action, expected=401, client=self.anonymous)
        self.rejected('customer.save', {'name': 'No login'}, expected=401, client=self.anonymous)
        self.rejected('login', {'email': EMAIL, 'password': 'incorrect'}, expected=401, client=self.anonymous)
        self.rejected('customer.save', {'name': 'No CSRF'}, expected=403, csrf=False)
        self.rejected('customer.save', {'name': 'Wrong CSRF'}, expected=403, extra_headers={'X-CSRF-Token': 'invalid'}, csrf=False)
        self.rejected('customer.save', {'name': 'Cross origin'}, expected=403, extra_headers={'Origin': 'https://attacker.invalid'})
        self.ok('logout', {})
        self.rejected('bootstrap', expected=401)

    def test_03_rental_lifecycle_and_double_booking(self):
        customer, held = self.customer(), self.unit(True)
        payload = {'unit_id': held['id'], 'customer_id': customer['id'], 'start_date': self.today, 'monthly_price_minor': 100000}
        self.rejected('rental.save', payload)
        unit = self.unit()
        payload['unit_id'] = unit['id']
        rental = self.ok('rental.save', payload)
        self.rejected('rental.save', payload)
        self.assertEqual(next(u for u in self.ok('bootstrap')['units'] if u['id'] == unit['id'])['status'], 'occupied')
        self.rejected('unit.save', {**unit, 'maintenance': True})
        self.rejected('unit.delete', {'id': unit['id']})
        self.rejected('customer.delete', {'id': customer['id']})
        self.ok('rental.end', {'id': rental['id'], 'end_date': self.today})
        self.assertEqual(next(u for u in self.ok('bootstrap')['units'] if u['id'] == unit['id'])['status'], 'available')
        self.rejected('rental.save', {**payload, 'id': rental['id']})
        self.ok('rental.save', payload)
        self.assertEqual(self.sql("SELECT count(*) n FROM rentals WHERE unit_id=? AND status='active'", (unit['id'],))[0]['n'], 1)
        self.rejected('unit.save', {**unit, 'id': 0})

    def test_04_invoice_arithmetic_edit_issue_and_snapshots(self):
        customer = self.customer('Original legal name')
        draft = self.draft(customer=customer, items=[{'description': 'Fractional storage', 'quantity': 1.25, 'unit_price_minor': 9999, 'tax_percent': 18}, {'description': 'Extra', 'quantity': 3, 'unit_price_minor': 333, 'tax_percent': 5}])
        self.assertEqual((draft['subtotal_minor'], draft['tax_minor'], draft['total_minor']), (13498, 2300, 15798))
        self.assertIsNone(draft['number'])
        self.assertEqual(draft['status'], 'draft')
        self.rejected('payment.confirm', {'invoice_id': draft['id'], 'amount_minor': 10, 'paid_date': self.today})
        edited = self.ok('invoice.save', {**draft, 'items': [{'description': 'Edited line', 'quantity': .5, 'unit_price_minor': 101, 'tax_percent': 18}]})
        self.assertEqual((edited['subtotal_minor'], edited['tax_minor'], edited['total_minor']), (51, 9, 60))
        invoice = self.ok('invoice.issue', {'id': draft['id']})
        self.assertRegex(invoice['number'], r'^DM-\d{4}-\d{5}$')
        self.assertEqual(invoice['status'], 'issued')
        self.assertEqual(invoice['paid_minor'], 0)
        self.rejected('invoice.issue', {'id': invoice['id']})
        self.rejected('invoice.save', {**edited, 'items': [{'description': 'Forbidden edit', 'quantity': 1, 'unit_price_minor': 999, 'tax_percent': 0}]})
        self.ok('customer.save', {**customer, 'name': 'Changed live customer'})
        status, page, headers = self.client.raw(f'/document.php?type=invoice&id={invoice["id"]}')
        self.assertEqual(status, 200)
        self.assertIn(customer['name'].encode(), page)
        self.assertNotIn(b'Changed live customer', page)
        self.assertNotIn(b'<script>alert(1)</script>', page)
        self.assertIn('no-store', headers.get('Cache-Control', ''))
        second = self.issued()
        self.assertEqual(int(second['number'].split('-')[-1]), int(invoice['number'].split('-')[-1]) + 1)

    def test_05_payments_partial_overpayment_reversal_and_audit(self):
        invoice = self.issued()
        p1 = self.ok('payment.confirm', {'invoice_id': invoice['id'], 'amount_minor': 3000, 'paid_date': self.today, 'method': 'bank', 'reference': 'CONFIRM-1'})
        self.assertEqual((p1['invoice']['status'], p1['invoice']['balance_minor']), ('partial', 8800))
        self.rejected('payment.confirm', {'invoice_id': invoice['id'], 'amount_minor': 8801, 'paid_date': self.today})
        self.rejected('invoice.void', {'id': invoice['id'], 'reason': 'Has payment'})
        p2 = self.ok('payment.confirm', {'invoice_id': invoice['id'], 'amount_minor': 8800, 'paid_date': self.today, 'method': 'cash'})
        self.assertEqual((p2['invoice']['status'], p2['invoice']['balance_minor']), ('paid', 0))
        reversed_payment = self.ok('payment.void', {'id': p2['payment']['id'], 'reason': 'Payment entered in error'})
        self.assertEqual((reversed_payment['invoice']['status'], reversed_payment['invoice']['balance_minor']), ('partial', 8800))
        self.rejected('payment.void', {'id': p2['payment']['id'], 'reason': 'Repeat reversal'})
        audit = self.sql("SELECT action,details FROM audit WHERE entity='payments' AND entity_id=? ORDER BY id", (p2['payment']['id'],))
        self.assertEqual([row['action'] for row in audit], ['confirm', 'void'])
        self.assertIn('Payment entered in error', audit[1]['details'])
        report = self.ok('report&from=' + self.today + '&to=' + self.today)
        mkd = next(row for row in report['payments'] if row['currency'] == 'MKD')
        self.assertEqual(mkd['amount_minor'], 3000)

    def test_06_share_links_are_scoped_expiring_revocable(self):
        invoice, other = self.issued(), self.issued()
        share = self.ok('invoice.share', {'id': invoice['id'], 'days': 1})
        parts = urllib.parse.urlsplit(share['url'])
        path = parts.path + '?' + parts.query
        token = urllib.parse.parse_qs(parts.query)['token'][0]
        self.assertNotEqual(self.sql('SELECT token_hash FROM invoice_shares WHERE invoice_id=?', (invoice['id'],))[0]['token_hash'], token)
        self.assertEqual(self.anonymous.raw(path)[0], 200)
        self.assertEqual(self.anonymous.raw(path.replace(f'id={invoice["id"]}&', f'id={other["id"]}&'))[0], 404)
        self.assertEqual(self.anonymous.raw(f'/document.php?type=contract&id=1&token={token}')[0], 401)
        self.assertEqual(self.anonymous.raw(f'/document.php?type=invoice&id={invoice["id"]}')[0], 401)
        self.sql("UPDATE invoice_shares SET expires_at='2000-01-01 00:00:00' WHERE invoice_id=?", (invoice['id'],))
        self.assertEqual(self.anonymous.raw(path)[0], 404)
        share = self.ok('invoice.share', {'id': invoice['id']})
        path = share['url'].removeprefix(self.base)
        self.ok('invoice.revoke', {'id': invoice['id']})
        self.assertEqual(self.anonymous.raw(path)[0], 404)
        self.rejected('invoice.share', {'id': self.draft()['id']})
        self.rejected('invoice.email', {'id': invoice['id']})

    def test_07_csv_formula_neutralization(self):
        customer = self.customer('=HYPERLINK("https://example.invalid")')
        self.ok('customer.save', {**customer, 'address': '@SUM(1+1)', 'phone': '+38970000000'})
        status, body, headers = self.client.raw('/api.php?action=export&type=customers')
        self.assertEqual(status, 200)
        self.assertIn('text/csv', headers['Content-Type'])
        self.assertIn(b"'=HYPERLINK", body)
        self.assertIn(b"'@SUM", body)
        self.assertIn(b"'+389", body)
        self.assertNotIn(b'PRIVATE-NOTES-NOT-PUBLIC', body)

    def test_08_private_pdf_upload_and_path_protection(self):
        rental = self.rental()
        contract = self.ok('contract.save', {'rental_id': rental['id'], 'title': 'Fixture agreement', 'body': 'Private agreement text', 'status': 'draft'})
        def upload(content, filename):
            boundary = 'integration-boundary-73b9'
            fields = {'id': str(contract['id']), 'csrf': self.client.csrf}
            body = b''
            for key, value in fields.items():
                body += f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n{value}\r\n'.encode()
            body += f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{filename}"\r\nContent-Type: application/pdf\r\n\r\n'.encode() + content + f'\r\n--{boundary}--\r\n'.encode()
            status, response, _ = self.client.raw('/api.php?action=contract.upload', body, {'Content-Type': 'multipart/form-data; boundary=' + boundary})
            return status, json.loads(response)
        self.assertEqual(upload(b'<?php echo "not PDF";', 'evil.pdf')[0], 422)
        pdf = b'%PDF-1.4\n1 0 obj\n<< /Type /Catalog >>\nendobj\ntrailer\n<< /Root 1 0 R >>\n%%EOF\n'
        status, response = upload(pdf, '../../unsafe-name.pdf')
        self.assertEqual(status, 200, response)
        stored = response['data']['attachment']
        self.assertRegex(stored, r'^[a-f0-9]{48}\.pdf$')
        path = f'/document.php?type=attachment&id={contract["id"]}'
        self.assertEqual(self.anonymous.raw(path)[0], 401)
        status, body, headers = self.client.raw(path)
        self.assertEqual(status, 200)
        self.assertEqual(body, pdf)
        self.assertEqual(headers['Content-Type'], 'application/pdf')
        public_body = self.anonymous.raw('/storage/contracts/' + stored)[1]
        self.assertNotEqual(public_body, pdf)
        self.assertNotIn(pdf, public_body)
        self.assertTrue((self.root / 'storage' / 'contracts' / stored).is_file())
        self.sql('UPDATE contracts SET attachment=? WHERE id=?', ('../../config.php', contract['id']))
        self.assertEqual(self.client.raw(path)[0], 404)


    def test_09_monthly_drafts_are_idempotent_and_require_review(self):
        customer = self.customer('Monthly billing fixture')
        rental = self.rental(customer=customer)
        month = self.today[:7]
        first = self.ok('invoice.monthly', {'month': month})
        created = [invoice for invoice in first['created'] if invoice['rental_id'] == rental['id']]
        self.assertEqual(len(created), 1)
        invoice = created[0]
        self.assertEqual(invoice['total_minor'], rental['monthly_price_minor'])
        self.assertEqual(invoice['status'], 'draft')
        self.assertEqual(invoice['tax_minor'], 0)
        self.assertEqual(invoice['paid_minor'], 0)
        self.assertIsNone(invoice['number'])
        second = self.ok('invoice.monthly', {'month': month})
        self.assertEqual(second['created_count'], 0)
        self.assertGreaterEqual(second['skipped_count'], 1)
        self.assertEqual(self.sql('SELECT count(*) n FROM invoice_periods WHERE rental_id=? AND period=?', (rental['id'], month))[0]['n'], 1)
        self.rejected('invoice.monthly', {'month': '2026-13'})
        self.rejected('invoice.monthly', {'month': '2026-1'})
        # Existing documents retain the original customer/unit relationship.
        other = self.customer('Another customer')
        self.rejected('rental.save', {**rental, 'customer_id': other['id']})
        self.assertFalse(self.sql("SELECT id FROM audit WHERE action='email-accepted'"))

    def test_10_password_change_rotates_csrf_and_alias_restores(self):
        self.rejected('password.change', {'current_password': 'wrong', 'password': 'Replacement-Password-2026!'})
        self.rejected('password.change', {'current_password': PASSWORD, 'password': 'short'})
        old_csrf = self.client.csrf
        replacement = 'Replacement-Password-2026!'
        changed = self.ok('password.change', {'current_password': PASSWORD, 'password': replacement})
        self.assertNotEqual(changed['csrf'], old_csrf)
        self.rejected('customer.save', {'name': 'Stale CSRF'}, expected=403)
        self.client.csrf = changed['csrf']
        self.rejected('login', {'email': EMAIL, 'password': PASSWORD}, expected=401, client=self.anonymous)
        status, result, _ = self.anonymous.api('login', {'email': EMAIL, 'password': replacement}, csrf=False)
        self.assertEqual(status, 200, result)
        try:
            restored = self.ok('profile.password', {'current_password': replacement, 'new_password': PASSWORD})
            self.assertNotEqual(restored['csrf'], changed['csrf'])
            self.client.csrf = restored['csrf']
        finally:
            # Keep the isolated fixture usable if alias support regresses.
            status, result, _ = self.client.api('password.change', {'current_password': replacement, 'password': PASSWORD})
        self.client.login()

    def test_11_login_rate_limit(self):
        for _ in range(8):
            self.rejected('login', {'email': 'blocked-fixture@example.invalid', 'password': 'incorrect'}, expected=401, client=self.anonymous)
        self.rejected('login', {'email': 'blocked-fixture@example.invalid', 'password': 'incorrect'}, expected=429, client=self.anonymous)


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