#!/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()