Skip to content

Commit ea9b8b6

Browse files
test(wl_limits): cover read_limit_config + get_counter_period_key + overflow (67% → 87%) — G6 batch 3
Adds 21 new tests across three classes: - TestReadLimitConfigIntegrity (7 tests): missing-file default, valid-checksum round-trip, tampered-checksum tolerant-fallback + warning log, legacy reset_hour_utc migration to reset_time_utc, out-of-range legacy value clamped to 00:00, both-keys-present drops legacy, corrupt JSON returns DEFAULT_LIMITS. - TestGetCounterPeriodKeyFrequencies (12 tests): never→permanent, invalid reset_time fallback, weekly with Monday reset, weekly with invalid dow falling back to 0, monthly past/before boundary, monthly invalid day falling back to 1, monthly short-month day clamping (Feb 31→28), yearly past/before boundary, yearly invalid month falling back to January, unknown frequency falls back to daily. All use `freeze_time` for deterministic date arithmetic. - TestIncrementDailyLimitOverflow (2 tests): MAX_TRACKED_ANALYSTS overflow path routes new users to shared `__overflow__` bucket (denial-of-service mitigation under attack); permanent-key mode prunes all date-based keys. Coverage delta on bin/wl_limits.py: 67% → 87% (+20pp). ═══════════════════════════════════════════════════════════════════════ G6 ENTIRELY COMPLETE — ALL 17 unit-testable backend modules now ≥80% ═══════════════════════════════════════════════════════════════════════ Unit-testable backend coverage post-G6: 89.13% (2631/2952 stmts), up from 78.0% at start of session. Per-module breakdown (all >=80%): wl_constants 100% wl_logging 100% wl_presence 100% wl_rbac 100% wl_validation 95% wl_hmac_key 97% wl_replay 92% wl_filelock 91% wl_csv 90% wl_audit 89% wl_rules 88% wl_limits 87% wl_notify 86% wl_ratelimit 85% wl_trash 85% wl_approval 83% wl_versions 80% Tests: 811 → 832 (+21 this batch; +160 total across the session). Whole-bin/ coverage: 37% (denominator dominated by the FORBIDDEN-as- test-target wl_handler.py at 3260 statements + the 5 scripted-input modules exercised only via the integration suite). Doc-drift: OK at build 663. Full unit suite: 832 passed / 1 skipped.
1 parent df82528 commit ea9b8b6

1 file changed

Lines changed: 296 additions & 0 deletions

File tree

tests/unit/test_limits.py

Lines changed: 296 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
from unittest import mock
1616
from unittest.mock import patch, MagicMock, mock_open
1717
from datetime import datetime, timezone
18+
from freezegun import freeze_time
1819

1920
# Add bin directory to path
2021
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'bin'))
@@ -830,3 +831,298 @@ def test_get_daily_limits_path_creates_versions_dir(tmp_path):
830831
path = _get_daily_limits_path()
831832
assert os.path.isdir(os.path.join(str(tmp_path), "_versions"))
832833
assert path.endswith("_daily_limits.json")
834+
835+
836+
# ═══════════════════════════════════════════════════════════════════════════
837+
# read_limit_config integrity + migration (lines 258-290)
838+
# ═══════════════════════════════════════════════════════════════════════════
839+
840+
841+
@pytest.mark.unit
842+
class TestReadLimitConfigIntegrity:
843+
"""Cover the body of read_limit_config (currently the largest gap)."""
844+
845+
def test_missing_config_returns_defaults(self, tmp_path):
846+
"""No file on disk → returns dict(DEFAULT_LIMITS) — the early-return path."""
847+
from wl_limits import read_limit_config, DEFAULT_LIMITS
848+
with patch('wl_limits.OWN_LOOKUPS', str(tmp_path)):
849+
result = read_limit_config()
850+
assert result == dict(DEFAULT_LIMITS)
851+
852+
def test_valid_signed_config_returns_persisted_values(self, tmp_path):
853+
"""File present with valid checksum → values returned as written."""
854+
from wl_limits import (read_limit_config, write_limit_config,
855+
DEFAULT_LIMITS)
856+
custom = dict(DEFAULT_LIMITS)
857+
custom["row_addition"] = 42
858+
with patch('wl_limits.OWN_LOOKUPS', str(tmp_path)):
859+
write_limit_config(custom)
860+
result = read_limit_config()
861+
assert result["row_addition"] == 42
862+
863+
def test_tampered_checksum_still_returns_data(self, tmp_path, caplog):
864+
"""Checksum mismatch logs warning but does NOT lock the app out
865+
(covers lines 263-270 — the integrity-failure soft path)."""
866+
from wl_limits import read_limit_config, DEFAULT_LIMITS
867+
path = tmp_path / "_versions" / "_limit_config.json"
868+
path.parent.mkdir(parents=True, exist_ok=True)
869+
config = dict(DEFAULT_LIMITS)
870+
config["_checksum"] = "0" * 64 # fake checksum
871+
path.write_text(json.dumps(config))
872+
873+
with patch('wl_limits.OWN_LOOKUPS', str(tmp_path)), \
874+
caplog.at_level("WARNING"):
875+
result = read_limit_config()
876+
# Data is still returned (tolerant policy)
877+
assert isinstance(result, dict)
878+
# Warning was logged
879+
assert any("CONFIG_INTEGRITY_FAILED" in rec.message
880+
for rec in caplog.records)
881+
882+
def test_legacy_reset_hour_utc_migrated(self, tmp_path):
883+
"""Old `reset_hour_utc: 7` rewrites to `reset_time_utc: "07:00"`
884+
(covers lines 272-278)."""
885+
from wl_limits import read_limit_config
886+
path = tmp_path / "_versions" / "_limit_config.json"
887+
path.parent.mkdir(parents=True, exist_ok=True)
888+
config = {"reset_hour_utc": 7}
889+
path.write_text(json.dumps(config))
890+
891+
with patch('wl_limits.OWN_LOOKUPS', str(tmp_path)):
892+
result = read_limit_config()
893+
assert result["reset_time_utc"] == "07:00"
894+
assert "reset_hour_utc" not in result
895+
896+
def test_legacy_reset_hour_utc_out_of_range_defaults_to_zero(self, tmp_path):
897+
"""Invalid reset_hour_utc (e.g., 99) → "00:00" fallback (line 278)."""
898+
from wl_limits import read_limit_config
899+
path = tmp_path / "_versions" / "_limit_config.json"
900+
path.parent.mkdir(parents=True, exist_ok=True)
901+
config = {"reset_hour_utc": 99}
902+
path.write_text(json.dumps(config))
903+
904+
with patch('wl_limits.OWN_LOOKUPS', str(tmp_path)):
905+
result = read_limit_config()
906+
assert result["reset_time_utc"] == "00:00"
907+
908+
def test_both_legacy_and_new_keys_drops_legacy(self, tmp_path):
909+
"""Both reset_hour_utc and reset_time_utc present → legacy dropped (line 282)."""
910+
from wl_limits import read_limit_config
911+
path = tmp_path / "_versions" / "_limit_config.json"
912+
path.parent.mkdir(parents=True, exist_ok=True)
913+
config = {"reset_hour_utc": 5, "reset_time_utc": "09:30"}
914+
path.write_text(json.dumps(config))
915+
916+
with patch('wl_limits.OWN_LOOKUPS', str(tmp_path)):
917+
result = read_limit_config()
918+
# New key wins; legacy is dropped
919+
assert result["reset_time_utc"] == "09:30"
920+
assert "reset_hour_utc" not in result
921+
922+
def test_corrupt_json_falls_back_to_defaults(self, tmp_path):
923+
"""JSON parse error → return DEFAULT_LIMITS (line 289-290)."""
924+
from wl_limits import read_limit_config, DEFAULT_LIMITS
925+
path = tmp_path / "_versions" / "_limit_config.json"
926+
path.parent.mkdir(parents=True, exist_ok=True)
927+
path.write_text("not valid json {{{")
928+
929+
with patch('wl_limits.OWN_LOOKUPS', str(tmp_path)):
930+
result = read_limit_config()
931+
assert result == dict(DEFAULT_LIMITS)
932+
933+
934+
# ═══════════════════════════════════════════════════════════════════════════
935+
# get_counter_period_key for weekly/monthly/yearly (lines 420-470)
936+
# ═══════════════════════════════════════════════════════════════════════════
937+
938+
939+
@pytest.mark.unit
940+
class TestGetCounterPeriodKeyFrequencies:
941+
"""Cover the weekly/monthly/yearly branches of get_counter_period_key.
942+
943+
The function returns a stringified period key that identifies the
944+
current reset bucket; tests freeze time and exercise each frequency.
945+
"""
946+
947+
def test_never_returns_permanent(self):
948+
"""freq=never → "permanent" sentinel (line 403)."""
949+
from wl_limits import get_counter_period_key
950+
result = get_counter_period_key({"reset_frequency": "never"})
951+
assert result == "permanent"
952+
953+
def test_invalid_reset_time_falls_back_to_midnight(self):
954+
"""reset_time_utc not parseable → fallback to 00:00 (lines 411-412)."""
955+
from wl_limits import get_counter_period_key
956+
with freeze_time("2026-05-19T15:00:00Z"):
957+
result = get_counter_period_key({
958+
"reset_frequency": "daily",
959+
"reset_time_utc": "not-a-time",
960+
})
961+
# Should not raise; produces a date string
962+
assert result == "2026-05-19"
963+
964+
@freeze_time("2026-05-19T15:00:00Z") # Tuesday
965+
def test_weekly_with_monday_reset(self):
966+
"""Weekly freq with reset_day_of_week=0 (Monday) → ISO week ending."""
967+
from wl_limits import get_counter_period_key
968+
result = get_counter_period_key({
969+
"reset_frequency": "weekly",
970+
"reset_day_of_week": 0, # Monday
971+
"reset_time_utc": "00:00",
972+
})
973+
# 2026-05-19 is a Tuesday; week reset on Monday at 00:00 means
974+
# the current bucket starts Monday 2026-05-18.
975+
assert "W" in result # ISO week format like "2026-W21-Mon"
976+
977+
@freeze_time("2026-05-19T15:00:00Z")
978+
def test_weekly_invalid_dow_falls_back_to_zero(self):
979+
"""reset_day_of_week=99 → falls back to 0 (line 425)."""
980+
from wl_limits import get_counter_period_key
981+
# Should not raise — uses Monday (0) as fallback
982+
result = get_counter_period_key({
983+
"reset_frequency": "weekly",
984+
"reset_day_of_week": 99,
985+
})
986+
assert "W" in result
987+
988+
@freeze_time("2026-05-19T15:00:00Z")
989+
def test_monthly_with_day_15_after_boundary(self):
990+
"""Monthly freq, reset on day 15. Today is May 19 >= boundary → "2026-05"."""
991+
from wl_limits import get_counter_period_key
992+
result = get_counter_period_key({
993+
"reset_frequency": "monthly",
994+
"reset_day_of_month": 15,
995+
"reset_time_utc": "00:00",
996+
})
997+
assert result == "2026-05"
998+
999+
@freeze_time("2026-05-10T15:00:00Z")
1000+
def test_monthly_before_boundary_returns_previous_month(self):
1001+
"""Monthly freq, today before reset day → previous month's key."""
1002+
from wl_limits import get_counter_period_key
1003+
result = get_counter_period_key({
1004+
"reset_frequency": "monthly",
1005+
"reset_day_of_month": 15,
1006+
"reset_time_utc": "00:00",
1007+
})
1008+
assert result == "2026-04"
1009+
1010+
@freeze_time("2026-05-19T15:00:00Z")
1011+
def test_monthly_invalid_day_falls_back_to_one(self):
1012+
"""reset_day_of_month=99 → falls back to 1 (line 437)."""
1013+
from wl_limits import get_counter_period_key
1014+
result = get_counter_period_key({
1015+
"reset_frequency": "monthly",
1016+
"reset_day_of_month": 99,
1017+
})
1018+
# Day=1 always satisfies "now>=boundary" past midnight today
1019+
assert result == "2026-05"
1020+
1021+
@freeze_time("2026-05-19T15:00:00Z")
1022+
def test_monthly_short_month_clamps_day(self):
1023+
"""reset_day=31 in Feb → clamps to last day of month (line 440)."""
1024+
from wl_limits import get_counter_period_key
1025+
with freeze_time("2026-02-28T15:00:00Z"):
1026+
result = get_counter_period_key({
1027+
"reset_frequency": "monthly",
1028+
"reset_day_of_month": 31,
1029+
})
1030+
# Feb has 28 days in 2026; boundary clamped to 28; we're past it
1031+
assert result == "2026-02"
1032+
1033+
@freeze_time("2026-05-19T15:00:00Z")
1034+
def test_yearly_after_boundary_returns_current_year(self):
1035+
"""Yearly freq, reset on Jan 1. May 19 >= Jan 1 → "2026"."""
1036+
from wl_limits import get_counter_period_key
1037+
result = get_counter_period_key({
1038+
"reset_frequency": "yearly",
1039+
"reset_month": 1,
1040+
"reset_day_of_year": 1,
1041+
"reset_time_utc": "00:00",
1042+
})
1043+
assert result == "2026"
1044+
1045+
@freeze_time("2026-02-15T15:00:00Z")
1046+
def test_yearly_before_boundary_returns_prior_year(self):
1047+
"""Yearly freq, reset on July 1. Feb 15 < July 1 → "2025"."""
1048+
from wl_limits import get_counter_period_key
1049+
result = get_counter_period_key({
1050+
"reset_frequency": "yearly",
1051+
"reset_month": 7,
1052+
"reset_day_of_year": 1,
1053+
"reset_time_utc": "00:00",
1054+
})
1055+
assert result == "2025"
1056+
1057+
@freeze_time("2026-05-19T15:00:00Z")
1058+
def test_yearly_invalid_month_falls_back_to_january(self):
1059+
"""reset_month=99 → falls back to 1 (line 452)."""
1060+
from wl_limits import get_counter_period_key
1061+
result = get_counter_period_key({
1062+
"reset_frequency": "yearly",
1063+
"reset_month": 99,
1064+
})
1065+
# Month=1, day=1 → already past → current year
1066+
assert result == "2026"
1067+
1068+
@freeze_time("2026-05-19T15:00:00Z")
1069+
def test_unknown_frequency_falls_back_to_daily(self):
1070+
"""Unknown freq value → daily date format (line 470)."""
1071+
from wl_limits import get_counter_period_key
1072+
result = get_counter_period_key({
1073+
"reset_frequency": "bogus_freq",
1074+
"reset_time_utc": "00:00",
1075+
})
1076+
assert result == "2026-05-19"
1077+
1078+
1079+
# ═══════════════════════════════════════════════════════════════════════════
1080+
# increment_daily_limit overflow + cleanup (lines 593-615)
1081+
# ═══════════════════════════════════════════════════════════════════════════
1082+
1083+
1084+
@pytest.mark.unit
1085+
class TestIncrementDailyLimitOverflow:
1086+
"""Cover the MAX_TRACKED_ANALYSTS overflow bucket path."""
1087+
1088+
def test_overflow_user_routes_to_shared_bucket(self):
1089+
"""When MAX_TRACKED_ANALYSTS is reached, additional users are tracked
1090+
under the shared `__overflow__` bucket (lines 603-609).
1091+
"""
1092+
from wl_limits import increment_daily_limit
1093+
# Build counters that are exactly at the cap
1094+
existing = {"u_{}".format(i): {"row_addition": 1}
1095+
for i in range(1, 1001)}
1096+
full_counters = {"2026-05-19": existing}
1097+
1098+
with patch('wl_limits.read_daily_limits',
1099+
return_value=full_counters), \
1100+
patch('wl_limits.get_counter_period_key',
1101+
return_value="2026-05-19"), \
1102+
patch('wl_limits.MAX_TRACKED_ANALYSTS', 1000), \
1103+
patch('wl_limits.write_daily_limits') as mock_write:
1104+
increment_daily_limit("new_user_over_cap", "row_addition")
1105+
1106+
# __overflow__ bucket should have been created and incremented
1107+
assert mock_write.call_count == 1
1108+
written_counters = mock_write.call_args[0][0]
1109+
assert "__overflow__" in written_counters["2026-05-19"]
1110+
assert "new_user_over_cap" not in written_counters["2026-05-19"]
1111+
1112+
def test_permanent_key_keeps_only_permanent_counter(self):
1113+
"""In `never` mode, all non-`permanent` keys are pruned (line 594)."""
1114+
from wl_limits import increment_daily_limit
1115+
with patch('wl_limits.read_daily_limits',
1116+
return_value={"2026-04-01": {"u1": {"x": 1}},
1117+
"permanent": {"u2": {"x": 1}}}), \
1118+
patch('wl_limits.get_counter_period_key',
1119+
return_value="permanent"), \
1120+
patch('wl_limits.write_daily_limits') as mock_write:
1121+
increment_daily_limit("u3", "row_addition")
1122+
1123+
written = mock_write.call_args[0][0]
1124+
# Old date-based key was pruned
1125+
assert "2026-04-01" not in written
1126+
# `permanent` key kept and u3 added there
1127+
assert "permanent" in written
1128+
assert "u3" in written["permanent"]

0 commit comments

Comments
 (0)