code-to10 v6 A7: unify row_get on peer_messages
- peer_messages.row_get gains default=None (was None-only); the three old variants agree on all real row types (None / dict / sqlite3.Row): auto_memory's old default was already None, and lifecycle's old None special-case and dict .get give identical results on those types, so no call-site behavior changes and no call site passes a non-None default. - auto_memory and lifecycle delete their private _row_get copies and import peer_messages.row_get as _row_get (precedent: server, worker, handler_project, handler_support already do this); peer_messages imports no cloud modules, so no import cycle. lifecycle's old broad-Exception catch differs only for pathological __getitem__ inputs, which are not DB rows and are out of contract; this is documented on row_get, not left as a second story. - Scope note: other modules' own row_get copies (approvals, automations, marker_tools, todos, wait_ops) are out of this item's stated scope and are untouched. - Also drops auto_memory's unused 'import os' (pre-existing F401), so touched files are ruff E,F clean. - Tests: cloud/tests/test_row_get_unified.py locks the identity (both aliases ARE row_get), the real-row-type matrix, the None-default fallback, and snapshot_to_dict via the unified path. (cherry picked from commit 1644a0b837bfefb13cdc3c93d87fc1b246026607)
This commit is contained in:
1 parent
c33bfd240e
commit
be2962525b
4 files changed
+116
-24
No files matched your search
@@ -0,0 +1,92 @@
|
||||
"""row_get 簇统一单测(A7):三处走同一函数,语义钉住。
|
||||
|
||||
只用标准库 + 内存 SQLite,不联网。锁定点:
|
||||
- 统一身份:auto_memory._row_get 与 lifecycle._row_get 就是
|
||||
peer_messages.row_get 本体(不是又一份复写)。
|
||||
- 真实行类型矩阵(None / dict / sqlite3.Row)下取值与回落
|
||||
与三处旧版一致:现值原样回(含 0 / 空串等假值),缺列 /
|
||||
缺键 / 空行回 default(默认 None)。
|
||||
- default 回落差异钉住:三处旧版的默认回落都是 None——
|
||||
auto_memory 旧版 default 参数默认 None,lifecycle 旧版无
|
||||
default 参数一律回 None,peer 旧版回 None;统一后两模块
|
||||
调用处仍不传 default,缺失时仍为 None,不算改语义。
|
||||
- 统一路径实测:lifecycle.snapshot_to_dict 在 dict 缺键行与
|
||||
sqlite3.Row 行上都经统一口出正确字段。
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
import cloud.auto_memory as auto_memory
|
||||
import cloud.lifecycle as lifecycle
|
||||
from cloud.peer_messages import row_get
|
||||
except ImportError:
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
import cloud.auto_memory as auto_memory
|
||||
import cloud.lifecycle as lifecycle
|
||||
from cloud.peer_messages import row_get
|
||||
|
||||
|
||||
def _sqlite_row():
|
||||
conn = sqlite3.connect(":memory:")
|
||||
conn.row_factory = sqlite3.Row
|
||||
try:
|
||||
conn.execute("CREATE TABLE t(a TEXT, b INTEGER)")
|
||||
conn.execute("INSERT INTO t(a, b) VALUES('x', 0)")
|
||||
full = conn.execute("SELECT a, b FROM t").fetchone()
|
||||
partial = conn.execute("SELECT a FROM t").fetchone()
|
||||
return full, partial
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
class RowGetUnifiedTestCase(unittest.TestCase):
|
||||
def test_all_three_are_the_same_function(self):
|
||||
self.assertIs(auto_memory._row_get, row_get)
|
||||
self.assertIs(lifecycle._row_get, row_get)
|
||||
|
||||
def test_real_row_type_matrix(self):
|
||||
full, partial = _sqlite_row()
|
||||
# 现值原样回(含假值 0,不被当成缺失)。
|
||||
self.assertEqual(row_get({"a": 0}, "a"), 0)
|
||||
self.assertEqual(row_get({"a": ""}, "a"), "")
|
||||
self.assertEqual(row_get(full, "a"), "x")
|
||||
self.assertEqual(row_get(full, "b"), 0)
|
||||
# 缺键 / 缺列 / 空行一律回 default(默认 None)。
|
||||
self.assertIsNone(row_get({"a": 1}, "missing"))
|
||||
self.assertIsNone(row_get(partial, "b"))
|
||||
self.assertIsNone(row_get(None, "a"))
|
||||
# 显式 default 只在缺失时生效,现值 None 也不被替换
|
||||
#(旧 auto_memory 版同款语义:取到什么回什么)。
|
||||
self.assertEqual(row_get({"a": 1}, "missing", default="d"), "d")
|
||||
self.assertEqual(row_get(None, "a", default=7), 7)
|
||||
self.assertIsNone(row_get({"a": None}, "a", default="d"))
|
||||
|
||||
def test_default_fallback_is_none_in_both_modules(self):
|
||||
# 两模块调用处均不传 default;缺失回落必须仍是 None,
|
||||
# 否则就是改了旧语义。
|
||||
self.assertIsNone(auto_memory._row_get({}, "nope"))
|
||||
self.assertIsNone(lifecycle._row_get({}, "nope"))
|
||||
self.assertIsNone(auto_memory._row_get(None, "nope"))
|
||||
self.assertIsNone(lifecycle._row_get(None, "nope"))
|
||||
|
||||
def test_lifecycle_snapshot_to_dict_via_unified_get(self):
|
||||
out = lifecycle.snapshot_to_dict({"id": "s1", "task_id": "t1"})
|
||||
self.assertEqual(out["id"], "s1")
|
||||
self.assertEqual(out["task_id"], "t1")
|
||||
self.assertIsNone(out["project_id"])
|
||||
self.assertIsNone(out["branch"])
|
||||
self.assertEqual(out["untracked"], [])
|
||||
full, _partial = _sqlite_row()
|
||||
out2 = lifecycle.snapshot_to_dict(
|
||||
{"id": "s2", "untracked_json": '["a.txt"]'})
|
||||
self.assertEqual(out2["untracked"], ["a.txt"])
|
||||
# sqlite Row 行同样走统一口:缺列不炸。
|
||||
self.assertIsNone(row_get(full, "untracked_json"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user