general: add some magic for guessing job names based on variable it b…

…inds to + tests

src/dron/api.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
 import getpass
+import inspect
+import re
 import sys
-from typing import NamedTuple, Sequence
+from dataclasses import dataclass
+from typing import Sequence
 
 from .common import (
     IS_SYSTEMD,
@@ -15,7 +18,8 @@
 OnFailureAction = str
 
 
-class Job(NamedTuple):
+@dataclass
+class Job:
     when: When | None
     command: Command
     unit_name: str
@@ -41,14 +45,45 @@ def email(to: str) -> str:
     telegram = f'{sys.executable} -m dron.notify.telegram --job %n'
 
 
-def job(when: When | None, command: Command, *, unit_name: str, on_failure: Sequence[OnFailureAction]=(notify.email_local,), **kwargs) -> Job:
-    assert 'extra_email' not in kwargs, unit_name  # deprecated
-
+def job(
+    when: When | None,
+    command: Command,
+    *,
+    unit_name: str | None = None,
+    on_failure: Sequence[OnFailureAction] = (notify.email_local,),
+    **kwargs,
+) -> Job:
     """
     when: if None, then timer won't be created (still allows running job manually)
+    unit_name: if None, then will attempt to guess from source code (experimental!)
     """
-    # TODO later, autogenerate unit name
-    # I guess warn user about non-unique names and prompt to give a more specific name?
+    assert 'extra_email' not in kwargs, unit_name  # deprecated
+
+    stacklevel: int = kwargs.pop('stacklevel', 1)
+
+    def guess_name() -> str | Exception:
+        stack = inspect.stack()
+        frame = stack[stacklevel + 1]  # +1 for guess_name itself
+        code_context_lines = frame.code_context
+        # python should alway keep single line for code context? but just in case
+        if len(code_context_lines) != 1:
+            return RuntimeError(f"Expected single code context line, got {code_context_lines=}")
+        [code_context] = code_context_lines
+        code_context = code_context.strip()
+        rgx = r'(\w+)\s+='
+        m = re.match(rgx, code_context)  # find assignment to variable
+        if m is None:
+            return RuntimeError(f"Couldn't guess from {code_context=} (regex {rgx=})")
+        return m.group(1)
+
+    if unit_name is None:
+        guessed_name = guess_name()
+
+        if isinstance(guessed_name, Exception):
+            raise RuntimeError(f"{when} {command}: couldn't guess job name: {guessed_name}")
+
+        unit_name = guessed_name
+
     return Job(
         when=when,
         command=command,
@@ -62,9 +97,9 @@ def job(when: When | None, command: Command, *, unit_name: str, on_failure: Sequ
     'When',
     'OnCalendar',
     'OnFailureAction',
-    'Command', 'wrap',
+    'Command',
+    'wrap',
     'job',
     'notify',
-
     'Job',  # todo maybe don't expose it?
 )

src/dron/dron.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+import importlib.util
 import os
 import shlex
 import shutil
@@ -12,7 +13,7 @@
 from pprint import pprint
 from subprocess import check_call, run
 from tempfile import TemporaryDirectory
-from typing import Any, Iterable, Iterator, NamedTuple, Union
+from typing import Iterable, Iterator, NamedTuple, Union
 
 import click
 
@@ -428,7 +429,7 @@ def test_do_lint(tmp_path: Path) -> None:
 
 
     def ok(body: str) -> None:
-        tpath = Path(tmp_path) / 'drontab'
+        tpath = Path(tmp_path) / 'drontab.py'
         tpath.write_text(body)
         do_lint(tabfile=tpath)
 
@@ -485,21 +486,23 @@ def drontab_dir() -> str:
 
 
 def load_jobs(tabfile: Path, ppath: Path) -> Iterator[Job]:
-    globs: dict[str, Any] = {}
-
-    # TODO also need to modify pythonpath here??? ugh!
-
     pp = str(ppath)
     sys.path.insert(0, pp)
     try:
-        exec(tabfile.read_text(), globs)
+        spec = importlib.util.spec_from_file_location(tabfile.name, tabfile)
+        assert spec is not None, tabfile
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
     finally:
         sys.path.remove(pp)  # extremely meh..
 
-    jobs = globs['jobs']
+    jobs = module.jobs
+    emitted: dict[str, Job] = {}
     for job in jobs():
         assert isinstance(job, Job), job  # just in case for dumb typos
+        assert job.unit_name not in emitted, (job, emitted[job.unit_name])
         yield job
+        emitted[job.unit_name] = job
 
 
 def apply(tabfile: Path) -> None:

src/dron/tests/test_dron.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from ..dron import load_jobs
+
+
+def test_load_jobs_basic(tmp_path: Path) -> None:
+    tpath = Path(tmp_path) / 'drontab.py'
+    tpath.write_text(
+        '''
+from typing import Iterator
+
+from dron.api import job, Job
+
+
+def jobs() -> Iterator[Job]:
+    job3 = job(
+        '03:10',
+        ['/path/to/command.py', 'some', 'args', '3'],
+        unit_name='job3',
+    )
+    job1 = job(
+        '01:10',
+        ['/path/to/command.py', 'some', 'args', '1'],
+        unit_name='job1',
+    )
+    yield job1
+    yield job(
+        '02:10',
+        ['/path/to/command.py', 'some', 'args', '2'],
+        unit_name='job2',
+    )
+    yield job3
+
+'''
+    )
+    loaded = list(load_jobs(tabfile=tpath, ppath=tmp_path))
+    [job1, job2, job3] = loaded
+
+    assert job1.when == '01:10'
+    assert job1.command == ['/path/to/command.py', 'some', 'args', '1']
+    assert job1.unit_name == 'job1'
+
+    assert job2.when == '02:10'
+    assert job2.command == ['/path/to/command.py', 'some', 'args', '2']
+    assert job2.unit_name == 'job2'
+
+    assert job3.when == '03:10'
+    assert job3.command == ['/path/to/command.py', 'some', 'args', '3']
+    assert job3.unit_name == 'job3'
+
+
+def test_load_jobs_dupes(tmp_path: Path) -> None:
+    tpath = Path(tmp_path) / 'drontab.py'
+    tpath.write_text(
+        '''
+from typing import Iterator
+
+from dron.api import job, Job
+
+def jobs() -> Iterator[Job]:
+    yield job('00:00', 'echo', unit_name='job3')
+    yield job('00:00', 'echo', unit_name='job1')
+    # whoops! duplicate job name
+    yield job('00:00', 'echo', unit_name='job3')
+'''
+    )
+    with pytest.raises(AssertionError):
+        _loaded = list(load_jobs(tabfile=tpath, ppath=tmp_path))
+
+
+def test_jobs_auto_naming(tmp_path: Path) -> None:
+    tpath = Path(tmp_path) / 'drontab.py'
+    tpath.write_text(
+        '''
+from typing import Iterator
+
+from dron.api import job, Job
+
+
+job2 = job(
+    '00:02',
+    'echo',
+)
+
+
+def job_maker(when) -> Job:
+    return job(when, 'echo job maker', stacklevel=2)
+
+
+def jobs() -> Iterator[Job]:
+    job_1 = job('00:01',
+        'echo',
+    )
+    yield job2
+    yield job('00:00', 'echo', unit_name='job_named')
+    yield job_1
+    job4 = \
+       job('00:04', 'echo')
+    job5     = job_maker('00:05')
+    yield job5
+    yield job4
+'''
+    )
+    loaded = list(load_jobs(tabfile=tpath, ppath=tmp_path))
+    (job2, job_named, job_1, job5, job4) = loaded
+    assert job_1.unit_name == 'job_1'
+    assert job_1.when == '00:01'
+    assert job2.unit_name == 'job2'
+    assert job2.when == '00:02'
+    assert job_named.unit_name == 'job_named'
+    assert job_named.when == '00:00'
+    assert job4.unit_name == 'job4'
+    assert job4.when == '00:04'
+    assert job5.unit_name == 'job5'
+    assert job5.when == '00:05'