test_deployment_query.py
4.16 KB
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
"""U-043 tests: deployment_query.py. All real (no mocking needed -- sqlite
is fast/free/local).
Run: python3 -m unittest src.test_deployment_query -v
"""
import unittest
from src import deployment_query, schema
class DeploymentQueryTestBase(unittest.TestCase):
def setUp(self):
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
def tearDown(self):
self.conn.close()
def _insert(self, deployment_id, status, **kwargs):
cols = ["deployment_id", "git_commit_hash", "previous_commit_hash", "status", "previous_instance_dir"]
vals = [deployment_id, kwargs.get("git_commit_hash", "c1"), kwargs.get("previous_commit_hash", "c0"),
status, kwargs.get("previous_instance_dir", "/prev")]
for extra_col in ("candidate_port", "candidate_instance_dir", "started_at", "cutover_at",
"monitoring_closed_at", "rollback_at", "rollback_reason"):
if extra_col in kwargs:
cols.append(extra_col)
vals.append(kwargs[extra_col])
self.conn.execute(
"INSERT INTO deployment ({}) VALUES ({})".format(", ".join(cols), ", ".join("?" * len(cols))), vals
)
self.conn.commit()
class ListDeploymentsTest(DeploymentQueryTestBase):
def test_newest_first(self):
self._insert("d1", "finalized")
self._insert("d2", "finalized")
result = deployment_query.list_deployments(self.conn)
self.assertEqual([d["deployment_id"] for d in result], ["d2", "d1"])
def test_respects_limit(self):
for i in range(5):
self._insert("d{}".format(i), "finalized")
result = deployment_query.list_deployments(self.conn, limit=2)
self.assertEqual(len(result), 2)
def test_is_active_flag(self):
self._insert("d1", "monitoring")
self._insert("d2", "finalized")
by_id = {d["deployment_id"]: d for d in deployment_query.list_deployments(self.conn)}
self.assertTrue(by_id["d1"]["is_active"])
self.assertFalse(by_id["d2"]["is_active"])
class GetCurrentDeploymentTest(DeploymentQueryTestBase):
def test_none_when_nothing_active(self):
self._insert("d1", "finalized")
self.assertIsNone(deployment_query.get_current_deployment(self.conn))
def test_returns_the_one_active_deployment(self):
self._insert("d1", "finalized")
self._insert("d2", "health_check_running")
result = deployment_query.get_current_deployment(self.conn)
self.assertEqual(result["deployment_id"], "d2")
self.assertTrue(result["is_active"])
def test_blocked_pending_human_is_not_active(self):
# 01-M 1.15's ux_deployment_active constraint treats blocked_pending_human
# as terminal too -- it's the one status that needs a human, not the
# scheduler, to move it forward again.
self._insert("d1", "blocked_pending_human")
self.assertIsNone(deployment_query.get_current_deployment(self.conn))
class GetDeploymentTest(DeploymentQueryTestBase):
def test_not_found_raises(self):
with self.assertRaises(deployment_query.DeploymentNotFoundError):
deployment_query.get_deployment(self.conn, "nonexistent")
def test_includes_recent_health_signals_newest_first(self):
self._insert("d1", "monitoring")
self.conn.execute(
"INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) "
"VALUES ('s1', 'd1', '2020-01-01 00:00:00', 'process_alive', 'ok')"
)
self.conn.execute(
"INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) "
"VALUES ('s2', 'd1', '2020-01-01 00:05:00', 'http_health', 'ok')"
)
self.conn.commit()
result = deployment_query.get_deployment(self.conn, "d1")
self.assertEqual([s["signal_id"] for s in result["health_signals"]], ["s2", "s1"])
def test_no_signals_yet_gives_empty_list(self):
self._insert("d1", "staged")
result = deployment_query.get_deployment(self.conn, "d1")
self.assertEqual(result["health_signals"], [])
if __name__ == "__main__":
unittest.main()