test(minimal-workspace): cover the onboarding notebook and .scripts helpers

src/create_dlthub_workspace/scaffolds/minimal_workspace/.scripts/serve_headless.py

@@ -22,17 +22,17 @@
 
 import sys
 
+if len(sys.argv) < 2:
+    sys.exit("usage: uv run .scripts/serve_headless.py <job-ref>   e.g. jobs.onboarding_success")
+ref = sys.argv[1]
+
 import dlt_runtime._runtime_command as rc
 
 # The only browser-open in serve's flow (_runtime_command.py: _do_launch). serve
 # is decorated with @requires_login/@requires_workspace, which inject
 # auth_service + api_client, so calling it directly runs the full flow.
 rc._open_app_url = lambda *args, **kwargs: None
 
-if len(sys.argv) < 2:
-    sys.exit("usage: uv run .scripts/serve_headless.py <job-ref>   e.g. jobs.onboarding_success")
-ref = sys.argv[1]
-
 # `interactive` selects all interactive jobs as candidates (regardless of
 # trigger); `job_ref=ref` narrows to exactly this one without prompting.
 rc.serve(selector_or_job_ref="interactive", job_ref=ref, follow=False)

tests/test_notebook.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import json
+import py_compile
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from create_dlthub_workspace.scaffold import SCAFFOLDS_DIR, copy_scaffold
+
+if sys.version_info >= (3, 11):
+    import tomllib
+else:
+    import tomli as tomllib
+
+WORKSPACE = SCAFFOLDS_DIR / "minimal_workspace"
+NOTEBOOK = WORKSPACE / "notebooks" / "onboarding_success"
+SNAPSHOT = NOTEBOOK / "__marimo__" / "session" / "onboarding_success.py.json"
+
+
+def _locked_version(package: str) -> str:
+    lock = tomllib.loads((WORKSPACE / "uv.lock").read_text())
+    return next(p["version"] for p in lock["package"] if p["name"] == package)
+
+
+class SessionSnapshotTests(unittest.TestCase):
+    def test_snapshot_exists_and_is_valid_json(self) -> None:
+        self.assertTrue(SNAPSHOT.is_file(), "session snapshot should ship with the scaffold")
+        json.loads(SNAPSHOT.read_text())  # raises on invalid JSON
+
+    def test_snapshot_marimo_version_matches_lock(self) -> None:
+        # A mismatch silently disables the cache on deploy (key includes the version).
+        snap = json.loads(SNAPSHOT.read_text())
+        self.assertEqual(snap["metadata"]["marimo_version"], _locked_version("marimo"))
+
+    def test_snapshot_has_no_inlined_anywidget_module(self) -> None:
+        # A cached anywidget ESM becomes a data: URL the frontend refuses to load.
+        self.assertNotIn("data:text/javascript", SNAPSHOT.read_text())
+
+
+class NotebookWiringTests(unittest.TestCase):
+    def test_deployment_deploys_the_notebook(self) -> None:
+        self.assertIn("onboarding_success", (WORKSPACE / "__deployment__.py").read_text())
+
+    def test_shipped_python_files_compile(self) -> None:
+        files = [*NOTEBOOK.rglob("*.py"), *(WORKSPACE / ".scripts").glob("*.py")]
+        self.assertTrue(files)
+        for path in files:
+            py_compile.compile(str(path), doraise=True)
+
+
+class ScaffoldShipsNotebookTests(unittest.TestCase):
+    def test_copy_scaffold_ships_notebook_and_scripts(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            ws = Path(tmp) / "ws"
+            copy_scaffold(ws, scaffold="minimal_workspace", agent="claude")
+            for rel in (
+                "notebooks/onboarding_success/onboarding_success.py",
+                "notebooks/onboarding_success/__marimo__/session/onboarding_success.py.json",
+                ".scripts/serve_headless.py",
+                ".scripts/show_notebook.py",
+            ):
+                self.assertTrue((ws / rel).exists(), f"{rel!r} should be scaffolded")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_workspace_scripts.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import io
+import os
+import re
+import runpy
+import sys
+import tempfile
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+from unittest import mock
+
+from create_dlthub_workspace.scaffold import SCAFFOLDS_DIR
+
+SCRIPTS = SCAFFOLDS_DIR / "minimal_workspace" / ".scripts"
+
+
+def _run(script: Path, argv: list[str], app_url: str | None = None):
+    """Return (exit_code, stdout, exit_message, browser_mock)."""
+    env = dict(os.environ)
+    if app_url is None:
+        env.pop("DLTHUB_APP_URL", None)
+    else:
+        env["DLTHUB_APP_URL"] = app_url
+    out = io.StringIO()
+    code, msg = 0, None
+    with (
+        mock.patch.object(sys, "argv", argv),
+        mock.patch.dict(os.environ, env, clear=True),
+        mock.patch("webbrowser.open") as browser,
+        redirect_stdout(out),
+    ):
+        try:
+            runpy.run_path(str(script), run_name="__main__")
+        except SystemExit as exc:
+            if isinstance(exc.code, int) or exc.code is None:
+                code = exc.code or 0
+            else:
+                code, msg = 1, str(exc.code)
+    return code, out.getvalue(), msg, browser
+
+
+def _workspace_with_config(tmp: Path, body: str) -> Path:
+    (tmp / ".dlt").mkdir(parents=True)
+    (tmp / ".dlt" / "config.toml").write_text(body)
+    (tmp / ".scripts").mkdir()
+    for name in ("show_notebook.py", "serve_headless.py"):
+        (tmp / ".scripts" / name).write_text((SCRIPTS / name).read_text())
+    return tmp
+
+
+class ScriptsRequireRefTests(unittest.TestCase):
+    def test_show_notebook_requires_ref(self) -> None:
+        code, _, msg, browser = _run(SCRIPTS / "show_notebook.py", ["show_notebook.py"])
+        self.assertEqual(code, 1)
+        self.assertIn("usage", (msg or "").lower())
+        browser.assert_not_called()
+
+    def test_serve_headless_requires_ref(self) -> None:
+        # Arg-check runs before `import dlt_runtime`, so this works without it installed.
+        code, _, msg, _ = _run(SCRIPTS / "serve_headless.py", ["serve_headless.py"])
+        self.assertEqual(code, 1)
+        self.assertIn("usage", (msg or "").lower())
+
+
+class ExampleJobRefTests(unittest.TestCase):
+    def test_scripts_reference_a_deployed_job(self) -> None:
+        deployment = (SCAFFOLDS_DIR / "minimal_workspace" / "__deployment__.py").read_text()
+        for script in SCRIPTS.glob("*.py"):
+            for name in set(re.findall(r"jobs\.([a-z_]+)", script.read_text())):
+                self.assertIn(
+                    name,
+                    deployment,
+                    f"{script.name} references jobs.{name}, not deployed in __deployment__.py",
+                )
+
+
+class ShowNotebookUrlTests(unittest.TestCase):
+    CONFIG = '[runtime]\nworkspace_id = "ws-123"\n'
+
+    def test_builds_show_url_and_opens_browser(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            ws = _workspace_with_config(Path(d), self.CONFIG)
+            code, out, _, browser = _run(
+                ws / ".scripts" / "show_notebook.py", ["show_notebook.py", "jobs.onboarding_success"]
+            )
+        url = "https://app.dlthub.com/w/ws-123/notebooks/jobs.onboarding_success/show"
+        self.assertEqual(code, 0)
+        self.assertIn(url, out)
+        browser.assert_called_once_with(url)
+
+    def test_respects_app_url_override(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            ws = _workspace_with_config(Path(d), self.CONFIG)
+            code, out, _, browser = _run(
+                ws / ".scripts" / "show_notebook.py",
+                ["show_notebook.py", "jobs.onboarding_success"],
+                app_url="https://app.dlthub.test/",
+            )
+        self.assertEqual(code, 0)
+        self.assertIn("https://app.dlthub.test/w/ws-123/", out)
+
+    def test_errors_when_workspace_id_missing(self) -> None:
+        with tempfile.TemporaryDirectory() as d:
+            ws = _workspace_with_config(Path(d), '[runtime]\nlog_level = "INFO"\n')
+            code, _, msg, browser = _run(
+                ws / ".scripts" / "show_notebook.py", ["show_notebook.py", "jobs.onboarding_success"]
+            )
+        self.assertEqual(code, 1)
+        self.assertIn("workspace_id", msg or "")
+        browser.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()