add tests for StatusHistory

.gitignore

@@ -8,6 +8,7 @@ tmp/
 IGNORE-ME*
 .pyre/*
 .draft
+.coverage
 
 # local env files
 .env*.local

pyproject.toml

@@ -25,6 +25,8 @@ dependencies=[
   "pre-commit",
   "cryptography",
   "httpx",
+  "pytest",
+  "pytest-cov",
 ]
 
 [project.scripts]
@@ -50,6 +52,15 @@ disable = [
   # "R0913",
 ]
 
+[tool.pylint.main]
+source-roots = ["sidecar"]
+
 [tool.black]
 target-version = ["py311", ]
 include = '\.pyi?$'
+
+[tool.pytest.ini_options]
+addopts = [
+    "--import-mode=importlib",
+]
+pythonpath = "sidecar"

sidecar/app/query/status.py

@@ -11,13 +11,13 @@
 
 class Status(IntEnum):
     UNKNOWN = auto()
+    NOT_FOUND = auto()
     STARTING = auto()
     COMPILING = auto()
     WAITING_TO_START = auto()
     IN_PROGRESS = auto()
     COMPLETE = auto()
     KILLED = auto()
-    NOT_FOUND = auto()
     CRASHED = auto()
 
 
@@ -29,7 +29,7 @@ class Status(IntEnum):
 @dataclass
 class StatusHistory:
     file_path: Path = field(init=True, repr=False)
-    logger: loguru.Logger = field(init=True, repr=False)
+    logger: loguru.Logger = field(init=True, repr=False, compare=False)
     _status_history: list[StatusChangeEvent] = field(
         init=False, default_factory=list, repr=True
     )
@@ -46,7 +46,14 @@ def __post_init__(self):
                         )
                     )
 
+    @property
+    def locking_status(self):
+        """Cannot add to history after this or higher status is reached"""
+        return Status.COMPLETE
+
     def add(self, status: Status, timestamp: float = time.time()):
+        assert status > self.current_status
+        assert self.current_status < self.locking_status
         self._status_history.append(
             StatusChangeEvent(status=status, timestamp=timestamp)
         )
@@ -55,22 +62,22 @@ def add(self, status: Status, timestamp: float = time.time()):
             f.write(f"{status.name},{timestamp}\n")
 
     @property
-    def last_status_event(self):
+    def current_status_event(self):
         if not self._status_history:
             return StatusChangeEvent(status=Status.UNKNOWN, timestamp=time.time())
         return self._status_history[-1]
 
     @property
     def current_status(self):
-        return self.last_status_event.status
+        return self.current_status_event.status
 
     @property
     def status_event_json(self):
         status_event = {
-            "status": self.last_status_event.status.name,
-            "start_time": self.last_status_event.timestamp,
+            "status": self.current_status_event.status.name,
+            "start_time": self.current_status_event.timestamp,
         }
         if self.current_status >= Status.COMPLETE and len(self._status_history) >= 2:
             status_event["start_time"] = self._status_history[-2].timestamp
-            status_event["end_time"] = self.last_status_event.timestamp
+            status_event["end_time"] = self.current_status_event.timestamp
         return status_event

sidecar/tests/app/query/test_status.py

@@ -0,0 +1,114 @@
+import time
+from pathlib import Path
+
+import loguru
+import pytest
+from app.query.status import Status, StatusChangeEvent, StatusHistory
+
+
+@pytest.fixture(name="status_history_fixture")
+def _status_history_fixture(tmp_path):
+    status_history = StatusHistory(
+        file_path=tmp_path / Path("status"),
+        logger=loguru.logger,
+    )
+
+    return status_history
+
+
+@pytest.fixture(name="full_status_history_fixture")
+def _full_status_history_fixture(status_history_fixture):
+    status_events = [
+        (Status.STARTING, 1.0),
+        (Status.COMPILING, 2.0),
+        (Status.WAITING_TO_START, 3.0),
+        (Status.IN_PROGRESS, 4.0),
+        (Status.COMPLETE, 5.0),
+    ]
+
+    for status, timestamp in status_events:
+        status_history_fixture.add(status, timestamp)
+
+    return status_history_fixture
+
+
+def test_status_history_add(status_history_fixture):
+    now = time.time()
+    status_history_fixture.add(Status.COMPILING, now)
+    assert status_history_fixture.current_status_event == StatusChangeEvent(
+        Status.COMPILING, now
+    )
+    now = time.time()
+    status_history_fixture.add(Status.IN_PROGRESS, now)
+    assert status_history_fixture.current_status_event == StatusChangeEvent(
+        Status.IN_PROGRESS, now
+    )
+
+
+def test_status_history_add_write_to_file(status_history_fixture):
+    status_history_fixture.add(Status.COMPILING, 1.0)
+    status_history_fixture.add(Status.IN_PROGRESS, 2.0)
+    with status_history_fixture.file_path.open("r", encoding="utf-8") as f:
+        assert f.readline() == "COMPILING,1.0\n"
+        assert f.readline() == "IN_PROGRESS,2.0\n"
+
+
+def test_status_history_add_load_from_file(tmp_path, full_status_history_fixture):
+    status_history = StatusHistory(
+        file_path=tmp_path / Path("status"),
+        logger=loguru.logger,
+    )
+    assert status_history == full_status_history_fixture
+
+
+def test_status_history_cannot_add_when_locked(full_status_history_fixture):
+    with pytest.raises(AssertionError):
+        now = time.time()
+        full_status_history_fixture.add(Status.KILLED, now)
+
+
+def test_status_history_cannot_add_lower_status(status_history_fixture):
+    now = time.time()
+    status_history_fixture.add(Status.IN_PROGRESS, now)
+    assert status_history_fixture.current_status_event == StatusChangeEvent(
+        Status.IN_PROGRESS, now
+    )
+    with pytest.raises(AssertionError):
+        now = time.time()
+        status_history_fixture.add(Status.COMPILING, now)
+
+
+def test_status_history_current_status_event(full_status_history_fixture):
+    assert full_status_history_fixture.current_status_event == StatusChangeEvent(
+        Status.COMPLETE, 5.0
+    )
+
+
+def test_status_history_current_status(full_status_history_fixture):
+    assert full_status_history_fixture.current_status == Status.COMPLETE
+
+
+def test_status_history_status_event_json(
+    status_history_fixture,
+):
+    now = time.time()
+    status_history_fixture.add(Status.COMPILING, now)
+    assert status_history_fixture.status_event_json == {
+        "status": Status.COMPILING.name,
+        "start_time": now,
+    }
+
+    now = time.time()
+    status_history_fixture.add(Status.IN_PROGRESS, now)
+    assert status_history_fixture.status_event_json == {
+        "status": Status.IN_PROGRESS.name,
+        "start_time": now,
+    }
+
+    now2 = time.time()
+    status_history_fixture.add(Status.COMPLETE, now2)
+    assert status_history_fixture.status_event_json == {
+        "status": Status.COMPLETE.name,
+        "start_time": now,
+        "end_time": now2,
+    }

sidecar/tests/demo_test.py

@@ -0,0 +1,6 @@
+def func(x):
+    return x + 1
+
+
+def test_answer():
+    assert func(3) == 4