#!/usr/bin/env python3
"""
Tests for scripts/export-django-links.py.

Run:
    python3 scripts/export-django-links.test.py
or
    pytest scripts/export-django-links.test.py

These tests build a throwaway SQLite DB that mimics the old Django `links`
app schema (auth_user, django_content_type, links_link, links_tag,
links_link_tags) and asserts the exporter reads it correctly.
"""
from __future__ import annotations

import io
import json
import os
import sqlite3
import tempfile
import unittest
from contextlib import redirect_stderr, redirect_stdout

# Make the script importable by path (scripts/ is not a package and the
# filename contains hyphens, so a normal `import` won't work).
HERE = os.path.dirname(os.path.abspath(__file__))
_SCRIPT_PATH = os.path.join(HERE, "export-django-links.py")

import importlib.util  # noqa: E402

_spec = importlib.util.spec_from_file_location("export_django_links", _SCRIPT_PATH)
assert _spec is not None and _spec.loader is not None
exporter_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(exporter_mod)


def _make_fixture(db_path: str, with_owner: bool = False) -> sqlite3.Connection:
    """Create a Django-shaped links DB at db_path with a couple of rows."""
    conn = sqlite3.connect(db_path)
    c = conn.cursor()

    c.execute("CREATE TABLE django_content_type (id INTEGER, app_label TEXT, model TEXT)")
    c.execute("INSERT INTO django_content_type (id, app_label, model) VALUES (1,'links','link')")
    c.execute("INSERT INTO django_content_type (id, app_label, model) VALUES (2,'links','tag')")

    c.execute(
        "CREATE TABLE auth_user (id INTEGER PRIMARY KEY, username TEXT, email TEXT, is_staff INTEGER)"
    )
    c.execute("INSERT INTO auth_user (id, username, email, is_staff) VALUES (1,'alice','alice@example.com',1)")
    c.execute("INSERT INTO auth_user (id, username, email, is_staff) VALUES (2,'bob','bob@example.com',1)")

    link_cols = [
        "id INTEGER PRIMARY KEY",
        "alias TEXT UNIQUE",
        "original_url TEXT",
        "text TEXT",
        "link_type TEXT",
        "click_count INTEGER DEFAULT 0",
        "description TEXT",
        "created_at TEXT",
        "updated_at TEXT",
    ]
    if with_owner:
        link_cols.append("owner_id INTEGER REFERENCES auth_user(id)")
    c.execute(f"CREATE TABLE links_link ({', '.join(link_cols)})")

    c.execute(
        "CREATE TABLE links_tag (id INTEGER PRIMARY KEY, name TEXT UNIQUE, slug TEXT, "
        "description TEXT, created_at TEXT, updated_at TEXT)"
    )

    c.execute(
        "CREATE TABLE links_link_tags (id INTEGER PRIMARY KEY, link_id INTEGER, tag_id INTEGER)"
    )

    # Link 1: redirect with two tags, owned by alice (when with_owner).
    c.execute(
        "INSERT INTO links_link (id, alias, original_url, text, link_type, description, "
        "created_at, updated_at, owner_id) "
        "VALUES (1,'go','https://example.com/go','','LINK','A redirect',"
        "'2024-01-01 00:00:00','2024-01-02 00:00:00',1)"
        if with_owner else
        "INSERT INTO links_link (id, alias, original_url, text, link_type, description, "
        "created_at, updated_at) "
        "VALUES (1,'go','https://example.com/go','','LINK','A redirect',"
        "'2024-01-01 00:00:00','2024-01-02 00:00:00')"
    )
    c.execute("INSERT INTO links_tag (id, name, slug) VALUES (1,'work','work')")
    c.execute("INSERT INTO links_tag (id, name, slug) VALUES (2,'search','search')")
    c.execute("INSERT INTO links_link_tags (id, link_id, tag_id) VALUES (1,1,1)")
    c.execute("INSERT INTO links_link_tags (id, link_id, tag_id) VALUES (2,1,2)")

    # Link 2: custom (markdown content), owned by bob (when with_owner).
    c.execute(
        "INSERT INTO links_link (id, alias, original_url, text, link_type, description, "
        "created_at, updated_at, owner_id) "
        "VALUES (2,'note','','# Notes','CUSTOM','A custom page',"
        "'2024-02-01 00:00:00','2024-02-03 00:00:00',2)"
        if with_owner else
        "INSERT INTO links_link (id, alias, original_url, text, link_type, description, "
        "created_at, updated_at) "
        "VALUES (2,'note','','# Notes','CUSTOM','A custom page',"
        "'2024-02-01 00:00:00','2024-02-03 00:00:00')"
    )
    c.execute("INSERT INTO links_tag (id, name, slug) VALUES (3,'personal','personal')")
    c.execute("INSERT INTO links_link_tags (id, link_id, tag_id) VALUES (3,2,3)")

    conn.commit()
    return conn


def _read_jsonl(path: str) -> list[dict]:
    out = []
    with open(path, "r", encoding="utf-8") as fh:
        for line in fh:
            line = line.rstrip("\n")
            if line:
                out.append(json.loads(line))
    return out


class ExportTests(unittest.TestCase):
    def setUp(self):
        self.tmpdir = tempfile.mkdtemp(prefix="heygo-export-test-")
        self.db_path = os.path.join(self.tmpdir, "db.sqlite3")

    def tearDown(self):
        # Clear env so it doesn't leak between tests.
        os.environ.pop("DJANGO_DB_PATH", None)

    def _run(self, argv: list[str]) -> tuple[int, str, str]:
        out = io.StringIO()
        err = io.StringIO()
        try:
            with redirect_stdout(out), redirect_stderr(err):
                code = exporter_mod.main(argv)
            return code, out.getvalue(), err.getvalue()
        except SystemExit as e:
            code = e.code if e.code is not None else 1
            return int(code), out.getvalue(), err.getvalue()

    def test_basic_export_to_file(self):
        _make_fixture(self.db_path, with_owner=False)
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, err = self._run(["--db", self.db_path, "--output", out_path])
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(len(rows), 2)
        # Ordered by alias ASC: 'go' then 'note'
        self.assertEqual(rows[0]["alias"], "go")
        self.assertEqual(rows[1]["alias"], "note")
        # Field mapping for the redirect link
        self.assertEqual(rows[0]["link_type"], "redirect")
        self.assertEqual(rows[0]["target_url"], "https://example.com/go")
        self.assertEqual(rows[0]["content_markdown"], "")
        self.assertEqual(rows[0]["description"], "A redirect")
        self.assertEqual(rows[0]["tags"], ["search", "work"])  # sorted asc by name
        self.assertEqual(rows[0]["created_at"], "2024-01-01 00:00:00")
        self.assertEqual(rows[0]["updated_at"], "2024-01-02 00:00:00")
        # Field mapping for the custom link
        self.assertEqual(rows[1]["link_type"], "custom")
        self.assertEqual(rows[1]["content_markdown"], "# Notes")
        self.assertEqual(rows[1]["tags"], ["personal"])
        self.assertIn("exported 2 link(s)", err)

    def test_link_type_mapping_defaults(self):
        # link_type null -> redirect; unknown value preserved as-is
        conn = _make_fixture(self.db_path, with_owner=False)
        conn.execute(
            "INSERT INTO links_link (id, alias, original_url, text, link_type, created_at, updated_at) "
            "VALUES (3,'bare','https://example.com/bare',NULL,NULL,'2024-03-01','2024-03-01')"
        )
        conn.execute(
            "INSERT INTO links_link (id, alias, original_url, text, link_type, created_at, updated_at) "
            "VALUES (4,'weird','https://example.com/weird','','BOGUS','2024-03-02','2024-03-02')"
        )
        conn.commit()
        conn.close()
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, _ = self._run(["--db", self.db_path, "--output", out_path])
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        by_alias = {r["alias"]: r for r in rows}
        self.assertEqual(by_alias["bare"]["link_type"], "redirect")
        self.assertEqual(by_alias["weird"]["link_type"], "BOGUS")

    def test_owner_filter_with_owner_column(self):
        _make_fixture(self.db_path, with_owner=True)
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, err = self._run(
            ["--db", self.db_path, "--output", out_path, "--owner-email", "alice@example.com"]
        )
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["alias"], "go")  # owned by alice (id=1)
        self.assertIn("exported 1 link(s)", err)

    def test_owner_filter_email_case_insensitive(self):
        _make_fixture(self.db_path, with_owner=True)
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, _ = self._run(
            ["--db", self.db_path, "--output", out_path, "--owner-email", "ALICE@example.COM"]
        )
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["alias"], "go")

    def test_owner_filter_no_match_is_empty(self):
        _make_fixture(self.db_path, with_owner=True)
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, err = self._run(
            ["--db", self.db_path, "--output", out_path, "--owner-email", "nobody@example.com"]
        )
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(rows, [])
        self.assertIn("no auth_user row matches", err)

    def test_owner_filter_without_owner_column_warns_and_exports_all(self):
        _make_fixture(self.db_path, with_owner=False)
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, err = self._run(
            ["--db", self.db_path, "--output", out_path, "--owner-email", "alice@example.com"]
        )
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(len(rows), 2)  # no owner column -> exported all
        self.assertIn("owner filter cannot be applied", err)

    def test_env_var_db_path(self):
        _make_fixture(self.db_path, with_owner=False)
        os.environ["DJANGO_DB_PATH"] = self.db_path
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, _ = self._run(["--output", out_path])  # no --db
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(len(rows), 2)

    def test_db_arg_overrides_env_var(self):
        _make_fixture(self.db_path, with_owner=False)
        other = os.path.join(self.tmpdir, "empty.sqlite3")
        sqlite3.connect(other).close()  # empty DB (no tables) -> error
        os.environ["DJANGO_DB_PATH"] = other
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, _ = self._run(["--db", self.db_path, "--output", out_path])
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(len(rows), 2)

    def test_missing_db_path_exits_nonzero(self):
        code, _, err = self._run(["--db", "/no/such/file.sqlite3"])
        self.assertEqual(code, 2)
        self.assertIn("DB file not found", err)

    def test_no_db_path_arg_or_env_exits_nonzero(self):
        os.environ.pop("DJANGO_DB_PATH", None)
        code, _, err = self._run([])
        self.assertEqual(code, 2)
        self.assertIn("no DB path provided", err)

    def test_db_path_is_directory_exits_nonzero(self):
        code, _, err = self._run(["--db", self.tmpdir])
        self.assertEqual(code, 2)
        self.assertIn("is a directory", err)

    def test_empty_links_table_exports_zero(self):
        conn = _make_fixture(self.db_path, with_owner=False)
        conn.execute("DELETE FROM links_link")
        conn.execute("DELETE FROM links_link_tags")
        conn.commit()
        conn.close()
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, err = self._run(["--db", self.db_path, "--output", out_path])
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(rows, [])
        self.assertIn("exported 0 link(s)", err)

    def test_stdout_default_output(self):
        _make_fixture(self.db_path, with_owner=False)
        code, out, _ = self._run(["--db", self.db_path])
        self.assertEqual(code, 0)
        lines = [ln for ln in out.splitlines() if ln.strip()]
        self.assertEqual(len(lines), 2)
        first = json.loads(lines[0])
        self.assertEqual(first["alias"], "go")

    def test_no_tables_in_db_exits_nonzero(self):
        empty = os.path.join(self.tmpdir, "empty.sqlite3")
        sqlite3.connect(empty).close()
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, err = self._run(["--db", empty, "--output", out_path])
        self.assertEqual(code, 1)
        self.assertIn("could not find the links_link table", err)

    def test_table_discovery_fallback_without_content_type(self):
        # Drop django_content_type to exercise the sqlite_master fallback.
        conn = _make_fixture(self.db_path, with_owner=False)
        conn.execute("DROP TABLE django_content_type")
        conn.commit()
        conn.close()
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, _ = self._run(["--db", self.db_path, "--output", out_path])
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(len(rows), 2)

    def test_tags_absent_when_no_tag_table(self):
        conn = _make_fixture(self.db_path, with_owner=False)
        conn.execute("DROP TABLE links_link_tags")
        conn.execute("DROP TABLE links_tag")
        conn.commit()
        conn.close()
        out_path = os.path.join(self.tmpdir, "out.jsonl")
        code, _, _ = self._run(["--db", self.db_path, "--output", out_path])
        self.assertEqual(code, 0)
        rows = _read_jsonl(out_path)
        self.assertEqual(len(rows), 2)
        for r in rows:
            self.assertEqual(r["tags"], [])


class ArgParserTests(unittest.TestCase):
    def test_help_lists_all_options(self):
        p = exporter_mod.build_arg_parser()
        h = p.format_help()
        for token in ("--db", "--output", "--owner-email", "DJANGO_DB_PATH"):
            self.assertIn(token, h)

    def test_defaults(self):
        ns = exporter_mod.build_arg_parser().parse_args([])
        self.assertIsNone(ns.db)
        self.assertIsNone(ns.output)
        self.assertIsNone(ns.owner_email)


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