123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
"""code-to10 v6 A9：caps/quota 剩余缺口补测。
|
||
|
||
补测基线（v5 头 10023fb 全量覆盖率）：
|
||
- cloud/caps.py 31 语句缺 2 行：70-71，running_count 的
|
||
TypeError 分支（快照无 len 时回 0）。
|
||
- cloud/quota.py 46 语句缺 10 行：79-80（utc_month 直接
|
||
调用）、128-131（used 行回落）、140-143（unpriced 行回落）。
|
||
|
||
口径说明（写在测试里，不写成定稿）：
|
||
- actual-only 会计口径（C4）在本批仍为临时口径、未确认。
|
||
本文件用伪造连接测的是“行取不到时的回落”这一防御分支，
|
||
完全不经过 SQL 口径：若 C4 日后翻转，SQL 层（actual-only
|
||
过滤）的测试须重审，本文件的回落断言不受影响。
|
||
- 真实 PG 未接通：本地无 DSN 是预期，本文件只用伪造连接
|
||
与标准库 datetime，不连真实数据库。
|
||
|
||
只加测试，不改生产代码。
|
||
"""
|
||
|
||
import sys
|
||
import unittest
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.caps import running_count
|
||
from cloud.quota import monthly_usage, utc_month
|
||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.caps import running_count
|
||
from cloud.quota import monthly_usage, utc_month
|
||
|
||
|
||
class CapsRunningCountFallbackTestCase(unittest.TestCase):
|
||
def test_none_snapshot_returns_zero(self):
|
||
# len(None) 抛 TypeError -> 回落 0（caps.py:70-71）。
|
||
self.assertEqual(running_count(None), 0)
|
||
|
||
def test_len_less_snapshot_returns_zero(self):
|
||
# int / 无 __len__ 对象同样走 TypeError 分支。
|
||
self.assertEqual(running_count(123), 0)
|
||
self.assertEqual(running_count(object()), 0)
|
||
|
||
def test_normal_snapshot_still_counts(self):
|
||
# 回落分支不得改变正常计数语义。
|
||
self.assertEqual(running_count({"a": {}, "b": {}}), 2)
|
||
|
||
|
||
class QuotaUtcMonthTestCase(unittest.TestCase):
|
||
def test_utc_month_direct_call(self):
|
||
# quota.py:79-80：显式 now 直接格式化，不走 now=None 分支。
|
||
moment = datetime(2026, 11, 2, 8, 30, tzinfo=timezone.utc)
|
||
self.assertEqual(utc_month(moment), "2026-11")
|
||
|
||
def test_utc_month_default_is_current_utc_month(self):
|
||
# now=None 分支：返回当前 UTC 自然月，格式 YYYY-MM。
|
||
got = utc_month()
|
||
expected = datetime.now(timezone.utc).strftime("%Y-%m")
|
||
self.assertEqual(got, expected)
|
||
self.assertRegex(got, r"^\d{4}-\d{2}$")
|
||
|
||
|
||
class _FakeCursor:
|
||
def __init__(self, row):
|
||
self._row = row
|
||
|
||
def fetchone(self):
|
||
return self._row
|
||
|
||
|
||
class _FakeConn:
|
||
"""按 execute 调用顺序依次返回预设行（used 查询在前，
|
||
unpriced 查询在后）；不执行任何 SQL。"""
|
||
|
||
def __init__(self, rows):
|
||
self._rows = list(rows)
|
||
self.calls = []
|
||
|
||
def execute(self, sql, params=None):
|
||
self.calls.append(sql)
|
||
return _FakeCursor(self._rows.pop(0))
|
||
|
||
|
||
class _StatefulRow:
|
||
"""第一次按位置取抛 KeyError（模拟 _row_value 取不到），
|
||
第二次按位置取返回预设值（模拟位置回落成功）。"""
|
||
|
||
def __init__(self, value):
|
||
self._value = value
|
||
self._calls = 0
|
||
|
||
def __getitem__(self, key):
|
||
self._calls += 1
|
||
if self._calls == 1:
|
||
raise KeyError(key)
|
||
return self._value
|
||
|
||
|
||
class QuotaMonthlyUsageFallbackTestCase(unittest.TestCase):
|
||
def test_none_rows_fall_back_to_zero(self):
|
||
# fetchone 回 None：_row_value 回 None 后位置取再次抛
|
||
# TypeError -> used/unpriced 双双回 0（quota.py:128-131、
|
||
# 140-143 的 except 分支）。
|
||
conn = _FakeConn([None, None])
|
||
self.assertEqual(monthly_usage(conn, "u1", "2026-11"), (0, 0))
|
||
self.assertEqual(len(conn.calls), 2)
|
||
|
||
def test_positional_fallback_recovers_value(self):
|
||
# 第一次取不到、第二次按位置取到：回落值被采纳，
|
||
# 不是一律归零（quota.py:129、141 的成功回落行）。
|
||
conn = _FakeConn([_StatefulRow(7000), _StatefulRow(42)])
|
||
self.assertEqual(monthly_usage(conn, "u1", "2026-11"),
|
||
(7000, 42))
|
||
|
||
def test_positional_fallback_partial(self):
|
||
# used 行回落成功、unpriced 行为 None 回零：两路独立。
|
||
conn = _FakeConn([_StatefulRow(500), None])
|
||
self.assertEqual(monthly_usage(conn, "u1", "2026-11"), (500, 0))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|