mirror of
https://github.com/wahyd4/heygo.git
synced 2026-08-09 05:15:52 +10:00
341 lines
14 KiB
Python
341 lines
14 KiB
Python
#!/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)
|