195 lines
7.3 KiB
Python
195 lines
7.3 KiB
Python
"""gapfill G2：goal_live_update.py 边缘分支补测。
|
||
|
||
补测基线（055da3c 线定向子集对本模块的覆盖）：
|
||
80 语句缺 15 行：57、59（写信号校验抛错）、79-80（读信号
|
||
applied_version 非整数回落）、88（读信号非 dict）、109-110
|
||
（删信号 applied_version 非整数回落）、115-116（删信号
|
||
文件缺失/坏 JSON）、118（删信号非 dict）、121（删信号
|
||
version 非整数）、126-127（删信号 os.remove 失败）、181
|
||
（apply_goal_update old_version 为 0 分支）、187（apply
|
||
CAS 未命中回 None）。
|
||
|
||
只加测试，不改生产代码；信号只写临时目录，不联网。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
try:
|
||
from cloud.goal_live_update import (
|
||
apply_goal_update,
|
||
consume_goal_update_signal,
|
||
ensure_goal_history_table,
|
||
read_goal_update_signal,
|
||
signal_path,
|
||
write_goal_update_signal,
|
||
)
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.goal_live_update import (
|
||
apply_goal_update,
|
||
consume_goal_update_signal,
|
||
ensure_goal_history_table,
|
||
read_goal_update_signal,
|
||
signal_path,
|
||
write_goal_update_signal,
|
||
)
|
||
|
||
|
||
class WriteValidationTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.state = os.path.join(self.tmp.name, "host-state.json")
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def test_bad_version_raises(self):
|
||
# 57：version 非正整数（0、负数、bool、字符串）一律
|
||
# ValueError，绝不写出信号文件。
|
||
for bad in (0, -1, True, "2", 2.5):
|
||
with self.assertRaises(ValueError):
|
||
write_goal_update_signal(
|
||
self.state, "t1", bad, "new goal")
|
||
self.assertFalse(os.path.exists(
|
||
signal_path(self.state, "t1")))
|
||
|
||
def test_bad_goal_raises(self):
|
||
# 59：goal 非空字符串校验（空串、空白串、非字符串）。
|
||
for bad in ("", " ", None, 123):
|
||
with self.assertRaises(ValueError):
|
||
write_goal_update_signal(self.state, "t1", 1, bad)
|
||
|
||
|
||
class ReadEdgeTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.state = os.path.join(self.tmp.name, "host-state.json")
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def test_bad_applied_version_falls_back_to_zero(self):
|
||
# 79-80：applied_version 非整数（int() 抛 ValueError /
|
||
# TypeError）时按 0 计，version=2 的信号仍可应用。
|
||
write_goal_update_signal(self.state, "t1", 2, "goal v2")
|
||
self.assertEqual(
|
||
read_goal_update_signal(self.state, "t1", "not-a-number"),
|
||
(2, "goal v2"))
|
||
|
||
def test_non_dict_signal_returns_none(self):
|
||
# 88：信号文件是合法 JSON 但不是 dict（列表）时回 None。
|
||
path = signal_path(self.state, "t1")
|
||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
json.dump([1, 2], fh)
|
||
self.assertIsNone(read_goal_update_signal(self.state, "t1", 0))
|
||
|
||
|
||
class ConsumeEdgeTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.state = os.path.join(self.tmp.name, "host-state.json")
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def _write_raw(self, payload):
|
||
path = signal_path(self.state, "t1")
|
||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
json.dump(payload, fh)
|
||
return path
|
||
|
||
def test_missing_file_returns_false(self):
|
||
# 115-116：信号文件不存在时回 False，不抛。
|
||
self.assertFalse(
|
||
consume_goal_update_signal(self.state, "t1", 5))
|
||
|
||
def test_bad_applied_version_falls_back_to_zero(self):
|
||
# 109-110：applied_version 非整数时按 0 计；version=1
|
||
# 的信号大于 0，故不删、回 False（下一条才是删的路）。
|
||
write_goal_update_signal(self.state, "t1", 1, "goal v1")
|
||
self.assertFalse(consume_goal_update_signal(
|
||
self.state, "t1", "not-a-number"))
|
||
self.assertTrue(os.path.exists(signal_path(self.state, "t1")))
|
||
|
||
def test_non_dict_signal_returns_false(self):
|
||
# 118：信号文件非 dict 时回 False，文件原样保留。
|
||
path = self._write_raw([1, 2])
|
||
self.assertFalse(
|
||
consume_goal_update_signal(self.state, "t1", 9))
|
||
self.assertTrue(os.path.exists(path))
|
||
|
||
def test_non_int_version_returns_false(self):
|
||
# 121：version 非整数（字符串）时回 False，文件保留。
|
||
path = self._write_raw({"version": "3", "goal": "g"})
|
||
self.assertFalse(
|
||
consume_goal_update_signal(self.state, "t1", 9))
|
||
self.assertTrue(os.path.exists(path))
|
||
|
||
def test_remove_failure_returns_false(self):
|
||
# 126-127：os.remove 失败时回 False，不抛。
|
||
write_goal_update_signal(self.state, "t1", 1, "goal v1")
|
||
with mock.patch("cloud.goal_live_update.os.remove",
|
||
side_effect=OSError("disk gone")):
|
||
self.assertFalse(
|
||
consume_goal_update_signal(self.state, "t1", 1))
|
||
self.assertTrue(os.path.exists(signal_path(self.state, "t1")))
|
||
|
||
|
||
class ApplyGoalUpdateTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.conn = sqlite3.connect(":memory:")
|
||
self.conn.execute(
|
||
"CREATE TABLE tasks(id TEXT PRIMARY KEY, goal TEXT, "
|
||
"goal_version INTEGER, updated_at REAL)")
|
||
ensure_goal_history_table(self.conn)
|
||
self.conn.execute(
|
||
"INSERT INTO tasks VALUES('t1', 'old', NULL, 0)")
|
||
self.conn.execute(
|
||
"INSERT INTO tasks VALUES('t2', 'old', 5, 0)")
|
||
self.conn.commit()
|
||
|
||
def tearDown(self):
|
||
self.conn.close()
|
||
|
||
def test_null_version_branch(self):
|
||
# 181：old_version 为 NULL（按 0 计）时走 NULL/0 分支，
|
||
# 新 version 为 1，history 写一行。
|
||
self.assertEqual(
|
||
apply_goal_update(self.conn, "t1", "old", None,
|
||
"new", "u1", 123.0), 1)
|
||
row = self.conn.execute(
|
||
"SELECT goal, goal_version FROM tasks WHERE id='t1'"
|
||
).fetchone()
|
||
self.assertEqual(row, ("new", 1))
|
||
hist = self.conn.execute(
|
||
"SELECT old_goal, new_goal, version FROM task_goal_history "
|
||
"WHERE task_id='t1'").fetchall()
|
||
self.assertEqual(hist, [("old", "new", 1)])
|
||
|
||
def test_cas_miss_returns_none_without_history(self):
|
||
# 187：CAS 未命中（库里 version=5，调用方按 4 来）时
|
||
# 回 None，任务行与 history 一律不动。
|
||
self.assertIsNone(
|
||
apply_goal_update(self.conn, "t2", "old", 4,
|
||
"new", "u1", 123.0))
|
||
row = self.conn.execute(
|
||
"SELECT goal, goal_version FROM tasks WHERE id='t2'"
|
||
).fetchone()
|
||
self.assertEqual(row, ("old", 5))
|
||
self.assertEqual(self.conn.execute(
|
||
"SELECT COUNT(*) FROM task_goal_history "
|
||
"WHERE task_id='t2'").fetchone()[0], 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|