fix tests

Author: CamDavidsonPilon

core/pioreactor/actions/self_test.py

@@ -38,6 +38,7 @@
 from pioreactor.pubsub import Client
 from pioreactor.pubsub import post_into_leader
 from pioreactor.pubsub import prune_retained_messages
+from pioreactor.states import JobState
 from pioreactor.types import LedChannel
 from pioreactor.types import PdChannel
 from pioreactor.utils import is_pio_job_running
@@ -91,9 +92,9 @@ def test_REF_is_in_correct_position(managed_state, logger: CustomLogger, unit: s
                 signals[channel].append(reading.ods[channel].od)
 
             if i % 5 == 0 and i % 2 == 0:
-                st.set_state("ready")
+                st.set_state(JobState.READY)
             elif i % 5 == 0:
-                st.set_state("sleeping")
+                st.set_state(JobState.SLEEPING)
 
             if i == 25:
                 break

core/pioreactor/cli/pio.py

@@ -384,10 +384,11 @@ def job_history() -> None:
         click.echo(_format_job_history_line(*job))
 
 
-def _format_timestamp_to_seconds(timestamp: str) -> str:
-    """
-    Truncate timestamps like 2024-01-01T00:00:00.123456Z to second precision.
-    """
+def _format_timestamp_to_seconds(timestamp: str | None) -> str:
+
+    if timestamp is None:
+        return ""
+
     try:
         dt = to_datetime(timestamp)
     except ValueError:
@@ -451,7 +452,7 @@ def job_info(job_id: int | None, job_name: str | None) -> None:
     for setting, value, created_at, updated_at in settings:
         setting_label = click.style(setting, fg="cyan")
         created_at_display = _format_timestamp_to_seconds(created_at)
-        updated_at_display = _format_timestamp_to_seconds(updated_at) if updated_at is not None else ""
+        updated_at_display = _format_timestamp_to_seconds(updated_at)
 
         def _stringify(val: Any) -> str:
             if val is None:

core/tests/test_cli.py

@@ -61,7 +61,7 @@ def test_list_plugins() -> None:
     assert "my-example-plugin==0.2.0" in result.output
 
 
-@pytest.mark.skip(reason="not sure why this fails")
+@pytest.mark.flakey
 def test_pio_log() -> None:
     with collect_all_logs_of_level("DEBUG", whoami.get_unit_name(), whoami.UNIVERSAL_EXPERIMENT) as bucket:
         runner = CliRunner()
@@ -192,9 +192,9 @@ def test_pio_job_info_lists_job() -> None:
             jm.set_not_running(job_id)
 
 
-def test_pio_job_history_lists_job() -> None:
+def test_pio_job_list_lists_job() -> None:
     runner = CliRunner()
-    job_name = "test_job_history"
+    job_name = "test_job_list"
     unit = whoami.get_unit_name()
     experiment = whoami.UNIVERSAL_EXPERIMENT
 
@@ -210,7 +210,7 @@ def test_pio_job_history_lists_job() -> None:
         )
         jm.set_not_running(job_id)
 
-    result = runner.invoke(pio, ["jobs", "history"])
+    result = runner.invoke(pio, ["jobs", "list"])
 
     assert result.exit_code == 0
     matching_lines = [line for line in result.output.splitlines() if f"[job_id={job_id}]" in line]