Spaces:
No application file
No application file
File size: 7,403 Bytes
57c06cb dfc56a2 57c06cb dfc56a2 57c06cb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 | import pytest
from pydantic import ValidationError
from server.actions import (
ScaleAction,
PatchAction,
DeletePodAction,
RolloutRestartAction,
SetHPAAction,
DrainNodeAction,
DescribeAction,
WaitAction,
parse_action,
)
class TestScaleAction:
def test_valid_scale_action(self):
action = ScaleAction(action_type="scale", deployment="frontend", replicas=3)
assert action.deployment == "frontend"
assert action.replicas == 3
def test_scale_action_rejects_zero_replicas(self):
with pytest.raises(ValidationError):
ScaleAction(action_type="scale", deployment="frontend", replicas=0)
def test_scale_action_rejects_negative_replicas(self):
with pytest.raises(ValidationError):
ScaleAction(action_type="scale", deployment="frontend", replicas=-1)
def test_scale_action_rejects_too_many_replicas(self):
with pytest.raises(ValidationError):
ScaleAction(action_type="scale", deployment="frontend", replicas=21)
def test_scale_action_accepts_boundary_values(self):
action_min = ScaleAction(action_type="scale", deployment="frontend", replicas=1)
action_max = ScaleAction(action_type="scale", deployment="frontend", replicas=20)
assert action_min.replicas == 1
assert action_max.replicas == 20
class TestPatchAction:
def test_valid_patch_action(self):
action = PatchAction(
action_type="patch",
resource_type="deployment",
name="frontend",
patch={"env": [{"name": "DB_HOST", "value": "db.prod.internal"}]}
)
assert action.resource_type == "deployment"
assert action.name == "frontend"
def test_patch_action_rejects_invalid_resource_type(self):
with pytest.raises(ValidationError):
PatchAction(
action_type="patch",
resource_type="invalid",
name="frontend",
patch={}
)
class TestDeletePodAction:
def test_valid_delete_pod_action(self):
action = DeletePodAction(action_type="delete_pod", pod_name="frontend-7d9f-xkp2")
assert action.pod_name == "frontend-7d9f-xkp2"
class TestRolloutRestartAction:
def test_valid_rollout_restart_action(self):
action = RolloutRestartAction(action_type="rollout_restart", deployment="frontend")
assert action.deployment == "frontend"
class TestSetHPAAction:
def test_valid_hpa_action(self):
action = SetHPAAction(
action_type="set_hpa",
deployment="api",
min_replicas=2,
max_replicas=10,
cpu_target_percent=70
)
assert action.deployment == "api"
assert action.min_replicas == 2
assert action.max_replicas == 10
def test_hpa_action_rejects_max_less_than_min(self):
with pytest.raises(ValidationError):
SetHPAAction(
action_type="set_hpa",
deployment="api",
min_replicas=5,
max_replicas=2,
cpu_target_percent=60
)
def test_hpa_action_rejects_invalid_cpu_target(self):
with pytest.raises(ValidationError):
SetHPAAction(
action_type="set_hpa",
deployment="api",
min_replicas=1,
max_replicas=10,
cpu_target_percent=5
)
def test_hpa_action_accepts_boundary_cpu_target(self):
action_min = SetHPAAction(
action_type="set_hpa",
deployment="api",
min_replicas=1,
max_replicas=10,
cpu_target_percent=10
)
action_max = SetHPAAction(
action_type="set_hpa",
deployment="api",
min_replicas=1,
max_replicas=10,
cpu_target_percent=90
)
assert action_min.cpu_target_percent == 10
assert action_max.cpu_target_percent == 90
class TestDrainNodeAction:
def test_valid_drain_node_action(self):
action = DrainNodeAction(action_type="drain_node", node_name="node-1")
assert action.node_name == "node-1"
class TestDescribeAction:
def test_valid_describe_action(self):
action = DescribeAction(
action_type="describe",
resource_type="deployment",
name="frontend"
)
assert action.resource_type == "deployment"
assert action.name == "frontend"
def test_describe_action_rejects_invalid_resource_type(self):
with pytest.raises(ValidationError):
DescribeAction(
action_type="describe",
resource_type="invalid",
name="frontend"
)
class TestParseAction:
def test_parse_scale_action(self):
raw = {"action_type": "scale", "deployment": "frontend", "replicas": 3}
action = parse_action(raw)
assert isinstance(action, ScaleAction)
assert action.deployment == "frontend"
assert action.replicas == 3
def test_parse_delete_pod_action(self):
raw = {"action_type": "delete_pod", "pod_name": "frontend-7d9f-xkp2"}
action = parse_action(raw)
assert isinstance(action, DeletePodAction)
assert action.pod_name == "frontend-7d9f-xkp2"
def test_parse_patch_action(self):
raw = {
"action_type": "patch",
"resource_type": "deployment",
"name": "frontend",
"patch": {"env": [{"name": "DB_HOST", "value": "db.prod.internal"}]}
}
action = parse_action(raw)
assert isinstance(action, PatchAction)
assert action.name == "frontend"
def test_parse_rollout_restart_action(self):
raw = {"action_type": "rollout_restart", "deployment": "frontend"}
action = parse_action(raw)
assert isinstance(action, RolloutRestartAction)
assert action.deployment == "frontend"
def test_parse_hpa_action(self):
raw = {
"action_type": "set_hpa",
"deployment": "api",
"min_replicas": 2,
"max_replicas": 10,
"cpu_target_percent": 70
}
action = parse_action(raw)
assert isinstance(action, SetHPAAction)
assert action.deployment == "api"
def test_parse_drain_node_action(self):
raw = {"action_type": "drain_node", "node_name": "node-1"}
action = parse_action(raw)
assert isinstance(action, DrainNodeAction)
assert action.node_name == "node-1"
def test_parse_describe_action(self):
raw = {"action_type": "describe", "resource_type": "deployment", "name": "frontend"}
action = parse_action(raw)
assert isinstance(action, DescribeAction)
assert action.name == "frontend"
def test_parse_wait_action(self):
raw = {"action_type": "wait"}
action = parse_action(raw)
assert isinstance(action, WaitAction)
def test_parse_unknown_action_type(self):
with pytest.raises(ValueError, match="Unknown action_type"):
parse_action({"action_type": "unknown_action"})
def test_parse_missing_action_type(self):
with pytest.raises(ValueError, match="Missing 'action_type'"):
parse_action({"deployment": "frontend"})
|