#203 Create initial Playwright tests

.gitignore

@@ -28,6 +28,9 @@ coverage/
 cypress/videos/
 cypress/screenshots/
 
+# Playwright
+test-results/
+
 # Data files
 *.db
 *.db-journal

paramview/_cli.py

@@ -9,7 +9,7 @@
 _VERSION = distribution(_PACKAGE_NAME).version
 
 
-def _parse_args(*args: str) -> Namespace:
+def _parse_args(args: list[str] | None = None) -> Namespace:
     """
     Parse command line arguments using argparse. If arguments are passed in, those are
     parsed instead of command line arguments.
@@ -33,7 +33,12 @@ def _parse_args(*args: str) -> Namespace:
         type=int,
         help="port to use (default is 5050)",
     )
-    return parser.parse_args(args or None)
+    parser.add_argument(
+        "--no-open",
+        action="store_true",
+        help="don't open a new browser window (default is to open one)",
+    )
+    return parser.parse_args(args)
 
 
 def main() -> None:
@@ -42,4 +47,4 @@ def main() -> None:
     program calls this function.
     """
     args = _parse_args()
-    start_server(args.db_path, default_port=args.port)
+    start_server(args.db_path, default_port=args.port, open_window=not args.no_open)

paramview/_server.py

@@ -33,21 +33,21 @@ def _available_port(host: str, default_port: int) -> int:
 
 def start_server(
     db_path: str,
-    host: str = "127.0.0.1",
+    host: str = "localhost",
     default_port: int = 5050,
-    auto_open: bool = True,
+    open_window: bool = True,
 ) -> None:
     """
     Start the server locally on the given port using SocketIO, and open in a new browser
-    window if auto_open is True. If the given port is in use, find another available
-    port.
+    window if ``open_window`` is ``True``. If the given port is in use, find another
+    available port.
     """
     port = _available_port(host, default_port)
     app, socketio = create_app(db_path)
     stop_watch_db = watch_db(db_path, socketio)
     try:
         print(f"Serving on http://{host}:{port}", file=sys.stderr)
-        if auto_open:
+        if open_window:
             webbrowser.open(f"http://{host}:{port}", new=2)
         socketio.run(app, host, port)
     finally:

pyproject.toml

@@ -26,7 +26,9 @@ flake8 = "^6.1.0"
 pylint = "^3.0.2"
 black = "^23.11.0"
 pytest = "^7.4.3"
+pytest-playwright = "^0.4.3"
 freezegun = "1.2.2"
+requests = "^2.31.0"
 sqlalchemy = "^2.0.23"
 astropy = "^6.0.0"
 
@@ -38,4 +40,5 @@ strict = true
 
 [tool.pytest.ini_options]
 addopts = ["--import-mode=importlib"]
+base_url = "http://localhost:5050"
 filterwarnings = ["ignore::DeprecationWarning:eventlet.support.greenlets"]

tests/e2e/conftest.py

@@ -0,0 +1,61 @@
+"""
+Setup and global fixtures for E2E tests.
+
+Called automatically by Pytest before running tests.
+"""
+
+from typing import Any
+from collections.abc import Generator
+import pytest
+from paramdb import ParamDB
+from tests.e2e.helpers import setup_db_and_start_server, clear, reset
+
+
+@pytest.fixture(name="db_name", scope="session")
+def fixture_db_name() -> str:
+    """Database file name."""
+    return "param.db"
+
+
+@pytest.fixture(name="db_path", scope="session")
+def fixture_db_path(
+    tmp_path_factory: pytest.TempPathFactory, base_url: str, db_name: str
+) -> Generator[str, None, None]:
+    """
+    Path to the ParamDB database. The ParamView server for the database is also started
+    and cleaned up by this fixture.
+    """
+    db_path = str(tmp_path_factory.mktemp("db") / db_name)
+    stop_server = setup_db_and_start_server(db_path, base_url)
+    yield db_path
+    stop_server()
+
+
+@pytest.fixture(name="db")
+def fixture_db(db_path: str) -> ParamDB[Any]:
+    """ParamDB database."""
+    return ParamDB[Any](db_path)
+
+
+@pytest.fixture(name="_clear_db")
+def fixture_clear_db(db: ParamDB[Any]) -> None:
+    """Clears the database."""
+    clear(db)
+
+
+@pytest.fixture(name="_reset_db")
+def fixture_reset_db(db: ParamDB[Any]) -> None:
+    """Resets the database."""
+    reset(db)
+
+
+@pytest.fixture(name="_reset_single_db")
+def fixture_reset_single_db(db: ParamDB[Any]) -> None:
+    """Resets the database to have a single commit."""
+    reset(db, num_commits=1)
+
+
+@pytest.fixture(name="_reset_long_db")
+def fixture_reset_long_db(db: ParamDB[Any]) -> None:
+    """Resets the database to have 100 commits."""
+    reset(db, num_commits=100)

tests/e2e/helpers.py

@@ -0,0 +1,122 @@
+"""Helper functions for E2E tests."""
+
+from __future__ import annotations
+from typing import Any
+from collections.abc import Callable
+import time
+import signal
+import subprocess
+from datetime import datetime, timedelta, timezone
+import requests  # type: ignore
+from sqlalchemy import delete
+from freezegun import freeze_time
+import astropy.units as u  # type: ignore
+from paramdb import ParamDB, Param, Struct, ParamList, ParamDict
+from paramdb._database import _Snapshot
+
+
+_SERVER_POLLING_WAIT = 0.1
+_SERVER_POLLING_MAX_RETRIES = int(5 / _SERVER_POLLING_WAIT)
+_SERVER_REQUEST_TIMEOUT = 1.0
+_START_DATE = datetime(2023, 1, 1, tzinfo=timezone.utc).astimezone()
+
+
+class _CustomParam(Param):
+    int: int
+    str: str
+
+
+class _CustomStruct(Struct):
+    int: int
+    str: str
+    param: _CustomParam
+
+
+def get_date(commit_id: int) -> datetime:
+    """Get the date corresponding to the given commit ID."""
+    return _START_DATE + timedelta(days=commit_id - 1)
+
+
+def clear(db: ParamDB[Any]) -> None:
+    """Clear the database."""
+    with db._Session.begin() as session:  # pylint: disable=no-member,protected-access
+        session.execute(delete(_Snapshot))  # Clear all commits
+
+
+def commit(
+    db: ParamDB[Any], message: str | None = None, data: Any | None = None
+) -> None:
+    """Make a commit with the given message."""
+    num_commits = db.num_commits
+    commit_id = num_commits + 1
+    message = f"Commit {commit_id}" if message is None else message
+    data = ParamDict(commit_id=commit_id, b=2, c=3) if data is None else data
+    with freeze_time(get_date(num_commits + 1)):
+        db.commit(message, data)
+
+
+def reset(db: ParamDB[Any], num_commits: int = 3) -> None:
+    """Clear the database and make some initial commits."""
+    clear(db)
+    initial_data = ParamDict(
+        {
+            "commit_id": 1,
+            "int": 123,
+            "float": 1.2345,
+            "bool": True,
+            "str": "test",
+            "None": None,
+            "datetime": get_date(1),
+            "Quantity": 1.2345 * u.m,
+            "list": [123, "test"],
+            "dict": {"int": 123, "str": "test"},
+            "paramList": ParamList([123, "test"]),
+            "paramDict": ParamDict(int=123, str="test"),
+            "struct": _CustomStruct(
+                int=123, str="test", param=_CustomParam(int=123, str="test")
+            ),
+            "param": _CustomParam(int=123, str="test"),
+        }
+    )
+    commit(db, "Initial commit", initial_data)
+    for _ in range(2, num_commits + 1):
+        commit(db)
+
+
+def setup_db_and_start_server(db_path: str, base_url: str) -> Callable[[], None]:
+    """
+    Set up the database, start the server, wait for the server to be up, and return a
+    function to stop the server.
+    """
+    # Verify that the base url is available
+    try:
+        requests.get(base_url, timeout=_SERVER_REQUEST_TIMEOUT)
+        raise RuntimeError(f"{base_url} is already in use.")
+    except requests.ConnectionError:
+        time.sleep(_SERVER_POLLING_WAIT)
+
+    reset(ParamDB(db_path))
+
+    # pylint: disable=consider-using-with
+    server_process = subprocess.Popen(["paramview", db_path, "--no-open"])
+
+    # Wait for server to be up
+    for _ in range(_SERVER_POLLING_MAX_RETRIES):
+        try:
+            requests.get(base_url, timeout=_SERVER_REQUEST_TIMEOUT)
+            break
+        except requests.ConnectionError:
+            time.sleep(_SERVER_POLLING_WAIT)
+
+    def stop_server() -> None:
+        server_process.send_signal(signal.SIGINT)
+
+    return stop_server
+
+
+def load_classes_from_db(db: ParamDB[Any]) -> None:
+    """
+    Load the last commit as Python classes. This helps to test that objects are
+    formatted properly, in particular datetime and Quantity objects.
+    """
+    db.load()

tests/e2e/test_page_title.py

@@ -0,0 +1,15 @@
+"""Tests for the page title."""
+
+from playwright.sync_api import Page, expect
+
+
+def test_title_is_db_name(_reset_db: None, page: Page, db_name: str) -> None:
+    """Page title is the database file name."""
+    page.goto("/")
+    expect(page).to_have_title(db_name)
+
+
+def test_title_is_error(_clear_db: None, page: Page) -> None:
+    """Page title is "Error" if an error has occurred."""
+    page.goto("/")
+    expect(page).to_have_title("Error")

tests/e2e/test_parameter_editing.py

@@ -0,0 +1,14 @@
+"""Tests for parameter editing."""
+
+from playwright.sync_api import Page, expect
+
+
+def test_displays_inputs(_reset_single_db: None, page: Page) -> None:
+    """Displays correct input for each parameter type."""
+    page.goto("/")
+    page.get_by_test_id("edit-button").click()
+    expect(
+        page.get_by_test_id("parameter-list-item-int")
+        .get_by_test_id("leaf-input")
+        .get_by_role("textbox")
+    ).to_have_value("123")

tests/conftest.py → tests/unit/conftest.py

@@ -1,4 +1,8 @@
-"""Defines global fixtures. Called automatically by Pytest before running tests."""
+"""
+Global fixtures for unit tests.
+
+Called automatically by Pytest before running tests.
+"""
 
 from __future__ import annotations
 from typing import Any
@@ -18,7 +22,7 @@ def fixture_db_name() -> str:
 
 
 @pytest.fixture(name="db_path")
-def fixture_db_path(db_name: str, tmp_path: Path) -> str:
+def fixture_db_path(tmp_path: Path, db_name: str) -> str:
     """
     Path to a ParamDB database. The initial database is created using a different path
     name to ensure that residual file system events do not trigger update events when

tests/test_cli.py → tests/unit/test_cli.py

@@ -9,7 +9,7 @@
 PROGRAM_NAME = "paramview"
 DB_PATH = "test.db"
 VERSION_MSG = f"{PROGRAM_NAME} {distribution(PROGRAM_NAME).version}"
-USAGE_MSG = f"usage: {PROGRAM_NAME} [-h] [-V] [-p PORT] <database path>"
+USAGE_MSG = f"usage: {PROGRAM_NAME} [-h] [-V] [-p PORT] [--no-open] <database path>"
 POSITIONAL_ARGS_MSG = """
 positional arguments:
   <database path>       path to the ParamDB database file
@@ -18,6 +18,7 @@
   -h, --help            show this help message and exit
   -V, --version         show program's version number and exit
   -p PORT, --port PORT  port to use (default is 5050)
+  --no-open             don't open a new browser window (default is to open one)
 """
 ERROR_MSG = f"{PROGRAM_NAME}: error:"
 REQUIRED_MSG = "the following arguments are required: <database path>"
@@ -34,28 +35,28 @@ def _sw(*strings: str) -> str:
 
 def test_db_path() -> None:
     """Parses the database path."""
-    args = _parse_args(DB_PATH)
+    args = _parse_args([DB_PATH])
     assert args.db_path == DB_PATH
 
 
 def test_port_default() -> None:
     """Default port is 5050."""
-    args = _parse_args(DB_PATH)
+    args = _parse_args([DB_PATH])
     assert args.port == 5050
 
 
 @pytest.mark.parametrize("port_arg", ["--port", "-p"])
 def test_port(port_arg: str) -> None:
     """Parses the port."""
-    args = _parse_args(DB_PATH, port_arg, "1234")
+    args = _parse_args([DB_PATH, port_arg, "1234"])
     assert args.port == 1234
 
 
 @pytest.mark.parametrize("version_arg", ["--version", "-V"])
 def test_version(version_arg: str, capsys: CaptureFixture[str]) -> None:
     """Prints version message to stdout and exists with code 0."""
     with pytest.raises(SystemExit) as exc_info:
-        _parse_args(version_arg)
+        _parse_args([version_arg])
     assert exc_info.value.code == 0
     assert _sw(capsys.readouterr().out) == _sw(VERSION_MSG)
 
@@ -64,7 +65,7 @@ def test_version(version_arg: str, capsys: CaptureFixture[str]) -> None:
 def test_help(help_arg: str, capsys: CaptureFixture[str]) -> None:
     """Prints help message to stdout and exists with code 0."""
     with pytest.raises(SystemExit) as exc_info:
-        _parse_args(help_arg)
+        _parse_args([help_arg])
     assert exc_info.value.code == 0
     help_message = _sw(capsys.readouterr().out)
     assert _sw(USAGE_MSG, POSITIONAL_ARGS_MSG) in help_message
@@ -74,7 +75,7 @@ def test_help(help_arg: str, capsys: CaptureFixture[str]) -> None:
 def test_no_args(capsys: CaptureFixture[str]) -> None:
     """Prints required argument message to stderr and exits with code 2."""
     with pytest.raises(SystemExit) as exc_info:
-        _parse_args()
+        _parse_args([])
     assert exc_info.value.code == 2
     assert _sw(capsys.readouterr().err) == _sw(USAGE_MSG, ERROR_MSG, REQUIRED_MSG)
 
@@ -85,7 +86,7 @@ def test_parse_args_unrecognized(
 ) -> None:
     """Prints message to stderr and exits with code 2."""
     with pytest.raises(SystemExit) as excinfo:
-        _parse_args(DB_PATH, unrecognized_arg)
+        _parse_args([DB_PATH, unrecognized_arg])
     assert excinfo.value.code == 2
     assert _sw(capsys.readouterr().err) == _sw(
         USAGE_MSG, ERROR_MSG, UNRECOGNIZED_MSG, unrecognized_arg