feat: Add CI workflow for Pytest and implement unit tests for simulation and world functionalities

.github/workflows/pytest.yml

@@ -0,0 +1,28 @@
+name: Pytest CI
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+          cache: 'pip'
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+          pip install pytest
+
+      - name: Run tests
+        run: pytest -q

tests/test_simulation_service.py

@@ -0,0 +1,109 @@
+import pytest
+
+from server.simulation_service import (
+    CoenvEnvironment,
+    calculate_reward,
+    check_task_complete,
+    get_objective_for_task,
+)
+from models import CoenvAction
+
+
+class StubPod:
+    def __init__(self, deployment: str, status: str):
+        self.deployment = deployment
+        self.status = status
+
+
+class StubWorld:
+    def __init__(self, pods):
+        self._pods = pods
+
+    def get_pods(self):
+        return self._pods
+
+
+def test_get_objective_for_task_known_and_unknown():
+    known = get_objective_for_task("pod_recovery")
+    unknown = get_objective_for_task("unknown-task")
+
+    assert "crash-looping" in known
+    assert unknown == "Maintain cluster health"
+
+
+def test_calculate_reward_pod_recovery():
+    world = StubWorld(
+        [
+            StubPod("frontend", "Running"),
+            StubPod("frontend", "Running"),
+            StubPod("frontend", "CrashLoopBackOff"),
+        ]
+    )
+
+    reward = calculate_reward(world, "pod_recovery")
+    assert reward == pytest.approx(2 / 3)
+
+
+def test_check_task_complete_incident_true_and_false():
+    healthy_world = StubWorld(
+        [
+            StubPod("auth-service", "Running"),
+            StubPod("api-gateway", "Running"),
+            StubPod("frontend", "Running"),
+        ]
+    )
+    unhealthy_world = StubWorld(
+        [
+            StubPod("auth-service", "Running"),
+            StubPod("api-gateway", "CrashLoopBackOff"),
+            StubPod("frontend", "Running"),
+        ]
+    )
+
+    assert check_task_complete(healthy_world, "incident") is True
+    assert check_task_complete(unhealthy_world, "incident") is False
+
+
+def test_environment_reset_sets_task_and_returns_observation():
+    env = CoenvEnvironment()
+
+    obs = env.reset(task="autoscaling")
+
+    assert env.current_task == "autoscaling"
+    assert obs.objective == env.current_objective
+    assert obs.done is False
+    assert obs.reward == 0.0
+    assert "task" in obs.metadata
+
+
+def test_environment_step_scale_and_describe_paths():
+    env = CoenvEnvironment()
+    env.reset(task="pod_recovery")
+
+    scale_obs = env.step(
+        CoenvAction(action_type="scale", deployment="frontend", replicas=4)
+    )
+    assert "scaled" in scale_obs.metadata
+    assert scale_obs.step >= 1
+
+    describe_obs = env.step(
+        CoenvAction(action_type="describe", resource_type="deployment", name="frontend")
+    )
+    assert "described" in describe_obs.metadata
+    assert "describe_detail" in describe_obs.metadata
+
+
+def test_environment_step_exception_is_captured_in_metadata(monkeypatch):
+    env = CoenvEnvironment()
+    env.reset(task="pod_recovery")
+
+    def _boom(*args, **kwargs):
+        raise RuntimeError("forced failure")
+
+    monkeypatch.setattr(env.world, "scale", _boom)
+
+    action = CoenvAction(action_type="scale", deployment="frontend", replicas=2)
+
+    obs = env.step(action)
+    assert "error" in obs.metadata
+    assert "forced failure" in obs.metadata["error"]

tests/test_world_advanced.py

@@ -0,0 +1,140 @@
+import pytest
+
+from server.COEnv_environment import World
+from server.models import ClusterEvent
+
+
+@pytest.fixture
+def world() -> World:
+    return World({"num_nodes": 3})
+
+
+def test_world_seed_makes_initial_state_deterministic():
+    config = {"num_nodes": 3}
+    world_a = World(config, seed=12345)
+    world_b = World(config, seed=12345)
+
+    pod_names_a = [p["name"] for p in world_a.cluster_state["pods"]]
+    pod_names_b = [p["name"] for p in world_b.cluster_state["pods"]]
+
+    assert pod_names_a == pod_names_b
+
+
+def test_get_pods_filters_by_namespace_and_selector(world: World):
+    world.cluster_state["pods"].append(
+        {
+            "name": "custom-frontend-pod",
+            "namespace": "staging",
+            "status": "Running",
+            "node": "node-1",
+            "restarts": 0,
+            "cpu_request": 100,
+            "mem_request": 128,
+            "cpu_limit": 200,
+            "mem_limit": 256,
+            "deployment": "frontend",
+            "labels": {"tier": "web", "app": "frontend"},
+            "last_updated": "2026-04-06T00:00:00",
+        }
+    )
+
+    staging_frontend = world.get_pods(namespace="staging", selector={"app": "frontend"})
+    assert len(staging_frontend) == 1
+    assert staging_frontend[0].name == "custom-frontend-pod"
+
+    by_custom_label = world.get_pods(namespace="staging", selector={"tier": "web"})
+    assert len(by_custom_label) == 1
+    assert by_custom_label[0].deployment == "frontend"
+
+    no_match = world.get_pods(namespace="prod", selector={"tier": "web"})
+    assert no_match == []
+
+
+def test_set_hpa_updates_existing_and_clamps_deployment_replicas(world: World):
+    world.scale("backend", 12)
+
+    updated = world.set_hpa("backend", min_replicas=2, max_replicas=5, cpu_target_percent=65)
+    assert updated is True
+
+    backend_dep = next(d for d in world.cluster_state["deployments"] if d["name"] == "backend")
+    backend_hpa = next(h for h in world.cluster_state["hpas"] if h["name"] == "backend-hpa")
+
+    assert backend_dep["desired_replicas"] == 5
+    assert backend_hpa["min_replicas"] == 2
+    assert backend_hpa["max_replicas"] == 5
+    assert backend_hpa["cpu_target_percent"] == 65
+    assert backend_hpa["current_replicas"] == 5
+
+    hpa_events = [e for e in world.events if e.reason == "HorizontalPodAutoscalerUpdated" and e.involved_object == "backend"]
+    assert len(hpa_events) >= 1
+
+
+def test_drain_node_evicts_and_reassigns_pods(world: World):
+    target_node = "node-1"
+    pods_on_node_before = [p for p in world.cluster_state["pods"] if p.get("node") == target_node]
+    assert len(pods_on_node_before) > 0
+
+    drained = world.drain_node(target_node)
+    assert drained is True
+
+    node = next(n for n in world.cluster_state["nodes"] if n["name"] == target_node)
+    assert node["status"] == "SchedulingDisabled"
+
+    pods_with_original_names = {
+        p["name"] for p in pods_on_node_before
+    }
+    pods_after = [p for p in world.cluster_state["pods"] if p["name"] in pods_with_original_names]
+
+    assert len(pods_after) == len(pods_on_node_before)
+    assert all(p["status"] == "Pending" for p in pods_after)
+    assert all(p.get("node") != target_node for p in pods_after)
+
+
+def test_drain_node_with_no_ready_targets_unassigns_pods(world: World):
+    for node in world.cluster_state["nodes"]:
+        if node["name"] != "node-1":
+            node["status"] = "NotReady"
+
+    pods_on_node_before = [p for p in world.cluster_state["pods"] if p.get("node") == "node-1"]
+    assert len(pods_on_node_before) > 0
+
+    drained = world.drain_node("node-1")
+    assert drained is True
+
+    names = {p["name"] for p in pods_on_node_before}
+    pods_after = [p for p in world.cluster_state["pods"] if p["name"] in names]
+    assert all(p.get("node") is None for p in pods_after)
+    assert all(p["status"] == "Pending" for p in pods_after)
+
+
+def test_describe_deployment_includes_related_pods_and_recent_events(world: World):
+    for i in range(12):
+        world.events.append(
+            ClusterEvent(
+                event_id=f"event-frontend-{i}",
+                timestamp="2026-04-06T00:00:00",
+                type="Normal",
+                reason="TestEvent",
+                message=f"frontend event {i}",
+                involved_object="frontend",
+            )
+        )
+
+    detail = world.describe("deployment", "frontend")
+
+    assert detail["found"] is True
+    assert detail["name"] == "frontend"
+    assert detail["type"] == "deployment"
+    assert all(p.get("deployment") == "frontend" for p in detail["related_pods"])
+    assert len(detail["recent_events"]) == 10
+    assert all(evt["involved_object"] == "frontend" for evt in detail["recent_events"])
+
+
+def test_describe_unsupported_or_missing_resource(world: World):
+    unsupported = world.describe("secret", "top-secret")
+    assert unsupported["found"] is False
+    assert "Unsupported resource_type" in unsupported["error"]
+
+    missing = world.describe("service", "does-not-exist")
+    assert missing["found"] is False
+    assert "not found" in missing["error"]