feat(runtime): Adding state to track runtime

devservices/utils/state.py

@@ -3,14 +3,21 @@
 import os
 import sqlite3
 from enum import Enum
+from typing import Literal
 
 from devservices.constants import DEVSERVICES_LOCAL_DIR
 from devservices.constants import STATE_DB_FILE
 
 
+class ServiceRuntime(Enum):
+    LOCAL = "local"
+    CONTAINERIZED = "containerized"
+
+
 class StateTables(Enum):
     STARTED_SERVICES = "started_services"
     STARTING_SERVICES = "starting_services"
+    SERVICE_RUNTIME = "service_runtime"
 
 
 class State:
@@ -30,7 +37,7 @@ def __new__(cls) -> State:
 
     def initialize_database(self) -> None:
         cursor = self.conn.cursor()
-        # Formatted strings here and throughout the fileshould be extremely low risk given these are constants
+        # Formatted strings here and throughout the file should be extremely low risk given these are constants
         cursor.execute(
             f"""
             CREATE TABLE IF NOT EXISTS {StateTables.STARTED_SERVICES.value} (
@@ -50,10 +57,26 @@ def initialize_database(self) -> None:
             )
         """
         )
+
+        cursor.execute(
+            f"""
+            CREATE TABLE IF NOT EXISTS {StateTables.SERVICE_RUNTIME.value} (
+                service_name TEXT PRIMARY KEY,
+                runtime TEXT
+                timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+        """
+        )
         self.conn.commit()
 
     def update_service_entry(
-        self, service_name: str, mode: str, table: StateTables
+        self,
+        service_name: str,
+        mode: str,
+        table: (
+            Literal[StateTables.STARTED_SERVICES]
+            | Literal[StateTables.STARTING_SERVICES]
+        ),
     ) -> None:
         cursor = self.conn.cursor()
         service_entries = self.get_service_entries(table)
@@ -96,7 +119,12 @@ def get_service_entries(self, table: StateTables) -> list[str]:
         return [row[0] for row in cursor.fetchall()]
 
     def get_active_modes_for_service(
-        self, service_name: str, table: StateTables
+        self,
+        service_name: str,
+        table: (
+            Literal[StateTables.STARTED_SERVICES]
+            | Literal[StateTables.STARTING_SERVICES]
+        ),
     ) -> list[str]:
         cursor = self.conn.cursor()
         cursor.execute(
@@ -110,6 +138,41 @@ def get_active_modes_for_service(
             return []
         return str(result[0]).split(",")
 
+    def get_service_runtime(self, service_name: str) -> ServiceRuntime:
+        cursor = self.conn.cursor()
+        cursor.execute(
+            f"""
+            SELECT runtime FROM {StateTables.SERVICE_RUNTIME.value} WHERE service_name = ?
+        """,
+            (service_name,),
+        )
+        result = cursor.fetchone()
+        if result is None:
+            return ServiceRuntime.CONTAINERIZED
+        return ServiceRuntime(result[0])
+
+    def update_service_runtime(
+        self, service_name: str, runtime: ServiceRuntime
+    ) -> None:
+        cursor = self.conn.cursor()
+        cursor.execute(
+            f"""
+            INSERT OR REPLACE INTO {StateTables.SERVICE_RUNTIME.value} (service_name, runtime) VALUES (?, ?)
+        """,
+            (service_name, runtime.value),
+        )
+        self.conn.commit()
+
+    def get_services_by_runtime(self, runtime: ServiceRuntime) -> list[str]:
+        cursor = self.conn.cursor()
+        cursor.execute(
+            f"""
+            SELECT service_name FROM {StateTables.SERVICE_RUNTIME.value} WHERE runtime = ?
+        """,
+            (runtime.value,),
+        )
+        return [row[0] for row in cursor.fetchall()]
+
     def clear_state(self) -> None:
         cursor = self.conn.cursor()
         cursor.execute(
@@ -122,4 +185,9 @@ def clear_state(self) -> None:
             DELETE FROM {StateTables.STARTING_SERVICES.value}
         """
         )
+        cursor.execute(
+            f"""
+            DELETE FROM {StateTables.SERVICE_RUNTIME.value}
+        """
+        )
         self.conn.commit()

tests/utils/test_state.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 from unittest import mock
 
+from devservices.utils.state import ServiceRuntime
 from devservices.utils.state import State
 from devservices.utils.state import StateTables
 
@@ -82,3 +83,78 @@ def test_get_mode_for_nonexistent_service(tmp_path: Path) -> None:
             )
             == []
         )
+
+
+def test_get_and_update_service_runtime(tmp_path: Path) -> None:
+    with mock.patch("devservices.utils.state.STATE_DB_FILE", str(tmp_path / "state")):
+        state = State()
+        state.update_service_runtime("example-service", ServiceRuntime.CONTAINERIZED)
+        assert (
+            state.get_service_runtime("example-service") == ServiceRuntime.CONTAINERIZED
+        )
+        state.update_service_runtime("example-service", ServiceRuntime.LOCAL)
+        assert state.get_service_runtime("example-service") == ServiceRuntime.LOCAL
+        state.update_service_runtime("example-service", ServiceRuntime.CONTAINERIZED)
+        assert (
+            state.get_service_runtime("example-service") == ServiceRuntime.CONTAINERIZED
+        )
+
+
+def test_get_service_runtime_defaults_to_containerized(tmp_path: Path) -> None:
+    with mock.patch("devservices.utils.state.STATE_DB_FILE", str(tmp_path / "state")):
+        state = State()
+        assert (
+            state.get_service_runtime("unknown-service") == ServiceRuntime.CONTAINERIZED
+        )
+
+
+def test_get_services_by_runtime(tmp_path: Path) -> None:
+    with mock.patch("devservices.utils.state.STATE_DB_FILE", str(tmp_path / "state")):
+        state = State()
+        assert state.get_services_by_runtime(ServiceRuntime.CONTAINERIZED) == []
+        assert state.get_services_by_runtime(ServiceRuntime.LOCAL) == []
+        state.update_service_runtime("first-service", ServiceRuntime.CONTAINERIZED)
+        state.update_service_runtime("second-service", ServiceRuntime.LOCAL)
+        state.update_service_runtime("third-service", ServiceRuntime.CONTAINERIZED)
+        assert state.get_services_by_runtime(ServiceRuntime.CONTAINERIZED) == [
+            "first-service",
+            "third-service",
+        ]
+        assert state.get_services_by_runtime(ServiceRuntime.LOCAL) == ["second-service"]
+
+        state.update_service_runtime("first-service", ServiceRuntime.LOCAL)
+        assert state.get_services_by_runtime(ServiceRuntime.CONTAINERIZED) == [
+            "third-service"
+        ]
+        assert state.get_services_by_runtime(ServiceRuntime.LOCAL) == [
+            "second-service",
+            "first-service",
+        ]
+
+
+def test_clear_state(tmp_path: Path) -> None:
+    with mock.patch("devservices.utils.state.STATE_DB_FILE", str(tmp_path / "state")):
+        state = State()
+        state.update_service_entry(
+            "first-service", "default", StateTables.STARTED_SERVICES
+        )
+        state.update_service_entry(
+            "second-service", "default", StateTables.STARTING_SERVICES
+        )
+        state.update_service_runtime("first-service", ServiceRuntime.CONTAINERIZED)
+        state.update_service_runtime("second-service", ServiceRuntime.LOCAL)
+        assert state.get_service_entries(StateTables.STARTED_SERVICES) == [
+            "first-service"
+        ]
+        assert state.get_service_entries(StateTables.STARTING_SERVICES) == [
+            "second-service"
+        ]
+        assert state.get_services_by_runtime(ServiceRuntime.CONTAINERIZED) == [
+            "first-service"
+        ]
+        assert state.get_services_by_runtime(ServiceRuntime.LOCAL) == ["second-service"]
+        state.clear_state()
+        assert state.get_service_entries(StateTables.STARTED_SERVICES) == []
+        assert state.get_service_entries(StateTables.STARTING_SERVICES) == []
+        assert state.get_services_by_runtime(ServiceRuntime.CONTAINERIZED) == []
+        assert state.get_services_by_runtime(ServiceRuntime.LOCAL) == []