general: minor random cleanup and add a test

src/dron/dron.py

@@ -5,11 +5,12 @@
 from collections import OrderedDict
 from collections.abc import Iterable, Iterator
 from concurrent.futures import ProcessPoolExecutor
+from dataclasses import dataclass
 from difflib import unified_diff
 from itertools import tee
 from pathlib import Path
 from subprocess import check_call
-from typing import NamedTuple
+from typing import assert_never
 
 import click
 
@@ -93,6 +94,14 @@ def make_state(jobs: Iterable[Job]) -> State:
             pre_units.append((uname + '.service', s))
 
             when = j.when
+            # NOTE: both None and ALWAYS currently compile to a timerless service.
+            # apply_state then treats timerless services as always-running, so these
+            # are effectively equivalent on systemd for now.
+            #
+            # This likely needs an explicit internal job mode once there is enough
+            # real ALWAYS usage to decide install-target semantics. In particular,
+            # desktop/session services may want graphical-session.target + PartOf
+            # rather than default.target.
             if when is None:
                 # manual job?
                 continue
@@ -115,7 +124,8 @@ def make_state(jobs: Iterable[Job]) -> State:
 
 
 # TODO bleh. too verbose..
-class Update(NamedTuple):
+@dataclass(frozen=True)
+class Update:
     unit_file: UnitFile
     old_body: Body
     new_body: Body
@@ -125,15 +135,17 @@ def unit(self) -> str:
         return self.unit_file.name
 
 
-class Delete(NamedTuple):
+@dataclass(frozen=True)
+class Delete:
     unit_file: UnitFile
 
     @property
     def unit(self) -> str:
         return self.unit_file.name
 
 
-class Add(NamedTuple):
+@dataclass(frozen=True)
+class Add:
     unit_file: UnitFile
     body: Body
 
@@ -142,8 +154,8 @@ def unit(self) -> str:
         return self.unit_file.name
 
 
-Action = Update | Delete | Add
-Plan = Iterable[Action]
+type Action = Update | Delete | Add
+type Plan = Iterable[Action]
 
 # TODO ugh. not sure how to verify them?
 
@@ -182,7 +194,7 @@ def is_always_running(unit_path: Path) -> bool:
         # TODO meh. not ideal
         return not has_timer
 
-    plan = list(compute_plan(current=current, pending=pending))
+    plan: list[Action] = list(compute_plan(current=current, pending=pending))
 
     deletes: list[Delete] = []
     adds: list[Add] = []
@@ -196,7 +208,7 @@ def is_always_running(unit_path: Path) -> bool:
         elif isinstance(a, Update):
             _updates.append(a)
         else:
-            raise TypeError("Can't happen", a)
+            assert_never(a)
 
     if len(deletes) == len(current) and len(deletes) > 0:
         msg = "Trying to delete all managed jobs"
@@ -205,12 +217,11 @@ def is_always_running(unit_path: Path) -> bool:
         else:
             raise RuntimeError(msg)
 
-    Diff = list[str]
+    type Diff = list[str]
     nochange: list[Update] = []
     updates: list[tuple[Update, Diff]] = []
 
     for u in _updates:
-        unit = a.unit
         diff: Diff = list(
             unified_diff(
                 u.old_body.splitlines(keepends=True),

src/dron/tests/test_dron.py

@@ -6,7 +6,8 @@
 
 import pytest
 
-from ..dron import do_lint, load_jobs
+from ..common import UnitState
+from ..dron import Add, Delete, Update, compute_plan, do_lint, load_jobs
 
 
 @pytest.fixture
@@ -85,6 +86,54 @@ def jobs() -> Iterator[Job]:
         _loaded = list(load_jobs(tab_module='test_drontab'))
 
 
+def test_compute_plan() -> None:
+    def unit(name: str, body: str) -> UnitState:
+        return UnitState(unit_file=Path('/units') / name, body=body, cmdline=None)
+
+    # fmt: off
+    unchanged_current = unit('unchanged.service', 'same')
+    changed_current   = unit('changed.service'  , 'old')
+    deleted_current   = unit('deleted.service'  , 'deleted')
+
+    unchanged_pending = unit('unchanged.service', 'same')
+    changed_pending   = unit('changed.service'  , 'new')
+    added_pending     = unit('added.service'    , 'added')
+    # fmt: on
+
+    plan = list(
+        compute_plan(
+            current=[
+                unchanged_current,
+                changed_current,
+                deleted_current,
+            ],
+            pending=[
+                unchanged_pending,
+                changed_pending,
+                added_pending,
+            ],
+        )
+    )
+
+    assert plan == [
+        Delete(unit_file=deleted_current.unit_file),
+        Update(
+            unit_file=unchanged_current.unit_file,
+            old_body='same',
+            new_body='same',
+        ),
+        Update(
+            unit_file=changed_current.unit_file,
+            old_body='old',
+            new_body='new',
+        ),
+        Add(
+            unit_file=added_pending.unit_file,
+            body='added',
+        ),
+    ]
+
+
 def test_jobs_auto_naming(tmp_pythonpath: Path) -> None:
     tpath = Path(tmp_pythonpath) / 'test_drontab.py'
     tpath.write_text(