#!/usr/bin/env python3
"""Focused regression tests for recommend_limits metric availability."""
import importlib.util
import pathlib
import tempfile
import unittest
SCRIPT = (pathlib.Path(__file__).parents[1]
/ "Server/Unix/p4/common/bin/recommend_limits.py")
SPEC = importlib.util.spec_from_file_location("recommend_limits", SCRIPT)
rl = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(rl)
def record(index, *, tracked=True, lapse=2.0, memory=512.0,
low_lock_event=True):
rec = rl.CommandRecord(
p4_port="test.example.com:1666",
pid=index,
user="normal_user",
command="user-fstat",
args="user-fstat //depot/...",
lapse_seconds=lapse,
memory_cmd_mb=memory,
)
if tracked:
rec.database_event_count = 1
rec.low_lock_database_event_count = int(low_lock_event)
rec.total_rows_get = index * 100_000
rec.total_rows_scan = index * 1_000_000
rec.max_read_lock_held_ms = index * 1_000
return rec
def structured_line(event_type, **overrides):
values = {field: "0" for field in rl.CSV_RECORD_FIELDS[event_type]}
values.update({
"f_eventtype": f"{event_type}.58",
"f_timestamp": "1700000000",
"f_pid": "123",
"f_cmdident": "command-1",
"f_user": "test-user",
"f_client": "test-client",
"f_func": "user-fstat",
"f_host": "127.0.0.1",
"f_prog": "p4",
"f_args": "//depot/...",
})
values.update(overrides)
return ",".join(values[field] for field in rl.CSV_RECORD_FIELDS[event_type])
class MetricAvailabilityTests(unittest.TestCase):
def test_service_account_detection_handles_domain_qualified_names(self):
self.assertTrue(rl.is_service_account("AD\\svc_jukebox_p4"))
self.assertTrue(rl.is_service_account("svc_local"))
self.assertFalse(rl.is_service_account("AD\\ordinary_user"))
def test_ascii_max_lock_uses_max_values_not_totals(self):
lines = [
"2026/09/24 12:00:00 pid 123 test-user@test-client "
"127.0.0.1 [p4/2026.1] 'user-fstat //depot/...'",
"--- db.rev",
"--- total lock wait+held read/write "
"100ms+250000ms/200ms+300000ms",
"--- max lock wait+held read/write 10ms+7000ms/20ms+8000ms",
"--- lapse 10s",
]
with tempfile.TemporaryDirectory() as directory:
path = pathlib.Path(directory) / "log.txt"
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
records = rl.parse_log(str(path), quiet=True)
self.assertEqual(1, len(records))
self.assertEqual(250000, records[0].total_read_lock_held_ms)
self.assertEqual(300000, records[0].total_write_lock_held_ms)
self.assertEqual(7000, records[0].max_read_lock_held_ms)
self.assertEqual(8000, records[0].max_write_lock_held_ms)
self.assertEqual(8000, records[0].max_lock_time_ms)
def test_untracked_fast_commands_do_not_become_database_zeroes(self):
tracked = [record(i) for i in range(1, 101)]
fast_untracked = [
record(1000 + i, tracked=False, lapse=0.05, memory=64)
for i in range(500)
]
expected, _ = rl.generate_report(
tracked,
top_pct=100,
limits_only=True,
allow_incomplete_database_metrics=True,
)
actual, report = rl.generate_report(
tracked + fast_untracked,
top_pct=20,
limits_only=True,
allow_incomplete_database_metrics=True,
)
self.assertEqual(expected, actual)
self.assertIn("100 of 600 commands", report)
self.assertIn("never treated as measured zeroes", report)
def test_tracked_zero_values_are_observed_not_missing(self):
lines = [
structured_line(0),
structured_line(9, f_dbName="db.rev"),
structured_line(
2,
f_lapse=".5s",
f_peakMemCommand="64",
f_peakMemProcess="128",
),
]
with tempfile.TemporaryDirectory() as directory:
path = pathlib.Path(directory) / "all.csv"
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
records = rl.parse_csv_log(str(path), quiet=True)
self.assertEqual(1, len(records))
self.assertTrue(records[0].has_database_metrics)
self.assertEqual(1, records[0].database_event_count)
self.assertEqual(1, records[0].low_lock_database_event_count)
self.assertEqual(0, records[0].total_scan_rows)
def test_command_end_memory_remains_available_without_db_tracking(self):
tracked = [record(i, memory=256) for i in range(1, 101)]
fast_untracked = [
record(1000 + i, tracked=False, lapse=0.05, memory=9000)
for i in range(100)
]
recommendations, _ = rl.generate_report(
tracked + fast_untracked,
top_pct=100,
limits_only=True,
allow_incomplete_database_metrics=True,
)
self.assertEqual(9_000, recommendations["MaxMemory"])
def test_historic_lock_filter_is_rejected_by_default(self):
filtered = [
record(i, low_lock_event=False) for i in range(1, 101)
]
with self.assertRaisesRegex(
rl.IncompleteDatabaseMetricsError, "selectively filtered"):
rl.generate_report(filtered, top_pct=100, limits_only=True)
def test_filtered_data_requires_explicit_diagnostic_override(self):
filtered = [
record(i, low_lock_event=False) for i in range(1, 101)
]
recommendations, report = rl.generate_report(
filtered,
top_pct=100,
limits_only=True,
allow_incomplete_database_metrics=True,
)
self.assertTrue(recommendations)
self.assertIn("diagnostic only", report)
def test_lock_filter_signature_is_adx_specific(self):
local_records = [
record(i, low_lock_event=False) for i in range(1, 101)
]
for rec in local_records:
rec.p4_port = ""
coverage = rl.validate_database_metric_quality(local_records)
self.assertEqual((100, 100, 0), coverage)
def test_adx_query_preserves_availability_signals(self):
self.assertIn("database_event_count = count()", rl.ADX_QUERY_TEMPLATE)
self.assertIn(
"low_lock_database_event_count = countif(",
rl.ADX_QUERY_TEMPLATE,
)
self.assertIn("by p4_port, cmd_ident", rl.ADX_QUERY_TEMPLATE)
self.assertIn("on p4_port, cmd_ident", rl.ADX_QUERY_TEMPLATE)
def test_no_database_records_cannot_produce_database_limits(self):
with self.assertRaisesRegex(
rl.IncompleteDatabaseMetricsError,
"No DatabasePerformance records"):
rl.generate_report(
[record(1, tracked=False)],
top_pct=100,
limits_only=True,
)
def test_small_database_sample_is_rejected_by_default(self):
with self.assertRaisesRegex(
rl.IncompleteDatabaseMetricsError,
"at least 2,000"):
rl.generate_report(
[record(i) for i in range(1, 101)],
top_pct=100,
limits_only=True,
)
def test_invalid_top_percentage_is_rejected(self):
with self.assertRaisesRegex(ValueError, "top_pct"):
rl.generate_report([record(1)], top_pct=0, limits_only=True)
def test_default_max_open_files_is_10000(self):
recommendations, _ = rl.generate_report(
[record(1)],
top_pct=100,
limits_only=True,
allow_incomplete_database_metrics=True,
)
self.assertEqual("10000", recommendations["MaxOpenFiles"])
if __name__ == "__main__":
unittest.main()