"""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()