import json import sys import tempfile import unittest from pathlib import Path SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" sys.path.insert(0, str(SCRIPTS)) import daily_prod_backend_log_report as report class CoverageTest(unittest.TestCase): def test_health_only_and_empty_are_not_healthy(self): window = report.resolve_report_window("2026-09-07", "Asia/Shanghai", "", "") with tempfile.TemporaryDirectory() as directory: log = Path(directory) / "backend.log" for text in ("", "2026-09-07T06:00:00Z INF HTTP Request method=GET path=/health status=200 latency=33.8\n"): log.write_text(text) data = report.build_report([log], window, 10000) self.assertTrue(data["summary"]["window_data_warning"]) self.assertIn("WARNING", report.build_feishu_text(data)) import gzip import os import shutil import subprocess from datetime import datetime, timezone import prod_http_sources as sources from prod_nginx_access_log import with_access_logging from compare_backend_latency_reports import build_comparison from notify_feishu_daily_prod_report import build_slow_log_card def event(identifier="one", when="2026-09-07T06:00:00+00:00", uri="/services/data/v60.0/query", status=200, seconds="12.345"): return json.dumps({"schema": "nexuscrm.prod.http.v1", "msec": str(datetime.fromisoformat(when).timestamp()), "method": "GET", "uri": uri, "status": str(status), "request_time": seconds, "nginx_request_id": identifier, "upstream_request_id": "request-one", "release": "release-a"}) + "\n" class NginxSourceTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.window = report.resolve_report_window("2026-09-07", "Asia/Shanghai", "", "") self.evidence = {"sources": [], "diagnostic_sources": [], "completeness": "unproven"} def write(self, name, text): path = self.root / name if name.endswith(".gz"): with gzip.open(path, "wt") as handle: handle.write(text) else: path.write_text(text) return path def build(self, files): return report.build_report(files, self.window, 10000, entries=sources.nginx_entries(files, self.window, self.evidence)) def test_rotation_release_boundary_dedup_and_exact_window(self): a = self.write("nexuscrm-prod-d3-slot-3105.access.log.1.gz", event("old") + event("start", "2026-09-06T07:00:00+00:00")) b = self.write("nexuscrm-prod-d3-slot-3105.access.log", event("old") + event("new") + event("end", "2026-09-07T07:00:00+00:00")) alias = self.root / "alias.access.log" alias.symlink_to(a) # mtime is deliberately stale: event timestamps remain authoritative. os.utime(a, (1, 1)) files = sources.discover_files([str(self.root / "*")]) self.assertEqual(len(files), 2) data = self.build(files) self.assertEqual(data["summary"]["parsed_requests_in_window"], 3) self.assertEqual(data["summary"]["all_request_p95_ms"], 12345) self.assertEqual(sum(s["duplicate_events"] for s in self.evidence["sources"]), 1) self.assertEqual(data["summary"]["first_request_at"], "2026-09-06T07:00:00+00:00") self.assertFalse(data["summary"]["window_data_warning"]) def test_legacy_counts_status_but_never_fabricates_latency(self): legacy = '127.0.0.1 - - [07/Sep/2026:06:00:00 +0000] "GET /webhooks/hubspot HTTP/1.1" 503 31 "-" "probe"\n' data = self.build([self.write("nexuscrm-prod-d3-slot-3105.access.log", legacy)]) self.assertEqual(data["summary"]["status_buckets"], {"5xx": 1}) self.assertEqual(data["summary"]["request_categories"], {"integration": 1}) self.assertIsNone(data["summary"]["all_request_p95_ms"]) self.assertIsNone(data["summary"]["slow_request_count"]) self.assertIsNone(data["route_latency_profile"][0]["p95_latency_ms"]) sources.finish_coverage(data, self.evidence, False) self.assertEqual(self.evidence["status"], "INCONCLUSIVE") self.assertIn("n/a", report.build_markdown(data)) self.assertIn("WARNING", report.build_feishu_text(data)) self.assertEqual(build_slow_log_card(data, "https://example.com")["header"]["template"], "red") with self.assertRaisesRegex(ValueError, "incomplete"): build_comparison(data, data) def test_partial_latency_retains_total_and_known_slow_samples(self): old = '127.0.0.1 - - [07/Sep/2026:06:00:01 +0000] "GET /services/data/v60.0/query HTTP/1.1" 200 1 "-" "probe"\n' data = self.build([self.write("nexuscrm-prod-d3-slot-3105.access.log", event() + old)]) row = data["route_latency_profile"][0] self.assertEqual(row["request_count"], 2) self.assertEqual(row["latency_sample_count"], 1) self.assertIsNone(row["slow_rate"]) self.assertEqual(len(data["slow_samples"]), 1) self.assertEqual(data["summary"]["observed_slow_request_count"], 1) def test_health_only_empty_malformed_and_nonfinite(self): for content in ("", event(uri="/health"), "not nginx\n", event(seconds="NaN")): with self.subTest(content=content): data = self.build([self.write("nexuscrm-prod-d3-slot-3105.access.log", content)]) self.assertTrue(data["summary"]["window_data_warning"]) def test_classic_only_cannot_satisfy_d3_coverage(self): data = self.build([self.write("nexuscrm-prod.access.log", event())]) sources.finish_coverage(data, self.evidence, False) self.assertTrue(data["summary"]["window_data_warning"]) self.assertTrue(any("Classic-only" in reason for reason in self.evidence["reasons"])) def test_new_format_rendering_and_parser_contract(self): config = with_access_logging("server { access_log /var/log/nginx/nexuscrm-prod-d3-slot-3105.access.log; }", "slot-3105", "release-a") self.assertIn("escape=json", config) self.assertIn('"request_time":"$request_time"', config) self.assertIn('"nginx_request_id":"$request_id"', config) with self.assertRaises(ValueError): with_access_logging("server {}", "slot-3105", "release-a") request, identifier, release = sources.parse_nginx(event()) self.assertEqual((request.latency_ms, identifier, release), (12345, "one", "release-a")) def test_nginx_configuration_syntax(self): nginx = shutil.which("nginx") if not nginx: if os.environ.get("REQUIRE_NGINX_TEST") == "1": self.fail("nginx is required by CI") self.skipTest("nginx is not installed") server = "server { listen 127.0.0.1:18087; access_log /var/log/nginx/nexuscrm-prod-d3-slot-3105.access.log; }" rendered = with_access_logging(server, "slot-3105", "release-a") rendered = rendered.replace("/var/log/nginx/nexuscrm-prod-d3-slot-3105.access.log", str(self.root / "access.log")) conf = self.root / "nginx.conf" conf.write_text(f"pid {self.root}/nginx.pid; error_log {self.root}/error.log; events {{}} http {{ {rendered} }}") result = subprocess.run([nginx, "-t", "-c", str(conf), "-p", str(self.root)], capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) def test_docker_diagnostics_include_stopped_instances_but_not_internal_http(self): fake = self.root / "docker" fake.write_text('''#!/usr/bin/env python3 import sys,json args=sys.argv[1:] if args[0]=='ps': print('new\\nold\\nprivate') elif args[0]=='inspect': cid=args[1];name='nexuscrm-d3-prod-slot-3105-'+cid+'-runtime-api' if cid!='private' else 'nexuscrm-d3-test-runtime-api' print(json.dumps([{'Id':cid,'Name':name,'Image':'sha256:fixture','Created':'2026-09-05T01:00:00Z','RestartCount':25,'State':{'Status':'exited' if cid=='old' else 'running'},'Config':{'Env':['SECRET=DO_NOT_PUBLISH'],'Labels':{'io.nexuscrm.serviceification-d3.run':'fixture'}},'HostConfig':{'LogConfig':{'Type':'json-file'}},'LogPath':'/fixture'}])) elif args[0]=='logs': assert '--since' in args and '--until' in args and '--timestamps' in args print('2026-09-07T06:00:00.123456789Z 2026-09-07T06:00:00Z INF HTTP Request method=GET path=/internal status=200 latency=99') print('2026-09-07T06:00:00.123456789Z 2026-09-07T06:00:00Z DBG DEBUG_SEARCH_SQL params=["%example%"] sql="Name LIKE ?"') else: sys.exit(1) ''') fake.chmod(0o755) old_path = os.environ["PATH"] os.environ["PATH"] = str(self.root) + os.pathsep + old_path try: files = [self.write("nexuscrm-prod-d3-slot-3105.access.log", event())] data = report.build_report(files, self.window, 10000, entries=sources.prod_entries(files, self.window, self.evidence)) finally: os.environ["PATH"] = old_path self.assertEqual(data["summary"]["parsed_requests_in_window"], 1) self.assertEqual(data["search_term_profile"]["total_search_queries"], 2) self.assertEqual(len(self.evidence["diagnostic_sources"]), 2) self.assertNotIn("DO_NOT_PUBLISH", json.dumps(self.evidence)) self.assertEqual(self.evidence["diagnostic_sources"][1]["state"]["Status"], "exited") self.assertEqual(self.evidence["diagnostic_sources"][0]["restart_count"], 25) def test_nginx_emits_synthetic_request_consumable_by_report(self): import http.client import socket import time nginx = shutil.which("nginx") if not nginx: if os.environ.get("REQUIRE_NGINX_TEST") == "1": self.fail("nginx is required by CI") self.skipTest("nginx is not installed") with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] access = self.root / "nexuscrm-prod-d3-slot-3105.access.log" server = f"server {{ listen 127.0.0.1:{port}; access_log /var/log/nginx/nexuscrm-prod-d3-slot-3105.access.log; location / {{ return 503; }} }}" rendered = with_access_logging(server, "slot-3105", "release-a").replace("/var/log/nginx/nexuscrm-prod-d3-slot-3105.access.log", str(access)) conf = self.root / "nginx.conf" conf.write_text(f"pid {self.root}/nginx.pid; error_log {self.root}/error.log; events {{}} http {{ {rendered} }}") with subprocess.Popen([nginx, "-c", str(conf), "-p", str(self.root), "-g", "daemon off;"], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) as process: try: for attempt in range(100): connection = http.client.HTTPConnection("127.0.0.1", port, timeout=2) try: connection.request("GET", "/services/data/v60.0/query?probe=pr-5877") response = connection.getresponse() self.assertEqual(response.status, 503) response.read() break except ConnectionRefusedError: if process.poll() is not None: self.fail(process.stderr.read().decode()) time.sleep(0.02) finally: connection.close() else: self.fail("isolated nginx did not start") finally: process.terminate() process.wait(timeout=5) self.assertNotIn("pr-5877", access.read_text()) requests = [sources.parse_nginx(line)[0] for line in access.read_text().splitlines()] matches = [r for r in requests if r.path == "/services/data/v60.0/query"] self.assertEqual(len(matches), 1) self.assertEqual(matches[0].status, 503) self.assertIsNotNone(matches[0].latency_ms) self.assertLess(abs((datetime.now(timezone.utc) - matches[0].timestamp).total_seconds()), 10) def test_non_origin_requests_preserve_nginx_400(self): for request_line in ("PRI * HTTP/2.0", "-", r"\x16\x03garbage"): line = f'127.0.0.1 - - [07/Sep/2026:06:00:00 +0000] "{request_line}" 400 31 "-" "probe"' request, _, _ = sources.parse_nginx(line) self.assertEqual(request.status, 400) self.assertEqual(request.path, "/__invalid_request__") self.assertEqual(report.request_category(request.path), "ignored") def test_cli_outputs_reviewable_inconclusive_artifacts(self): marker = self.root / "d3/current.json" marker.parent.mkdir() marker.write_text(json.dumps({"schema": "nexuscrm.serviceification-d3-prod-current/v2"})) self.write("nexuscrm-prod-d3-slot-3105.access.log", event()) result = subprocess.run([ sys.executable, str(SCRIPTS / "daily_prod_backend_log_report.py"), "--source", "prod-nginx", "--prod-root", str(self.root), "--nginx-log-glob", str(self.root / "*.access.log"), "--skip-container-diagnostics", "--report-date", "2026-09-07", "--output-json", str(self.root / "report.json"), "--output-markdown", str(self.root / "report.md"), "--output-feishu-text", str(self.root / "report.txt"), ], capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) data = json.loads((self.root / "report.json").read_text()) self.assertEqual(data["source_coverage"]["status"], "INCONCLUSIVE") self.assertIn("WARNING", (self.root / "report.txt").read_text()) self.assertEqual(data["summary"]["parsed_requests_in_window"], 1) def test_queries_are_removed_from_nginx_records_and_artifacts(self): secret = "sensitive-query-value" for text in (event(uri="/services/data/v60.0/query?access_token=" + secret), f'127.0.0.1 - - [07/Sep/2026:06:00:00 +0000] "GET /services/data/v60.0/query?q={secret} HTTP/1.1" 200 1 "-" "probe"\n'): data = self.build([self.write("nexuscrm-prod-d3-slot-3105.access.log", text)]) self.assertNotIn(secret, json.dumps(data)) config = with_access_logging("server { access_log /var/log/nginx/nexuscrm-prod-d3-slot-3105.access.log; }", "slot-3105", "release-a") self.assertNotIn("$request_uri", config) def test_another_stream_cannot_mask_a_partial_d3_window(self): for other in ("nexuscrm-prod.access.log", "nexuscrm-prod-d3-slot-3104.access.log"): with self.subTest(other=other): self.evidence = {"sources": [], "diagnostic_sources": [], "completeness": "unproven"} a = self.write(other, event("before", "2026-09-06T06:59:59+00:00") + event("during") + event("after", "2026-09-07T07:00:01+00:00")) b = self.write("nexuscrm-prod-d3-slot-3105.access.log", event("only-d3")) data = self.build([a, b]) self.evidence["diagnostic_sources"] = [{"role": role, "selection": "retained_instance"} for role in ("runtime-api", "sfcrm", "integration")] sources.finish_coverage(data, self.evidence, True) self.assertEqual(self.evidence["status"], "INCONCLUSIVE") self.assertTrue(data["summary"]["window_data_warning"]) with self.assertRaises(ValueError): build_comparison(data, data) def test_prod_render_replacement_and_route_identity_across_slot_reuse(self): from types import SimpleNamespace from scripts import run_serviceification_d3_prod as prod current = "server {\n server_name nexuscrm-new-prod.pingcap.com;\n location / { proxy_pass http://127.0.0.1:3103; }\n}\n" for index, port in enumerate((3105, 3106, 3105, 3105)): slot = f"slot-{port}" context = SimpleNamespace( envv={"ENV_NAME": "multi-prod-d3-" + slot, "PUBLIC_BASE_URL": "https://nexuscrm-new-prod.pingcap.com"}, env_root=Path(f"release-{index}"), run_id=f"run-{index}", images={}, ports=SimpleNamespace(sfcrm_http=port, integration_http=3423 if port == 3105 else 3433)) rendered = prod.render_d3_nginx_server(context, self.root) # This must pass before activation: a distant marker misclassifies D3 as Classic. self.assertTrue(prod.parse_prod_route(rendered).is_d3) current = prod.replace_prod_nginx_server_block( prod.impl.remove_d3_nginx_markers(current), prod.PROD_SERVER_NAME, rendered) self.assertEqual(current.count("log_format nexuscrm_prod_http_v1_"), 1) self.assertTrue(prod.parse_prod_route(current).is_d3) self.assertEqual(prod.parse_prod_route(current).backend_port, port) if shutil.which("nginx"): import re import socket with socket.socket() as listener: listener.bind(("127.0.0.1", 0)) test_port = listener.getsockname()[1] local = current.replace("listen 80;", f"listen 127.0.0.1:{test_port};") local = re.sub(r"/var/log/nginx/([^;\s]+)", lambda m: str(self.root / m[1]), local) conf = self.root / "nginx-reuse.conf" conf.write_text(f"pid {self.root}/nginx.pid; error_log {self.root}/error.log; events {{}} http {{ {local} }}") checked = subprocess.run([shutil.which("nginx"), "-t", "-c", str(conf), "-p", str(self.root)], capture_output=True, text=True) self.assertEqual(checked.returncode, 0, checked.stderr) def test_rotation_bounds_are_scoped_to_release_and_preserve_healthy_control(self): for release in ("release-a", "release-b"): self.evidence = {"sources": [], "diagnostic_sources": [], "completeness": "unproven"} before = event("before", "2026-09-06T06:59:59+00:00").replace("release-a", release) after = event("after", "2026-09-07T07:00:01+00:00").replace("release-a", release) a = self.write("nexuscrm-prod-d3-slot-3105.access.log.1", before + event("during")) b = self.write("nexuscrm-prod-d3-slot-3105.access.log", after) data = self.build([a, b]) data["search_term_profile"]["by_input_type"] = [{"type": "ascii-text", "samples": ["private-search"], "count": 1}] self.evidence["diagnostic_sources"] = [{"role": role, "selection": "retained_instance"} for role in ("runtime-api", "sfcrm", "integration")] sources.finish_coverage(data, self.evidence, True) self.assertEqual(self.evidence["status"], "OBSERVED" if release == "release-a" else "INCONCLUSIVE") self.assertNotIn("private-search", json.dumps(data)) def test_multiple_bracketed_streams_require_cutover_authority(self): for second_slot in ("3105", "3106"): with self.subTest(second_slot=second_slot): self.evidence = {"sources": [], "diagnostic_sources": [], "completeness": "unproven"} first = event("a-before", "2026-09-06T06:59:59+00:00") + event("a-during") + event("a-after", "2026-09-07T07:00:01+00:00") second = (event("b-before", "2026-09-06T06:59:59+00:00") + event("b-during") + event("b-after", "2026-09-07T07:00:01+00:00")).replace("release-a", "release-b") a = self.write("nexuscrm-prod-d3-slot-3105.access.log.1", first) b = self.write(f"nexuscrm-prod-d3-slot-{second_slot}.access.log", second) data = self.build([a, b]) self.evidence["diagnostic_sources"] = [{"role": role, "selection": "retained_instance"} for role in ("runtime-api", "sfcrm", "integration")] sources.finish_coverage(data, self.evidence, True) self.assertEqual(data["summary"]["parsed_requests_in_window"], 2) self.assertTrue(all(s["brackets_window"] for s in self.evidence["d3_stream_bounds"])) self.assertEqual(self.evidence["status"], "INCONCLUSIVE") self.assertTrue(any("multiple D3" in reason for reason in self.evidence["reasons"])) with self.assertRaises(ValueError): build_comparison(data, data)