From 29af9c7bbfc1272d2d98fcf16a5a5cf84b800e2e Mon Sep 17 00:00:00 2001 From: host Date: Wed, 4 Nov 2026 09:09:58 +0000 Subject: [PATCH] gapfill G2 (goal): cover all 15 missing goal_live_update edge branches (write validation, bad applied_version fallback, non-dict signals, consume failure paths, apply CAS null-version branch and CAS miss); directed coverage 81% -> 100%, production code untouched --- cloud/tests/test_gapfill_g2_goal_edges.py | 194 ++++++++++++++++++++++ 1 file changed, 194 insertions(+) create mode 100644 cloud/tests/test_gapfill_g2_goal_edges.py diff --git a/cloud/tests/test_gapfill_g2_goal_edges.py b/cloud/tests/test_gapfill_g2_goal_edges.py new file mode 100644 index 0000000..0a997ce --- /dev/null +++ b/cloud/tests/test_gapfill_g2_goal_edges.py @@ -0,0 +1,194 @@ +"""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()