"""row_get 簇统一单测（A7）：三处走同一函数，语义钉住。

只用标准库 + 内存 SQLite，不联网。锁定点：
- 统一身份：auto_memory._row_get 与 lifecycle._row_get 就是
  peer_messages.row_get 本体（不是又一份复写）；gapfill G3
  之后 approvals、automations、marker_tools、todos、wait_ops
  的 row_get 以及 resident.py 导入的 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 行上都经统一口出正确字段。
- gapfill G3 全仓收尾：五处旧版与统一版在真实行类型下
  语义完全相同（缺键/缺列/空行一律回 None，调用处均不传
  default），故统一不算改语义；todos / wait_ops 模块
  docstring 里“不 import cloud 其它模块”的表述同步修正为
  “除行取值复用 peer_messages（本身只依赖标准库）外不
  import cloud 其它模块”。
"""

import sqlite3
import sys
import unittest
from pathlib import Path

try:
    import cloud.approvals as approvals
    import cloud.auto_memory as auto_memory
    import cloud.automations as automations
    import cloud.lifecycle as lifecycle
    import cloud.marker_tools as marker_tools
    import cloud.todos as todos
    import cloud.wait_ops as wait_ops
    from cloud.peer_messages import row_get
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    import cloud.approvals as approvals
    import cloud.auto_memory as auto_memory
    import cloud.automations as automations
    import cloud.lifecycle as lifecycle
    import cloud.marker_tools as marker_tools
    import cloud.todos as todos
    import cloud.wait_ops as wait_ops
    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_gapfill_five_modules_are_the_same_function(self):
        # gapfill G3：五处私有 row_get 已删，模块属性就是统一本体；
        # resident.py 同样从 peer_messages 导入，故一并钉住。
        import cloud.resident as resident

        self.assertIs(approvals.row_get, row_get)
        self.assertIs(automations.row_get, row_get)
        self.assertIs(marker_tools.row_get, row_get)
        self.assertIs(todos.row_get, row_get)
        self.assertIs(wait_ops.row_get, row_get)
        self.assertIs(resident.row_get, row_get)
        # 五处调用处均不传 default，缺失回落仍是 None。
        for fn in (approvals.row_get, automations.row_get,
                   marker_tools.row_get, todos.row_get,
                   wait_ops.row_get):
            self.assertIsNone(fn({}, "nope"))
            self.assertIsNone(fn(None, "nope"))
            self.assertEqual(fn({"a": 0}, "a"), 0)

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