"""Opt-in headed Chromium tests against a disposable *_test PostgreSQL database."""

from contextlib import redirect_stderr, redirect_stdout
from decimal import Decimal
import gzip
import io
import json
import os
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch

from psycopg import sql

from car_tracker import database, pipeline
from car_tracker.collect import search_url
import test_database_integration as database_tests


@unittest.skipUnless(
    os.environ.get("RUN_BROWSER_TESTS") == "1" and os.environ.get("TEST_DATABASE_URL"),
    "RUN_BROWSER_TESTS=1 and TEST_DATABASE_URL are required",
)
class BrowserPipelineTests(unittest.TestCase):
    tables = database_tests.DatabaseIntegrationTests.tables

    @classmethod
    def setUpClass(cls):
        cls.conn = database_tests.DatabaseIntegrationTests.connect()
        cls.addClassCleanup(cls.conn.close)
        database.migrate(cls.conn)

    def setUp(self):
        self.conn.execute(sql.SQL("TRUNCATE {} RESTART IDENTITY").format(
            sql.SQL(", ").join(sql.Identifier("silver", table) for table in self.tables)
        ))
        temporary = TemporaryDirectory()
        self.addCleanup(temporary.cleanup)
        self.directory = Path(temporary.name)
        self.config = self.directory / "searches.toml"
        self.config.write_text(
            '[[searches]]\nfamily = "synthetic"\ntext = "synthetic car"\n',
            encoding="utf-8",
        )
        self.output = self.directory / "raw"
        self.search = {"list_id": 123, "price": ["12000"]}
        self.detail = {
            "list_id": 123, "price": ["12500.50"],
            "body": "Synthetic private description.\nRetain this raw text exactly.",
            "location": {"lat": 48.125, "lng": 2.375},
        }

    def run_browser_pipeline(self, *, detail_status=200, import_status="succeeded"):
        # Import only when opted in; never replace the browser, collector or importer.
        from playwright.sync_api import BrowserContext

        original_new_page = BrowserContext.new_page
        search = search_url("synthetic car", 1)
        detail = "https://www.leboncoin.fr/ad/voitures/123"
        requests, responses, unexpected = [], [], []

        def handle(route):
            url = route.request.url
            requests.append(url)
            if url == search:
                status = 200
                props = {"searchData": {"ads": [self.search], "total": 1, "max_pages": 1}}
            elif url == detail:
                status = detail_status
                # Even the 403 has valid data: ignoring HTTP status must fail this test.
                props = {"ad": self.detail}
            else:
                unexpected.append(url)
                route.abort()
                return
            data = json.dumps({"props": {"pageProps": props}})
            route.fulfill(
                status=status, content_type="text/html",
                body=f'<!doctype html><script id="__NEXT_DATA__" type="application/json">{data}</script>',
            )

        def new_page(context, *args, **kwargs):
            page = original_new_page(context, *args, **kwargs)
            page.route("**/*", handle)
            page.on("response", lambda response: responses.append((response.url, response.status)))
            return page

        stdout, stderr = io.StringIO(), io.StringIO()
        # Preserve DISPLAY/browser paths, but discard inherited libpq/homelab settings.
        environment = {key: value for key, value in os.environ.items() if not key.startswith("PG")}
        environment["DATABASE_URL"] = os.environ["TEST_DATABASE_URL"]
        with patch.dict(os.environ, environment, clear=True), patch.object(
            BrowserContext, "new_page", new_page,
        ), patch("car_tracker.collect.time.sleep"), redirect_stdout(stdout), redirect_stderr(stderr):
            code = pipeline.main([
                "--headed", "--config", str(self.config), "--output", str(self.output),
                "--delay", "1", "--max-pages", "1", "--max-details", "1",
            ])

        self.assertEqual(unexpected, [])
        self.assertEqual(requests, [search, detail])
        self.assertEqual(responses, [(search, 200), (detail, detail_status)])
        self.assertNotIn(self.detail["body"], stdout.getvalue() + stderr.getvalue())
        self.assertNotIn("48.125", stdout.getvalue() + stderr.getvalue())
        self.assertNotIn("2.375", stdout.getvalue() + stderr.getvalue())
        paths = list(self.output.rglob("manifest.json"))
        self.assertEqual(len(paths), 1)
        path = paths[0]
        manifest = json.loads(path.read_text(encoding="utf-8"))
        state = json.loads((path.parent / "pipeline.json").read_text(encoding="utf-8"))
        self.assertEqual(state["collection_status"], manifest["status"])
        self.assertEqual(state["import_status"], import_status)
        if import_status == "succeeded":
            self.assertEqual(state["import_summary"]["status"], "imported")
            self.assertEqual(state["import_summary"]["source_status"], manifest["status"])
            self.assertEqual(state["import_summary"]["rejected"], 0)
        self.assertEqual(self.conn.execute(
            "SELECT run_id::text, source_status, manifest FROM silver.collection_runs"
        ).fetchall(), [(manifest["run_id"], manifest["status"], manifest)])
        self.assertEqual(self.conn.execute(
            "SELECT family, status, pages_observed, records_reported FROM silver.search_coverage"
        ).fetchall(), [("synthetic", "complete", 1, 1)])
        with gzip.open(path.parent / manifest["searches"][0]["pages"][0]["file"], "rt") as stream:
            records = [json.loads(line) for line in stream]
        self.assertEqual(len(records), 1)
        self.assertEqual(records[0]["record_type"], "search")
        self.assertEqual(records[0]["payload"], self.search)
        return code, path, manifest

    def test_complete_run_preserves_raw_and_typed_detail_and_replays(self):
        code, path, manifest = self.run_browser_pipeline()
        self.assertEqual(code, 0)
        self.assertEqual(manifest["status"], "complete")
        with gzip.open(path.parent / manifest["details"]["listings"]["123"]["file"], "rt") as stream:
            records = [json.loads(line) for line in stream]
        self.assertEqual(len(records), 1)
        self.assertEqual(records[0]["record_type"], "detail")
        self.assertEqual(records[0]["payload"], self.detail)
        self.assertEqual(self.conn.execute(
            "SELECT source_listing_id, record_type, price_eur, currency, description, "
            "latitude, longitude, raw_payload FROM silver.observations ORDER BY record_type"
        ).fetchall(), [
            ("123", "detail", Decimal("12500.50"), "EUR", self.detail["body"],
             48.125, 2.375, self.detail),
            ("123", "search", Decimal("12000"), "EUR", None, None, None, self.search),
        ])
        self.assertEqual(self.conn.execute(
            "SELECT source_listing_id, status FROM silver.detail_coverage"
        ).fetchall(), [("123", "complete")])
        before = self.state()
        self.assertEqual({table: len(rows) for table, rows in before.items()}, {
            "collection_runs": 1, "raw_files": 2, "search_coverage": 1,
            "detail_coverage": 1, "listings": 1, "observations": 2, "rejected_records": 0,
        })
        with redirect_stdout(io.StringIO()):
            replay = database.import_run(self.conn, path)
        self.assertEqual(replay, {
            "run_id": manifest["run_id"], "status": "already_imported", "rejected": 0,
        })
        self.assertEqual(self.state(), before)

    def test_blocked_detail_imports_partial_run_and_retains_search(self):
        code, path, manifest = self.run_browser_pipeline(detail_status=403)
        self.assertEqual(code, 1)
        self.assertEqual(manifest["status"], "partial")
        self.assertEqual(manifest["error_type"], "SourceAccessError")
        self.assertEqual(manifest["details"]["status"], "partial")
        self.assertEqual(manifest["details"]["listings"]["123"]["http_status"], 403)
        self.assertFalse((path.parent / "detail-123.jsonl.gz").exists())
        self.assertEqual(self.conn.execute(
            "SELECT source_listing_id, status, reason, http_status FROM silver.detail_coverage"
        ).fetchall(), [("123", "failed", "SourceAccessError", 403)])
        self.assertEqual(self.conn.execute(
            "SELECT source_listing_id, record_type, price_eur, raw_payload FROM silver.observations"
        ).fetchall(), [("123", "search", Decimal("12000"), self.search)])
        self.assertEqual(self.conn.execute(
            "SELECT record_type, expected_records, accepted_records, rejected_records FROM silver.raw_files"
        ).fetchall(), [("search", 1, 1, 0)])
        self.assertEqual(self.conn.execute("SELECT count(*) FROM silver.rejected_records").fetchone(), (0,))

    def state(self):
        return {table: self.conn.execute(sql.SQL(
            "SELECT to_jsonb(t) FROM silver.{} t ORDER BY to_jsonb(t)::text"
        ).format(sql.Identifier(table))).fetchall() for table in self.tables}

    def test_interrupt_after_commit_is_unknown_and_replay_is_safe(self):
        def committed_then_interrupted(conn, path):
            database.import_run(conn, path)
            raise KeyboardInterrupt

        with patch.object(pipeline, "import_run", side_effect=committed_then_interrupted):
            code, path, _ = self.run_browser_pipeline(import_status="unknown")
        self.assertEqual(code, 130)
        before = self.state()
        self.assertEqual(len(before["observations"]), 2)
        self.assertEqual(database.import_run(self.conn, path)["status"], "already_imported")
        self.assertEqual(self.state(), before)


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