284 lines
11 KiB
Python
284 lines
11 KiB
Python
import importlib.util
|
|
import io
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
|
|
SCRIPT_PATH = Path(__file__).resolve().parents[1] / "public-traffic-rollup-from-nginx.py"
|
|
SPEC = importlib.util.spec_from_file_location("public_traffic_rollup", SCRIPT_PATH)
|
|
ROLLUP = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(ROLLUP)
|
|
|
|
|
|
def nginx_line(timestamp, path="/asset.webp", status=200, body_bytes=100):
|
|
rendered = timestamp.strftime("%d/%b/%Y:%H:%M:%S %z")
|
|
return (
|
|
f'127.0.0.1 - - [{rendered}] "GET {path} HTTP/1.1" '
|
|
f'{status} {body_bytes} "-" "test" request_length=42 bytes_sent={body_bytes}\n'
|
|
)
|
|
|
|
|
|
class PublicTrafficRollupTest(unittest.TestCase):
|
|
def make_args(self, log_dir, state_dir, overlap=15):
|
|
return SimpleNamespace(
|
|
log_dir=str(log_dir),
|
|
domains="sc.jiansun.vip",
|
|
granularity="5m",
|
|
source_server="test-server",
|
|
state_dir=str(state_dir),
|
|
recent_overlap_minutes=overlap,
|
|
)
|
|
|
|
def test_reverse_reader_preserves_complete_lines(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "sample.log"
|
|
expected = [f"line-{index}-" + ("x" * 47) for index in range(250)]
|
|
path.write_text("\n".join(expected) + "\n", encoding="utf-8")
|
|
|
|
actual = [line.rstrip("\n") for line in ROLLUP.iter_reverse_lines(path, block_size=113)]
|
|
|
|
self.assertEqual(actual, list(reversed(expected)))
|
|
|
|
def test_incremental_scan_reads_only_overlap_and_keeps_complete_buckets(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
log_path = root / "sc.jiansun.vip.access.log"
|
|
timestamps = [
|
|
datetime(2026, 8, 1, 9, 0, tzinfo=ROLLUP.CHINA_TZ),
|
|
datetime(2026, 8, 1, 10, 10, tzinfo=ROLLUP.CHINA_TZ),
|
|
datetime(2026, 8, 1, 10, 14, tzinfo=ROLLUP.CHINA_TZ),
|
|
datetime(2026, 8, 1, 10, 17, tzinfo=ROLLUP.CHINA_TZ),
|
|
datetime(2026, 8, 1, 10, 19, tzinfo=ROLLUP.CHINA_TZ),
|
|
]
|
|
log_path.write_text("".join(nginx_line(value) for value in timestamps), encoding="utf-8")
|
|
os.utime(log_path, (timestamps[-1].timestamp(), timestamps[-1].timestamp()))
|
|
args = self.make_args(root, root / "state", overlap=5)
|
|
business_date = timestamps[-1].date()
|
|
state = {
|
|
"lastLogTimestamps": {
|
|
"sc.jiansun.vip": ROLLUP.iso_utc(timestamps[-1]),
|
|
},
|
|
}
|
|
|
|
recent_starts = ROLLUP.incremental_starts(
|
|
args,
|
|
state,
|
|
business_date,
|
|
business_date,
|
|
("sc.jiansun.vip",),
|
|
)
|
|
rows, scanned, matched, _, newest, _, complete = ROLLUP.aggregate(
|
|
args,
|
|
business_date,
|
|
business_date,
|
|
recent_since_by_domain=recent_starts,
|
|
)
|
|
|
|
self.assertEqual(
|
|
recent_starts["sc.jiansun.vip"],
|
|
datetime(2026, 8, 1, 10, 10, tzinfo=ROLLUP.CHINA_TZ),
|
|
)
|
|
self.assertEqual(matched, 4)
|
|
self.assertEqual(sum(row["totalRequests"] for row in rows), 4)
|
|
self.assertLess(scanned, len(timestamps) + 1)
|
|
self.assertEqual(newest["sc.jiansun.vip"], timestamps[-1])
|
|
self.assertTrue(complete["sc.jiansun.vip"])
|
|
|
|
def test_incremental_scan_tolerates_a_physically_late_older_log_line(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
log_path = root / "sc.jiansun.vip.access.log"
|
|
newer = [
|
|
datetime(2026, 8, 1, 10, 16, tzinfo=ROLLUP.CHINA_TZ),
|
|
datetime(2026, 8, 1, 10, 17, tzinfo=ROLLUP.CHINA_TZ),
|
|
]
|
|
physically_late_but_old = datetime(2026, 8, 1, 9, 0, tzinfo=ROLLUP.CHINA_TZ)
|
|
log_path.write_text(
|
|
"".join(nginx_line(value) for value in [*newer, physically_late_but_old]),
|
|
encoding="utf-8",
|
|
)
|
|
os.utime(log_path, (newer[-1].timestamp(), newer[-1].timestamp()))
|
|
args = self.make_args(root, root / "state", overlap=5)
|
|
business_date = newer[-1].date()
|
|
|
|
rows, _, matched, _, newest, _, complete = ROLLUP.aggregate(
|
|
args,
|
|
business_date,
|
|
business_date,
|
|
recent_since_by_domain={
|
|
"sc.jiansun.vip": datetime(
|
|
2026, 8, 1, 10, 10, tzinfo=ROLLUP.CHINA_TZ,
|
|
),
|
|
},
|
|
)
|
|
|
|
self.assertEqual(matched, 2)
|
|
self.assertEqual(sum(row["totalRequests"] for row in rows), 2)
|
|
self.assertEqual(newest["sc.jiansun.vip"], newer[-1])
|
|
self.assertTrue(complete["sc.jiansun.vip"])
|
|
|
|
def test_state_round_trip_is_atomic_and_reusable(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
path = root / "state" / "checkpoint.json"
|
|
expected = {"lastLogTimestamp": "2026-08-01T02:19:00Z", "completed": True}
|
|
|
|
ROLLUP.save_state(path, expected)
|
|
|
|
self.assertEqual(ROLLUP.load_state(path), expected)
|
|
self.assertEqual(list(path.parent.glob(f".{path.name}.*")), [])
|
|
|
|
def test_checkpoint_key_is_stable_across_blue_green_container_names(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
business_date = datetime(2026, 8, 1, tzinfo=ROLLUP.CHINA_TZ).date()
|
|
args_blue = self.make_args(root, root / "state")
|
|
args_green = self.make_args(root, root / "state")
|
|
args_blue.source_server = "blue-container-id"
|
|
args_green.source_server = "green-container-id"
|
|
|
|
blue_path = ROLLUP.state_path(
|
|
args_blue,
|
|
business_date,
|
|
business_date,
|
|
("sc.jiansun.vip",),
|
|
)
|
|
green_path = ROLLUP.state_path(
|
|
args_green,
|
|
business_date,
|
|
business_date,
|
|
("sc.jiansun.vip",),
|
|
)
|
|
|
|
self.assertEqual(blue_path, green_path)
|
|
|
|
def test_completed_day_is_not_rescanned_every_five_minutes(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
state_dir = root / "state"
|
|
business_date = datetime.now(ROLLUP.CHINA_TZ).date() - timedelta(days=1)
|
|
timestamp = datetime.combine(
|
|
business_date,
|
|
datetime.min.time(),
|
|
tzinfo=ROLLUP.CHINA_TZ,
|
|
).replace(hour=12)
|
|
after_day = timestamp + timedelta(hours=13)
|
|
log_path = root / "sc.jiansun.vip.access.log"
|
|
log_path.write_text(
|
|
nginx_line(timestamp) + nginx_line(after_day),
|
|
encoding="utf-8",
|
|
)
|
|
argv = [
|
|
str(SCRIPT_PATH),
|
|
"--granularity", "day",
|
|
"--start-date", business_date.isoformat(),
|
|
"--end-date", business_date.isoformat(),
|
|
"--log-dir", str(root),
|
|
"--state-dir", str(state_dir),
|
|
"--domains", "sc.jiansun.vip",
|
|
"--token-env", "TEST_ROLLUP_TOKEN",
|
|
"--source-server", "test-server",
|
|
]
|
|
|
|
with patch.dict(os.environ, {"TEST_ROLLUP_TOKEN": "test-token"}), \
|
|
patch.object(ROLLUP.sys, "argv", argv), \
|
|
patch.object(ROLLUP, "post_rows", return_value=1) as post_rows:
|
|
first_output = io.StringIO()
|
|
with redirect_stdout(first_output):
|
|
ROLLUP.main()
|
|
second_output = io.StringIO()
|
|
with redirect_stdout(second_output):
|
|
ROLLUP.main()
|
|
|
|
first_result = ROLLUP.json.loads(first_output.getvalue())
|
|
second_result = ROLLUP.json.loads(second_output.getvalue())
|
|
self.assertEqual(first_result["scanMode"], "full")
|
|
self.assertEqual(first_result["matchedLines"], 1)
|
|
self.assertTrue(second_result["skipped"])
|
|
self.assertEqual(second_result["scanMode"], "cached")
|
|
post_rows.assert_called_once()
|
|
|
|
def test_domain_checkpoint_does_not_follow_a_faster_domain(self):
|
|
args = SimpleNamespace(granularity="5m", recent_overlap_minutes=15)
|
|
business_date = datetime(2026, 8, 1, tzinfo=ROLLUP.CHINA_TZ).date()
|
|
state = {
|
|
"lastLogTimestamps": {
|
|
"sc.jiansun.vip": "2026-08-01T02:20:00Z",
|
|
"s3.jiansun.vip": "2026-08-01T01:00:00Z",
|
|
},
|
|
}
|
|
|
|
starts = ROLLUP.incremental_starts(
|
|
args,
|
|
state,
|
|
business_date,
|
|
business_date,
|
|
("sc.jiansun.vip", "s3.jiansun.vip", "cp.jiansun.vip"),
|
|
)
|
|
|
|
self.assertEqual(
|
|
starts["sc.jiansun.vip"],
|
|
datetime(2026, 8, 1, 10, 5, tzinfo=ROLLUP.CHINA_TZ),
|
|
)
|
|
self.assertEqual(
|
|
starts["s3.jiansun.vip"],
|
|
datetime(2026, 8, 1, 8, 45, tzinfo=ROLLUP.CHINA_TZ),
|
|
)
|
|
self.assertIsNone(starts["cp.jiansun.vip"])
|
|
|
|
def test_day_cache_has_a_bounded_revalidation_window(self):
|
|
now = datetime(2026, 8, 2, 12, 0, tzinfo=ROLLUP.CHINA_TZ)
|
|
args = SimpleNamespace(granularity="day", day_cache_hours=1.0)
|
|
end_date = now.date() - timedelta(days=1)
|
|
|
|
self.assertTrue(ROLLUP.is_fresh_day_cache(
|
|
args,
|
|
{"completedAt": ROLLUP.iso_utc(now - timedelta(minutes=30))},
|
|
end_date,
|
|
now,
|
|
))
|
|
self.assertFalse(ROLLUP.is_fresh_day_cache(
|
|
args,
|
|
{"completedAt": ROLLUP.iso_utc(now - timedelta(hours=2))},
|
|
end_date,
|
|
now,
|
|
))
|
|
|
|
def test_changed_source_file_drops_partial_domain_rows(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
timestamp = datetime(2026, 8, 1, 10, 0, tzinfo=ROLLUP.CHINA_TZ)
|
|
log_path = root / "sc.jiansun.vip.access.log"
|
|
log_path.write_text(nginx_line(timestamp), encoding="utf-8")
|
|
args = self.make_args(root, root / "state")
|
|
initial = {"sc.jiansun.vip": ((str(log_path), 100, 1),)}
|
|
changed = {"sc.jiansun.vip": ((str(log_path), 200, 2),)}
|
|
|
|
with patch.object(
|
|
ROLLUP,
|
|
"snapshot_domain_paths",
|
|
side_effect=[
|
|
(initial, {"sc.jiansun.vip": True}),
|
|
(changed, {"sc.jiansun.vip": True}),
|
|
],
|
|
):
|
|
rows, _, matched, _, _, _, complete = ROLLUP.aggregate(
|
|
args,
|
|
timestamp.date(),
|
|
timestamp.date(),
|
|
)
|
|
|
|
self.assertEqual(rows, [])
|
|
self.assertEqual(matched, 0)
|
|
self.assertFalse(complete["sc.jiansun.vip"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|